Add HTTP upload server and serve command

2026-06-03 15:22:52 +00:00
parent f0c10210eb
commit 6beef58dbf
12 changed files with 942 additions and 10 deletions
--- a/docs/cli.md
+++ b/docs/cli.md
@@ -14,6 +14,7 @@ This discovers the example source bundle and publishes source files to `workspac
 distributor [--help]
 distributor version [--format text|json]
 distributor run [--config <path>] [--dry-run] [--force] [--format text|json]
+distributor serve [--config <path>]
 distributor validate [--format text|json] <path>
 distributor validate --config <path> --pipeline <id> [--bundle <path>] [--format text|json]
 distributor inspect [--format text|json] <path>
@@ -23,11 +24,12 @@ distributor manifest create <bundle-path> --id <bundle-id> [options]

 - `version`: prints the application name and version. Development builds print `distributor dev`.
 - `run`: loads a YAML config, discovers source bundles, plans each configured destination, writes selected outputs unless `--dry-run` is set, and prints a final status summary.
+- `serve`: loads a YAML config, resolves HTTP upload bearer tokens, and runs the HTTP upload API.
 - `validate`: validates a local source bundle directory, a local source bundle tree, or one configured pipeline source.
 - `inspect`: validates source bundles and prints normalized bundle metadata for a local path or one configured pipeline source.
 - `manifest create`: creates `manifest.json` for a local source bundle directory.

-`validate` and `inspect` have two mutually exclusive modes: a local path shortcut, or configured source mode with `--config <path> --pipeline <id>`. Configured source mode opens only the selected pipeline source and supports configured `local`, `ssh`, and `s3` sources. It does not open destinations. `run` executes configured sources and destinations.
+`validate` and `inspect` have two mutually exclusive modes: a local path shortcut, or configured source mode with `--config <path> --pipeline <id>`. Configured source mode opens only the selected pipeline source and supports configured `local`, `ssh`, and `s3` sources. It does not open destinations. `run` executes configured `local`, `ssh`, and `s3` sources and destinations. `serve` executes configured `http_upload` sources through the upload API and normal destination fan-out.

 ## Flag reference

@@ -49,6 +51,10 @@ Output-producing subcommands:
 - `--dry-run`: load config, discover bundles, inspect destination state, print planned actions and final status, and do not write output files, destination state, or SSH `known_hosts` entries.
 - `--force`: allow explicit destructive replacement for supported conflict cases in this run only.

+`serve` flags:
+
+- `--config <path>`: config file to load. If omitted, `serve` uses `/usr/local/etc/distributor/config.yml`.
+
 `validate` and `inspect` configured source flags:

 - `--config <path>`: config file to load for source validation or inspection. Required in configured source mode.
@@ -127,6 +133,39 @@ Publish the local HTML example:
 go run ./cmd/distributor run --config examples/local-html.yml
 ```

+Start the HTTP upload API:
+
+```sh
+go run ./cmd/distributor serve --config <config-path>
+```
+
+Upload an archive to the configured `http_upload` pipeline associated with a bearer token:
+
+```sh
+curl -X POST http://127.0.0.1:8080/upload \
+  -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \
+  -H "Content-Type: application/gzip" \
+  --data-binary @bundle.tar.gz
+```
+
+The upload response is accepted asynchronously:
+
+```json
+{"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"}
+```
+
+Check upload status:
+
+```sh
+curl http://127.0.0.1:8080/runs/<run-id>
+```
+
+Check server readiness:
+
+```sh
+curl http://127.0.0.1:8080/healthz
+```
+
 Preview local fan-out publication:

 ```sh
--- a/docs/config.md
+++ b/docs/config.md
@@ -2,15 +2,18 @@

 ## Config File Location

-`distributor run --config <path>` loads the YAML config at the provided path.
+`distributor run --config <path>` and `distributor serve --config <path>` load
+the YAML config at the provided path.

-If `--config` is omitted, `run` uses:
+If `--config` is omitted, both commands use:

 ```text
 /usr/local/etc/distributor/config.yml
 ```

-Config parsing rejects unknown YAML fields. The executable `run` backends are `local`, `ssh`, and `s3`. The schema also accepts `http_upload` as a source-only ingestion backend configuration.
+Config parsing rejects unknown YAML fields. The executable `run` backends are
+`local`, `ssh`, and `s3`. The `serve` command executes `http_upload` sources
+through the HTTP upload API and normal destination fan-out.

 ## Minimal Local Config

@@ -62,7 +65,9 @@ pipelines:

 ## HTTP Upload Source Configuration

-HTTP upload sources are configured as pipeline sources only. They are not valid destination backends.
+HTTP upload sources are configured as pipeline sources only. They are not valid
+destination backends. `distributor serve` maps each configured upload token to
+exactly one `http_upload` pipeline.

 ```yaml
 server:
