distributor/docs/internal/app.md

Application Orchestration

Purpose

internal/app owns the top-level application use cases. It coordinates configuration loading, secret resolution, backend construction, source bundle discovery, destination selection, publish planning, publish execution, notification handoff, run reporting, and in-memory run coordination.

The package is the boundary between callers and lower-level domain packages. It does not own manifest validation rules, destination state comparison, storage path rules, output planning, transform rendering, or backend-specific behavior.

Use Cases

Run is the CLI-facing all-pipeline entrypoint. It accepts a context, optional config path, dry-run flag, force flag, stdout writer, output format, and optional notifier. It runs every configured pipeline, builds a RunReport, and projects the report to text or JSON when stdout is supplied.

RunPipeline is the app-layer single-pipeline entrypoint. It accepts a context, config path, pipeline ID, dry-run flag, force flag, and optional notifier. It loads the same config as Run, narrows execution to exactly one configured pipeline, and returns a RunReport without writing command output.

RunPipelineWithLocalSource is the app-layer single-pipeline entrypoint for an already prepared local source bundle root. It accepts the same pipeline selection and execution options as RunPipeline plus a local source root path. It loads config, selects one configured pipeline, opens the supplied source root as a local backend, validates exactly that root bundle, and then uses the same destination fan-out path as normal runs.

Validate and Inspect accept either a local path or one configured pipeline source. Configured-source mode uses the same runtime config and secret setup as run workflows, shares source backend construction, and never opens destination backends.

Serve is the CLI-facing HTTP upload server entrypoint. It uses the app runtime setup, resolves upload bearer tokens for configured http_upload sources, creates an UploadCoordinator, binds server.http.bind, and serves the upload API until its context is cancelled.

Run Reports

RunReport is the structured result model for run workflows. It includes dry-run state, pipeline summaries, action records, output metadata, summary counters, warnings, and destination-scoped output errors.

Text and JSON run output are projections of RunReport. JSON tags on report records match the CLI JSON output contract. Text output preserves the CLI summary shape while keeping output rendering outside the core planning and execution loop.

Destination-scoped failures produce a report plus an aggregated error. Fatal setup failures, such as config loading, source open, or source discovery failures, return before a complete run report is available.

Run Flow

The app runner:

1. builds runtime setup by resolving the config path, loading config, loading configured secret files, and projecting secret-conflict warnings;
2. builds the app-level backend factory from the config-owned environment resolver;
3. builds the app-level transform registry;
4. opens each selected pipeline source backend;
5. discovers validated source bundles from the source root;
6. selects source bundles for each destination according to path mapping;
7. opens destination backends independently;
8. builds publish plans for selected bundle and destination combinations;
9. records warnings, action records, output metadata, and summary counters;
10. executes publish or replacement plans unless dry-run is enabled;
11. invokes the notifier after successful publish or replacement actions;
12. returns the structured report and any aggregated destination failures.

RunPipeline follows the same flow after selecting a single configured pipeline. It uses the same backend factory, secret loading, transform registry, warning generation, destination planning, publish execution, notification behavior, and failure aggregation as Run.

RunPipelineWithLocalSource follows the same flow after pipeline selection except for source opening and source discovery. It opens the supplied local source root directly, validates the root bundle before opening any destinations, and passes the resulting local source backend and bundle into the same destination planning and execution loop. Destination code receives the normal storage backend and bundle values and does not depend on how the source root was prepared.

Upload Coordination

UploadCoordinator owns in-memory coordination for asynchronous upload processing. It admits uploads for configured http_upload pipelines, reserves queue capacity before request-body staging, stages and validates archives through internal/ingest, tracks accepted status records, and executes the selected pipeline through RunPipelineWithLocalSource.

Upload run IDs use:

<pipeline id>.<UTC timestamp>.<random suffix>

The timestamp uses YYYYMMDDThhmmssZ UTC format and the suffix is filesystem safe.

The coordinator records these statuses:

• accepted
• queued
• running
• succeeded
• failed
• expired

Admission is bounded by server.http.queue_size. Full queues are rejected before the upload body is read. Successfully reserved uploads are staged and validated before an accepted run record is created. Execution is bounded by server.http.max_concurrency, and only one upload for a given pipeline may run at a time. Later accepted uploads for the same pipeline remain queued until the active run finishes.

Completed records retain the final run report or error text until server.http.retention elapses. Expiration removes completed status records and their committed staged bundle directories. The coordinator is memory-only and does not persist queue state, status records, or run reports.

HTTP Upload Server

The HTTP upload server is app-layer transport wiring around UploadCoordinator. It owns request authentication, route dispatch, HTTP status mapping, and JSON response projection. Bundle staging and publication remain in the coordinator and staged-source run path.

Server startup resolves every configured http_upload source token_env through the config-owned environment resolver after secrets.directory has been loaded. Startup fails when a token is missing, empty, or duplicates another upload pipeline token. Error messages identify environment variable names and pipeline ids, but not token values.