@@ -92,6 +97,50 @@ pipelines:

 `source.max_upload_size` is optional. When omitted, it defaults to `server.http.max_upload_size`.

+The server resolves each `token_env` through the real process environment and
+the configured `secrets.directory` resolver. Startup fails if any configured
+upload token is missing, empty, or resolves to the same value as another upload
+pipeline. Token values are not read from YAML and are not printed in API
+responses.
+
+## HTTP Upload API
+
+`distributor serve` binds to `server.http.bind`, which defaults to
+`127.0.0.1:8080`.
+
+Routes:
+
+- `GET /healthz`: returns readiness status after config and upload tokens load.
+- `POST /upload`: accepts one tar or tar.gz source bundle archive.
+- `GET /runs/<run_id>`: returns an in-memory upload status record, or `404` if the run id is unknown or expired.
+
+`POST /upload` authenticates with:
+
+```text
+Authorization: Bearer <token>
+```
+
+The token selects the configured `http_upload` pipeline. Producers do not send a
+pipeline id. Requests with a submitted `pipeline` or `pipeline_id` query value
+are rejected.
+
+Accepted upload content types:
+
+- `application/x-tar`
+- `application/gzip`
+- `application/x-gzip`
+
+Accepted uploads return:
+
+```json
+{"run_id":"<id>","status":"accepted"}
+```
+
+The run id can be queried through `GET /runs/<run_id>` while the status record
+is retained in memory. Completed records expire after `server.http.retention`;
+expiration also removes committed staged bundle directories for completed
+uploads.
+
 ## HTML Publication

 To publish generated sidecar HTML from Markdown files:
--- a/docs/internal/app.md
+++ b/docs/internal/app.md
@@ -34,6 +34,11 @@ same destination fan-out path as normal runs.
 source. They share source backend construction with run workflows and never open
 destination backends.

+`Serve` is the CLI-facing HTTP upload server entrypoint. It loads config,
+loads the configured secrets directory, 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
@@ -116,6 +121,32 @@ Completed records retain the final run report or error text until
 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`: accepts authenticated tar and tar.gz archives and 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. 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.
@@ -171,6 +202,8 @@ Run helpers are grouped by responsibility:
 - `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, queueing, status tracking, staging handoff, and staged-source execution.
+- `upload_http.go`: HTTP upload authentication, routes, JSON response projection, and HTTP error mapping.
+- `serve.go`: config/secrets loading and 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`.
--- a/docs/internal/config.md
+++ b/docs/internal/config.md
@@ -14,7 +14,8 @@ Input is a YAML file containing optional `server`, optional `secrets`, and requi

 Known-field checking rejects misspelled or unknown YAML keys before defaults and validation run.

-`LoadFile` does not read secret files. `Run` loads the configured secrets directory after config validation and before backend construction.
+`LoadFile` does not read secret files. App entrypoints load the configured
+secrets directory after config validation and before credential-consuming work.

 ## Defaults

@@ -58,13 +59,13 @@ Destination links are optional. When a `links` block is present, `base_url` is r

 ## Executable support boundary

-Config validation accepts `local`, `ssh`, `s3`, and source-only `http_upload` backend shapes. Runtime execution opens `local`, `ssh`, and `s3` through `internal/app`.
+Config validation accepts `local`, `ssh`, `s3`, and source-only `http_upload` backend shapes. Runtime `run`, `validate`, and `inspect` workflows open `local`, `ssh`, and `s3` through `internal/app`. Runtime `serve` workflows execute `http_upload` sources through the app upload coordinator and HTTP server.

 SSH config uses structured fields: `host`, optional `user`, optional `port`, `path`, optional `ssh_key_file`, optional `known_hosts`, and optional `host_key_policy`. `host_key_policy` accepts YAML booleans and strings and normalizes `true`/`strict`, `accept-new`, and `false`/`off`.

 S3 config requires `endpoint` and `bucket`, normalizes optional `prefix`, defaults `region` to `us-east-1`, and defaults omitted `force_path_style` to `true` while preserving explicit `false`.

-HTTP upload config is source-only. Config owns its YAML shape, defaulting, size and duration parsing, and validation. The config package does not resolve `token_env`, authenticate requests, stage uploads, or execute HTTP upload sources.
+HTTP upload config is source-only. Config owns its YAML shape, defaulting, size and duration parsing, and validation. The config package does not authenticate requests, stage uploads, or execute HTTP upload sources. The app layer resolves `token_env` through the config-owned environment resolver before starting the HTTP server.

 ## Secrets and credential resolution

--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -26,7 +26,10 @@ Safe fix: compare the file to the reference in [configuration](config.md) and re

 ## `validate config ... backend ... is unsupported`

-Likely cause: a source or destination uses a backend name other than `local`, `ssh`, or `s3`.
+Likely cause: a source or destination uses an unsupported backend name, or a
+command is trying to execute a backend that is valid only for another workflow.
+`run`, `validate`, and `inspect` execute `local`, `ssh`, and `s3` sources.
+`serve` executes `http_upload` sources.

 Diagnostic:

@@ -34,7 +37,127 @@ Diagnostic:
 rg -n "backend:" <config-path>
 ```

-Safe fix: use `backend: local`, `backend: ssh`, or `backend: s3` for executable workflows.
+Safe fix: use `backend: local`, `backend: ssh`, or `backend: s3` for normal
+source and destination workflows. Use `backend: http_upload` only for sources
+handled by `distributor serve`.
+
+## `bind HTTP server ... address already in use`
+
+Likely cause: another process is already listening on `server.http.bind`.
+
+Diagnostic:
+
+```sh
+ss -ltnp | rg '<port>'
+```
+
+Safe fix: stop the conflicting process or configure a different
+`server.http.bind` value. The default bind address is `127.0.0.1:8080`.
+
+## `upload token environment variable ... is not set`
+
+Likely cause: a configured `http_upload` source references `token_env`, but the
+variable is absent from both the real process environment and
+`secrets.directory`.
+
+Diagnostic:
+
+```sh
+env | cut -d= -f1 | rg '^<token-variable>$'
+ls -l <secrets-directory>/<token-variable>
+```
+
+Safe fix: set the real environment variable or create a readable
+secrets-directory file with the same name. Do not place literal token values in
+YAML.
+
+## `upload token environment variables ... resolve to the same value`
+
+Likely cause: two configured `http_upload` pipelines resolve to the same bearer
+token value.
+
+Diagnostic:
+
+```sh
+rg -n 'token_env:' <config-path>
+```
+
+Safe fix: assign a distinct non-empty token value to each `http_upload`
+pipeline. Distributor does not print the duplicate token value.
+
+## `POST /upload` returns `401`
+
+Likely cause: the request is missing `Authorization: Bearer <token>` or the
+token does not match any configured `http_upload` pipeline.
+
+Diagnostic:
+
+```sh
+curl -i -X POST http://127.0.0.1:8080/upload \
+  -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \
+  -H "Content-Type: application/x-tar" \
+  --data-binary @bundle.tar
+```
+
+Safe fix: use the token value resolved by the configured `token_env`. Do not
+include token values in logs or tickets.
+
+## `POST /upload` returns `413`
+
+Likely cause: the request body exceeds the selected pipeline's
+`source.max_upload_size` or the default `server.http.max_upload_size`.
+
+Diagnostic:
+
+```sh
+ls -lh bundle.tar bundle.tar.gz
+rg -n 'max_upload_size:' <config-path>
+```
+
+Safe fix: upload a smaller archive, remove unnecessary files from the source
+bundle, or raise the configured upload size limit.
+
+## `POST /upload` returns `415`
+
+Likely cause: the upload uses an unsupported content type. The server accepts
+uncompressed tar and gzip-compressed tar archives only.
+
+Diagnostic:
+
+```sh
+file bundle.tar.gz
+```
+
+Safe fix: send `Content-Type: application/x-tar`, `application/gzip`, or
+`application/x-gzip`, matching the archive format.
+
+## `POST /upload` returns `503`
+
+Likely cause: the in-memory upload queue is full.
+
+Diagnostic:
+
+```sh
+rg -n 'queue_size|max_concurrency' <config-path>
+```
+
+Safe fix: retry after active uploads finish, or increase `server.http.queue_size`
+for the deployment.
+
+## `GET /runs/<run_id>` returns `404`
+
+Likely cause: the run id is wrong, the process restarted, or the completed
+status record expired after `server.http.retention`.
+
+Diagnostic:
+
+```sh
+curl -i http://127.0.0.1:8080/runs/<run-id>
+rg -n 'retention:' <config-path>
+```
+
+Safe fix: use the exact `run_id` returned by `POST /upload`. If status retention
+is too short for operators, increase `server.http.retention`.

 ## `--format: format must be text or json`