Routes:

• GET /healthz: returns 200 after config, secrets, tokens, coordinator, and route setup succeed.
• POST /upload: stages and validates an authenticated tar or tar.gz archive, then returns an accepted run id.
• GET /runs/<run_id>: returns the current in-memory upload status record or 404.

The upload token maps to exactly one configured pipeline. Producers do not submit pipeline ids, and submitted pipeline or pipeline_id query values are rejected. Full queues are rejected before the request body is read. Malformed archives and invalid staged bundles are rejected before a run id is issued. Oversized uploads, unsupported content types, invalid bearer tokens, full queues, and unknown status records are mapped to stable HTTP status codes without returning secret token values.

Coordination

PipelineRunCoordinator wraps RunPipeline with in-memory admission control. It allows different pipeline IDs to run concurrently and rejects a second active run for the same pipeline ID.

Coordinator records contain a run ID, pipeline ID, status, timestamps, completed report, and error text when applicable. Active state is memory-only and is cleared after success, failure, unknown pipeline ID, or context cancellation.

The admission context is checked before a run is accepted. Once accepted, the run uses the coordinator lifetime context, so caller cancellation can stop waiting for admission without owning the actual run lifetime.

The coordinator does not queue duplicate runs, persist run records, or define transport endpoints.

Errors

Run returns immediately for config loading errors, context cancellation before work starts, source open errors, and source discovery errors.

RunPipeline returns PipelineNotFoundError when the requested pipeline ID is not configured. Callers can detect that condition with IsPipelineNotFound.

RunPipelineWithLocalSource also returns PipelineNotFoundError for an unknown pipeline ID. It returns before destination opening when the supplied local source root is missing, cannot be opened, or does not validate as one complete source bundle.

Per-destination backend, planning, execution, and notification errors are aggregated into one run error after remaining destinations have been attempted. Destination diagnostics include pipeline ID, destination ID, backend, and bundle path.

PipelineRunCoordinator returns DuplicatePipelineRunError when the same pipeline already has an active run. Callers can detect that condition with IsDuplicatePipelineRun.

Stdout write errors are returned immediately because the caller's requested output stream can no longer be trusted.

Package Layout

Run helpers are grouped by responsibility:

• runtime.go: runtime config path resolution, config loading, secret loading, environment resolver handoff, and secret-conflict warning projection.
• run.go: Run, RunPipeline, and shared run orchestration.
• run_destination.go: destination-scoped planning, execution, action recording, and failure bookkeeping.
• run_output.go: RunReport, action/output records, and text/JSON report projection.
• output_projection.go: shared bundle and manifest-file result projection for command JSON output.
• run_summary.go: summary counters.
• run_failures.go: destination failure aggregation and partial-result detection.
• run_selection.go: destination bundle selection, path mapping decisions, and fixed-path warnings.
• run_warnings.go: secret and SSH warning records.
• run_notify.go: notification event projection and action filtering.
• run_coordinator.go: in-memory run admission, run IDs, status records, and duplicate-run errors.
• upload_coordinator.go: in-memory upload admission, queue reservation, staging handoff, status tracking, queueing, and staged-source execution.
• upload_http.go: HTTP upload authentication, routes, JSON response projection, and HTTP error mapping.
• serve.go: HTTP server startup.
• backends.go: app-level backend factory wiring.
• transforms.go: app-level transform registry wiring.
• source_select.go: configured-source selection shared by validate and inspect.

Backend And Transform Wiring

The app-level backend factory registers local, SSH, and S3 backends for runtime execution. Source and destination backend config is converted through a shared app-local open spec before adapter construction.

Credential references are resolved through the config environment resolver. Production app code must not read backend credential environment variables directly.

The app-level transform registry registers Markdown-to-HTML through internal/transform/markdown. Lower-level publish code receives a resolver and does not import concrete transform implementations.

Dry-Run Behavior

Dry-run loads config, opens backends, discovers bundles, inspects destinations, resolves transforms, and builds publish plans. It does not write destination outputs, write .distributor.json, delete managed outputs, perform forced prefix deletion, or invoke notifications.

Tests

Before changing app orchestration, inspect tests under:

• internal/app
• internal/cli
• internal/publish

Use focused app tests for report structure, single-pipeline execution, coordinator admission, warning generation, notification behavior, and partial-result aggregation.

Invariants

• One source fans out to each destination independently.
• Destination failures do not prevent later destinations from being planned.
• Destination-scoped failures still produce a structured report plus an aggregated error.
• Dry-run must not mutate destination storage or invoke notifications.
• RunPipeline must use the same run path as Run after pipeline selection.
• Duplicate in-flight runs are rejected only for the same pipeline ID.
• Different pipeline IDs may run concurrently.
• Concrete backend and transform registration stays at the app layer.
• The default notifier is notify.Noop.