--- a/internal/app/serve.go
+++ b/internal/app/serve.go
@@ -0,0 +1,62 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+)
+
+type ServeOptions struct {
+	ConfigPath string
+}
+
+func Serve(ctx context.Context, options ServeOptions) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+
+	configPath := options.ConfigPath
+	if configPath == "" {
+		configPath = config.DefaultConfigPath
+	}
+	cfg, err := config.LoadFile(configPath)
+	if err != nil {
+		return err
+	}
+	secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
+	if err != nil {
+		return err
+	}
+
+	handler, err := newUploadHTTPHandler(ctx, cfg, secretLoad.Environment)
+	if err != nil {
+		return err
+	}
+	listener, err := net.Listen("tcp", cfg.Server.HTTP.Bind)
+	if err != nil {
+		return fmt.Errorf("bind HTTP server %q: %w", cfg.Server.HTTP.Bind, err)
+	}
+	defer listener.Close()
+
+	server := &http.Server{Handler: handler}
+	shutdownDone := make(chan struct{})
+	go func() {
+		defer close(shutdownDone)
+		<-ctx.Done()
+		_ = server.Shutdown(context.Background())
+	}()
+
+	err = server.Serve(listener)
+	if errors.Is(err, http.ErrServerClosed) {
+		<-shutdownDone
+		return nil
+	}
+	return err
+}
--- a/internal/app/upload_coordinator.go
+++ b/internal/app/upload_coordinator.go
@@ -201,6 +201,13 @@ func (coordinator *UploadCoordinator) Expire() []UploadRunRecord {
 	return coordinator.expireLocked(coordinator.now().UTC())
 }

+func (coordinator *UploadCoordinator) CanAccept() bool {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	coordinator.expireLocked(coordinator.now().UTC())
+	return len(coordinator.pending) < coordinator.queueSize
+}
+
 func (coordinator *UploadCoordinator) QueueDepth() int {
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
--- a/internal/app/upload_http.go
+++ b/internal/app/upload_http.go
@@ -0,0 +1,209 @@
+package app
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"mime"
+	"net/http"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+)
+
+type uploadCoordinator interface {
+	CanAccept() bool
+	Submit(context.Context, UploadRequest) (UploadRunRecord, error)
+	Status(UploadRunID) (UploadRunRecord, bool)
+}
+
+type uploadHTTPHandler struct {
+	coordinator uploadCoordinator
+	tokens      map[string]string
+	limits      map[string]int64
+}
+
+type uploadAcceptedResponse struct {
+	RunID  UploadRunID  `json:"run_id"`
+	Status UploadStatus `json:"status"`
+}
+
+type httpErrorResponse struct {
+	Error string `json:"error"`
+}
+
+func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment config.Environment) (http.Handler, error) {
+	config.ApplyDefaults(&cfg)
+	tokens, limits, err := resolveUploadTokens(cfg, environment)
+	if err != nil {
+		return nil, err
+	}
+	return uploadHTTPHandler{
+		coordinator: NewUploadCoordinator(ctx, cfg),
+		tokens:      tokens,
+		limits:      limits,
+	}, nil
+}
+
+func resolveUploadTokens(cfg config.Config, environment config.Environment) (map[string]string, map[string]int64, error) {
+	tokens := make(map[string]string)
+	limits := make(map[string]int64)
+	for _, pipeline := range cfg.Pipelines {
+		if pipeline.Source.Backend != config.BackendHTTPUpload {
+			continue
+		}
+		tokenName := pipeline.Source.Upload.TokenEnv
+		token, ok := environment.Lookup(tokenName)
+		if !ok {
+			return nil, nil, fmt.Errorf("upload token environment variable %s is not set", tokenName)
+		}
+		if token == "" {
+			return nil, nil, fmt.Errorf("upload token environment variable %s is empty", tokenName)
+		}
+		if existing, exists := tokens[token]; exists {
+			return nil, nil, fmt.Errorf("upload token environment variables for pipelines %s and %s resolve to the same value", existing, pipeline.ID)
+		}
+		tokens[token] = pipeline.ID
+		limits[pipeline.ID] = int64(*pipeline.Source.Upload.MaxUploadSize)
+	}
+	return tokens, limits, nil
+}
+
+func (handler uploadHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	switch {
+	case r.Method == http.MethodGet && r.URL.Path == "/healthz":
+		handler.handleHealth(w)
+	case r.Method == http.MethodPost && r.URL.Path == "/upload":
+		handler.handleUpload(w, r)
+	case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, "/runs/"):
+		handler.handleRunStatus(w, r)
+	default:
+		writeHTTPError(w, http.StatusNotFound, "not found")
+	}
+}
+
+func (handler uploadHTTPHandler) handleHealth(w http.ResponseWriter) {
+	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
+}
+
+func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Request) {
+	if r.URL.Query().Has("pipeline") || r.URL.Query().Has("pipeline_id") {
+		writeHTTPError(w, http.StatusBadRequest, "pipeline id is not accepted")
+		return
+	}
+	pipelineID, ok := handler.authenticate(r.Header.Get("Authorization"))
+	if !ok {
+		writeHTTPError(w, http.StatusUnauthorized, "unauthorized")
+		return
+	}
+	contentType := r.Header.Get("Content-Type")
+	if !supportedUploadContentType(contentType) {
+		writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type")
+		return
+	}
+	if !handler.coordinator.CanAccept() {
+		writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
+		return
+	}
+	body, err := readUploadBody(r.Body, handler.limits[pipelineID])
+	if err != nil {
+		if errors.Is(err, ingest.ErrUploadTooLarge) {
+			writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size")
+			return
+		}
+		writeHTTPError(w, http.StatusBadRequest, "read upload body failed")
+		return
+	}
+	record, err := handler.coordinator.Submit(r.Context(), UploadRequest{
+		PipelineID:  pipelineID,
+		ContentType: contentType,
+		Body:        bytes.NewReader(body),
+	})
+	if err != nil {
+		writeUploadSubmitError(w, err)
+		return
+	}
+	writeJSON(w, http.StatusAccepted, uploadAcceptedResponse{
+		RunID:  record.ID,
+		Status: UploadStatusAccepted,
+	})
+}
+
+func (handler uploadHTTPHandler) handleRunStatus(w http.ResponseWriter, r *http.Request) {
+	rawRunID := strings.TrimPrefix(r.URL.Path, "/runs/")
+	if rawRunID == "" || strings.Contains(rawRunID, "/") {
+		writeHTTPError(w, http.StatusNotFound, "not found")
+		return
+	}
+	record, ok := handler.coordinator.Status(UploadRunID(rawRunID))
+	if !ok {
+		writeHTTPError(w, http.StatusNotFound, "run not found")
+		return
+	}
+	writeJSON(w, http.StatusOK, record)
+}
+
+func (handler uploadHTTPHandler) authenticate(header string) (string, bool) {
+	const prefix = "Bearer "
+	if !strings.HasPrefix(header, prefix) {
+		return "", false
+	}
+	token := strings.TrimSpace(strings.TrimPrefix(header, prefix))
+	if token == "" {
+		return "", false
+	}
+	pipelineID, ok := handler.tokens[token]
+	return pipelineID, ok
+}
+
+func supportedUploadContentType(contentType string) bool {
+	mediaType, _, err := mime.ParseMediaType(contentType)
+	if err != nil {
+		mediaType = contentType
+	}
+	switch mediaType {
+	case ingest.ContentTypeTar, ingest.ContentTypeGzip, ingest.ContentTypeXGzip:
+		return true
+	default:
+		return false
+	}
+}
+
+func readUploadBody(body io.Reader, maxSize int64) ([]byte, error) {
+	limited := &io.LimitedReader{R: body, N: maxSize + 1}
+	data, err := io.ReadAll(limited)
+	if err != nil {
+		return nil, err
+	}
+	if int64(len(data)) > maxSize {
+		return nil, ingest.ErrUploadTooLarge
+	}
+	return data, nil
+}
+
+func writeUploadSubmitError(w http.ResponseWriter, err error) {
+	switch {
+	case IsUploadQueueFull(err):
+		writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
+	case errors.Is(err, ingest.ErrUploadTooLarge):
+		writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size")
+	case errors.Is(err, ingest.ErrUnsupportedContentType):
+		writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type")
+	default:
+		writeHTTPError(w, http.StatusBadRequest, "upload rejected")
+	}
+}
+
+func writeHTTPError(w http.ResponseWriter, status int, message string) {
+	writeJSON(w, status, httpErrorResponse{Error: message})
+}
+
+func writeJSON(w http.ResponseWriter, status int, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(value)
+}
--- a/internal/app/upload_http_test.go
+++ b/internal/app/upload_http_test.go
@@ -0,0 +1,331 @@
+package app
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+)
+
+type fakeUploadCoordinator struct {
+	canAccept bool
+	submit    func(context.Context, UploadRequest) (UploadRunRecord, error)
+	status    func(UploadRunID) (UploadRunRecord, bool)
+}
+
+func (fake fakeUploadCoordinator) CanAccept() bool {
+	return fake.canAccept
+}
+
+func (fake fakeUploadCoordinator) Submit(ctx context.Context, request UploadRequest) (UploadRunRecord, error) {
+	if fake.submit == nil {
+		return UploadRunRecord{}, errors.New("unexpected submit")
+	}
+	return fake.submit(ctx, request)
+}
+
+func (fake fakeUploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) {
+	if fake.status == nil {
+		return UploadRunRecord{}, false
+	}
+	return fake.status(runID)
+}
+
+func TestResolveUploadTokensFailsForMissingAndDuplicateTokens(t *testing.T) {
+	cfg := uploadHTTPTestConfig()
+
+	_, _, err := resolveUploadTokens(cfg, config.NewEnvironment(nil, func(string) (string, bool) {
+		return "", false
+	}))
+	if err == nil || !strings.Contains(err.Error(), "UPLOAD_TOKEN") {
+		t.Fatalf("resolveUploadTokens() error = %v, want missing UPLOAD_TOKEN", err)
+	}
+
+	cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{
+		ID: "weekly",
+		Source: config.Backend{
+			Backend: config.BackendHTTPUpload,
+			Upload:  config.HTTPUpload{TokenEnv: "OTHER_UPLOAD_TOKEN"},
+		},
+		Destinations: cfg.Pipelines[0].Destinations,
+	})
+	config.ApplyDefaults(&cfg)
+	secret := "super-secret-token"
+	_, _, err = resolveUploadTokens(cfg, uploadHTTPTestEnvironment(map[string]string{
+		"UPLOAD_TOKEN":       secret,
+		"OTHER_UPLOAD_TOKEN": secret,
+	}))
+	if err == nil {
+		t.Fatal("resolveUploadTokens() error = nil, want duplicate token error")
+	}
+	if strings.Contains(err.Error(), secret) {
+		t.Fatalf("duplicate token error exposed secret value: %q", err)
+	}
+}
+
+func TestNewUploadHTTPHandlerAcceptsDefaultedConfig(t *testing.T) {
+	cfg := uploadHTTPTestConfig()
+	cfg.Server.HTTP.Bind = ""
+	cfg.Server.HTTP.StagingRoot = ""
+	cfg.Server.HTTP.MaxUploadSize = nil
+	cfg.Server.HTTP.QueueSize = 0
+	cfg.Server.HTTP.MaxConcurrency = 0
+	cfg.Server.HTTP.Retention = nil
+	cfg.Pipelines[0].Source.Upload.StagingPath = ""
+	cfg.Pipelines[0].Source.Upload.MaxUploadSize = nil
+
+	handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
+		"UPLOAD_TOKEN": "secret",
+	}))
+	if err != nil {
+		t.Fatalf("newUploadHTTPHandler() error = %v", err)
+	}
+	if handler == nil {
+		t.Fatal("newUploadHTTPHandler() = nil")
+	}
+}
+
+func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
+	var submitted UploadRequest
+	handler := uploadHTTPHandler{
+		coordinator: fakeUploadCoordinator{
+			canAccept: true,
+			submit: func(_ context.Context, request UploadRequest) (UploadRunRecord, error) {
+				submitted = request
+				body, err := io.ReadAll(request.Body)
+				if err != nil {
+					t.Fatalf("read submitted body: %v", err)
+				}
+				if string(body) != "archive" {
+					t.Fatalf("submitted body = %q, want archive", body)
+				}
+				return UploadRunRecord{ID: "reports.20260603T120000Z.abcdef12", Status: UploadStatusAccepted}, nil
+			},
+		},
+		tokens: map[string]string{"valid-token": "reports"},
+		limits: map[string]int64{"reports": 1024},
+	}
+	recorder := httptest.NewRecorder()
+	request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
+	request.Header.Set("Authorization", "Bearer valid-token")
+	request.Header.Set("Content-Type", "application/x-tar")
+
+	handler.ServeHTTP(recorder, request)
+
+	if recorder.Code != http.StatusAccepted {
+		t.Fatalf("status = %d, want %d; body = %q", recorder.Code, http.StatusAccepted, recorder.Body.String())
+	}
+	if submitted.PipelineID != "reports" {
+		t.Fatalf("submitted pipeline = %q, want reports", submitted.PipelineID)
+	}
+	var response uploadAcceptedResponse
+	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
+		t.Fatalf("decode response: %v", err)
+	}
+	if response.RunID != "reports.20260603T120000Z.abcdef12" || response.Status != UploadStatusAccepted {
+		t.Fatalf("response = %#v, want accepted run id", response)
+	}
+	if strings.Contains(recorder.Body.String(), "valid-token") {
+		t.Fatalf("response exposed token: %q", recorder.Body.String())
+	}
+}
+
+func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
+	handler := uploadHTTPHandler{
+		coordinator: fakeUploadCoordinator{canAccept: true},
+		tokens:      map[string]string{"valid-token": "reports"},
+		limits:      map[string]int64{"reports": 1024},
+	}
+
+	for _, authHeader := range []string{"", "Bearer wrong-token"} {
+		recorder := httptest.NewRecorder()
+		request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
+		request.Header.Set("Authorization", authHeader)
+		request.Header.Set("Content-Type", "application/x-tar")
+
+		handler.ServeHTTP(recorder, request)
+
+		if recorder.Code != http.StatusUnauthorized {
+			t.Fatalf("auth %q status = %d, want %d", authHeader, recorder.Code, http.StatusUnauthorized)
+		}
+		if strings.Contains(recorder.Body.String(), "valid-token") || strings.Contains(recorder.Body.String(), "wrong-token") {
+			t.Fatalf("unauthorized response exposed token: %q", recorder.Body.String())
+		}
+	}
+}
+
+func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *testing.T) {
+	tests := []struct {
+		name        string
+		canAccept   bool
+		url         string
+		contentType string
+		body        io.Reader
+		wantStatus  int
+	}{
+		{
+			name:        "unsupported content type",
+			canAccept:   true,
+			url:         "/upload",
+			contentType: "application/zip",
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusUnsupportedMediaType,
+		},
+		{
+			name:        "oversized",
+			canAccept:   true,
+			url:         "/upload",
+			contentType: "application/x-tar",
+			body:        strings.NewReader("too-large"),
+			wantStatus:  http.StatusRequestEntityTooLarge,
+		},
+		{
+			name:        "full queue",
+			canAccept:   false,
+			url:         "/upload",
+			contentType: "application/x-tar",
+			body:        &countingReader{reader: strings.NewReader("archive")},
+			wantStatus:  http.StatusServiceUnavailable,
+		},
+		{
+			name:        "submitted pipeline id",
+			canAccept:   true,
+			url:         "/upload?pipeline_id=reports",
+			contentType: "application/x-tar",
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handler := uploadHTTPHandler{
+				coordinator: fakeUploadCoordinator{
+					canAccept: tt.canAccept,
+					submit: func(context.Context, UploadRequest) (UploadRunRecord, error) {
+						t.Fatal("Submit should not be called")
+						return UploadRunRecord{}, nil
+					},
+				},
+				tokens: map[string]string{"valid-token": "reports"},
+				limits: map[string]int64{"reports": 4},
+			}
+			recorder := httptest.NewRecorder()
+			request := httptest.NewRequest(http.MethodPost, tt.url, tt.body)
+			request.Header.Set("Authorization", "Bearer valid-token")
+			request.Header.Set("Content-Type", tt.contentType)
+
+			handler.ServeHTTP(recorder, request)
+
+			if recorder.Code != tt.wantStatus {
+				t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
+			}
+			if reader, ok := tt.body.(*countingReader); ok && reader.reads != 0 {
+				t.Fatalf("full queue read body %d time(s), want zero", reader.reads)
+			}
+		})
+	}
+}
+
+func TestUploadHTTPHandlerRunStatusAndHealth(t *testing.T) {
+	finishedAt := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)
+	handler := uploadHTTPHandler{
+		coordinator: fakeUploadCoordinator{
+			canAccept: true,
+			status: func(runID UploadRunID) (UploadRunRecord, bool) {
+				if runID != "reports.20260603T120000Z.abcdef12" {
+					return UploadRunRecord{}, false
+				}
+				return UploadRunRecord{
+					ID:         runID,
+					PipelineID: "reports",
+					Status:     UploadStatusSucceeded,
+					FinishedAt: &finishedAt,
+				}, true
+			},
+		},
+		tokens: map[string]string{"valid-token": "reports"},
+		limits: map[string]int64{"reports": 1024},
+	}
+
+	recorder := httptest.NewRecorder()
+	handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/healthz", nil))
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("health status = %d, want %d", recorder.Code, http.StatusOK)
+	}
+
+	recorder = httptest.NewRecorder()
+	handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/runs/reports.20260603T120000Z.abcdef12", nil))
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("run status = %d, want %d; body = %q", recorder.Code, http.StatusOK, recorder.Body.String())
+	}
+	var record UploadRunRecord
+	if err := json.Unmarshal(recorder.Body.Bytes(), &record); err != nil {
+		t.Fatalf("decode run status: %v", err)
+	}
+	if record.ID != "reports.20260603T120000Z.abcdef12" || record.Status != UploadStatusSucceeded {
+		t.Fatalf("record = %#v, want succeeded run status", record)
+	}
+
+	recorder = httptest.NewRecorder()
+	handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/runs/unknown", nil))
+	if recorder.Code != http.StatusNotFound {
+		t.Fatalf("unknown run status = %d, want %d", recorder.Code, http.StatusNotFound)
+	}
+}
+
+type countingReader struct {
+	reader io.Reader
+	reads  int
+}
+
+func (reader *countingReader) Read(data []byte) (int, error) {
+	reader.reads++
+	return reader.reader.Read(data)
+}
+
+func uploadHTTPTestConfig() config.Config {
+	size := config.ByteSize(1024)
+	retention := config.Duration(24 * time.Hour)
+	cfg := config.Config{
+		Server: config.Server{HTTP: config.HTTPServer{
+			Bind:           config.DefaultHTTPBind,
+			StagingRoot:    "/tmp/distributor-test",
+			MaxUploadSize:  &size,
+			QueueSize:      2,
+			MaxConcurrency: 1,
+			Retention:      &retention,
+		}},
+		Pipelines: []config.Pipeline{{
+			ID: "reports",
+			Source: config.Backend{
+				Backend: config.BackendHTTPUpload,
+				Upload: config.HTTPUpload{
+					TokenEnv:      "UPLOAD_TOKEN",
+					StagingPath:   "/tmp/distributor-test/reports",
+					MaxUploadSize: &size,
+				},
+			},
+			Destinations: []config.Destination{{
+				ID:      "local",
+				Backend: config.BackendLocal,
+				Path:    "/tmp/distributor-output",
+				Publish: &config.PublishPolicy{Source: true},
+			}},
+		}},
+	}
+	config.ApplyDefaults(&cfg)
+	return cfg
+}
+
+func uploadHTTPTestEnvironment(values map[string]string) config.Environment {
+	return config.NewEnvironment(values, func(string) (string, bool) {
+		return "", false
+	})
+}
--- a/internal/cli/root.go
+++ b/internal/cli/root.go
@@ -29,6 +29,8 @@ func Execute(ctx context.Context, args []string, stdout, stderr io.Writer) int {
 		return versionCommand(ctx, args[1:], stdout, stderr)
 	case "run":
 		return runCommand(ctx, args[1:], stdout, stderr)
+	case "serve":
+		return serveCommand(ctx, args[1:], stdout, stderr)
 	case "validate":
 		return validateCommand(ctx, args[1:], stdout, stderr)
 	case "inspect":
@@ -51,6 +53,7 @@ Usage:
 Commands:
  version    Print version information
  run        Run configured distribution pipelines
+  serve      Run the HTTP upload server
  validate   Validate a source bundle or bundle tree
  inspect    Inspect bundles or distributor state
  manifest   Create source bundle manifests
--- a/internal/cli/root_test.go
+++ b/internal/cli/root_test.go
@@ -11,6 +11,7 @@ import (
 	"strings"
 	"testing"

+	"gitea.maximumdirect.net/eric/distributor/internal/app"
 	"gitea.maximumdirect.net/eric/distributor/internal/storage"
 	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
 	producerbundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
@@ -92,6 +93,34 @@ func TestExecuteVersionJSON(t *testing.T) {
 	}
 }

+func TestExecuteServeParsesConfig(t *testing.T) {
+	originalServeApp := serveApp
+	defer func() {
+		serveApp = originalServeApp
+	}()
+	var gotOptions app.ServeOptions
+	serveApp = func(_ context.Context, options app.ServeOptions) error {
+		gotOptions = options
+		return nil
+	}
+	var stdout, stderr bytes.Buffer
+
+	code := Execute(context.Background(), []string{"serve", "--config", "config.yml"}, &stdout, &stderr)
+
+	if code != exitOK {
+		t.Fatalf("exit code = %d, want %d; stderr = %q", code, exitOK, stderr.String())
+	}
+	if gotOptions.ConfigPath != "config.yml" {
+		t.Fatalf("ConfigPath = %q, want config.yml", gotOptions.ConfigPath)
+	}
+	if stdout.Len() != 0 {
+		t.Fatalf("stdout = %q, want empty", stdout.String())
+	}
+	if stderr.Len() != 0 {
+		t.Fatalf("stderr = %q, want empty", stderr.String())
+	}
+}
+
 func TestExecuteRejectsInvalidFormat(t *testing.T) {
 	var stdout, stderr bytes.Buffer

--- a/internal/cli/serve.go
+++ b/internal/cli/serve.go
@@ -0,0 +1,46 @@
+package cli
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/app"
+)
+
+var serveApp = app.Serve
+
+func serveCommand(ctx context.Context, args []string, stdout, stderr io.Writer) int {
+	if hasHelp(args) {
+		printServeHelp(stdout)
+		return exitOK
+	}
+
+	flags := flag.NewFlagSet("serve", flag.ContinueOnError)
+	flags.SetOutput(stderr)
+	configPath := flags.String("config", "", "path to config file")
+	if err := flags.Parse(args); err != nil {
+		return exitUsage
+	}
+	if rejectPositionalArgs(stderr, "serve", flags.Args()) {
+		return exitUsage
+	}
+
+	if err := serveApp(ctx, app.ServeOptions{ConfigPath: *configPath}); err != nil {
+		return fail(stderr, err)
+	}
+	return exitOK
+}
+
+func printServeHelp(w io.Writer) {
+	fmt.Fprint(w, `Usage:
+  distributor serve --config <path>
+
+Options:
+  --config <path>  Path to config file
+
+Serve loads configured HTTP upload sources, resolves upload tokens through the
+configured secret environment, and starts the HTTP upload API.
+`)
+}