Add HTTP upload end-to-end coverage

docs/operations.md

@@ -68,6 +68,83 @@ Validate one configured source without opening destinations:
 go run ./cmd/distributor validate --config examples/local-publish.yml --pipeline example-source-bundle
 ```
 
+## HTTP Upload Workflow
+
+`distributor serve` runs the HTTP upload API for pipelines whose source backend
+is `http_upload`. Each upload token maps to one configured pipeline, and each
+accepted archive is staged, validated, and published through the same
+destination fan-out path used by local source runs.
+
+Minimal local HTTP upload configuration:
+
+```yaml
+server:
+  http:
+    bind: 127.0.0.1:8080
+    staging_root: /var/spool/distributor
+    max_upload_size: 20MB
+    queue_size: 16
+    max_concurrency: 1
+    retention: 24h
+secrets:
+  directory: /run/secrets/distributor
+pipelines:
+  - id: reports
+    source:
+      backend: http_upload
+      token_env: REPORTS_UPLOAD_TOKEN
+    destinations:
+      - id: archive
+        backend: local
+        path: /srv/reports/archive
+```
+
+Create `/run/secrets/distributor/REPORTS_UPLOAD_TOKEN` or set the real process
+environment variable before starting the server. Distributor does not read
+literal upload tokens from YAML.
+
+Start the server:
+
+```sh
+go run ./cmd/distributor serve --config <config-path>
+```
+
+Submit a tar or tar.gz source bundle:
+
+```sh
+curl -X POST http://127.0.0.1:8080/upload \
+  -H "Authorization: Bearer $REPORTS_UPLOAD_TOKEN" \
+  -H "Content-Type: application/gzip" \
+  --data-binary @bundle.tar.gz
+```
+
+Successful admission returns a run id:
+
+```json
+{"run_id":"reports.20260603T120000Z.abcdef12","status":"accepted"}
+```
+
+Poll status until it reaches `succeeded` or `failed`:
+
+```sh
+curl http://127.0.0.1:8080/runs/<run-id>
+```
+
+The status record includes the completed run report on successful publication
+or error details on failure. Status is memory-only and expires after
+`server.http.retention`; completed staged bundle directories are removed on
+expiry. Restarting the process clears upload status and queue state.
+
+Use `GET /healthz` for readiness after config and tokens load:
+
+```sh
+curl http://127.0.0.1:8080/healthz
+```
+
+The default bind address is private loopback. Put TLS, public routing,
+rate-limiting, and external access policy in a reverse proxy or deployment
+layer.
+
 ## Filesystem Layout
 
 Source bundles are discovered beneath the configured source root. Each bundle is a directory containing `manifest.json`.
@@ -322,7 +399,10 @@ secrets:
   directory: /run/secrets/distributor
 ```
 
-The directory is loaded during `run` and configured-source `validate` or `inspect` before any backend is opened. If the directory is missing, unreadable, or contains an invalid secret filename, the command fails before storage work starts.
+The directory is loaded during `run`, `serve`, and configured-source `validate`
+or `inspect` before credential-consuming work starts. If the directory is
+missing, unreadable, or contains an invalid secret filename, the command fails
+before storage work starts.
 
 Real process environment values take precedence over files with the same name. If the values differ and stdout is enabled, `run` and configured-source diagnostics print a warning naming the ignored secret file variable without printing either value. The process environment is not changed.
 

internal/app/upload_http_integration_test.go

@@ -0,0 +1,419 @@
+package app
+
+import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/fs"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+	"gitea.maximumdirect.net/eric/distributor/internal/storage"
+	"gitea.maximumdirect.net/eric/distributor/internal/testutil"
+)
+
+func TestHTTPUploadPublishesTarAndGzipFanout(t *testing.T) {
+	tests := []struct {
+		name        string
+		compressed  bool
+		contentType string
+	}{
+		{name: "tar", contentType: ingest.ContentTypeTar},
+		{name: "gzip", compressed: true, contentType: ingest.ContentTypeGzip},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			firstDestination := t.TempDir()
+			secondDestination := t.TempDir()
+			cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+				id:           "reports",
+				tokenEnv:     "REPORTS_TOKEN",
+				stagingPath:  filepath.Join(t.TempDir(), "reports"),
+				destinations: []string{firstDestination, secondDestination},
+			}}, 4, 1)
+			handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
+				"REPORTS_TOKEN": "reports-secret",
+			}))
+			if err != nil {
+				t.Fatalf("newUploadHTTPHandler() error = %v", err)
+			}
+			server := httptest.NewServer(handler)
+			defer server.Close()
+
+			runID := submitHTTPUpload(t, server, "reports-secret", tt.contentType, bundleArchive(t, tt.compressed, testutil.BundleOptions{}))
+			record := waitForHTTPUploadStatus(t, server, runID, UploadStatusSucceeded)
+
+			if record.Report == nil {
+				t.Fatal("completed status report = nil, want run report")
+			}
+			if record.Report.Summary.Status != "ok" {
+				t.Fatalf("summary status = %q, want ok", record.Report.Summary.Status)
+			}
+			if got, want := len(record.Report.Actions), 2; got != want {
+				t.Fatalf("action count = %d, want %d", got, want)
+			}
+			assertPublishedBundle(t, firstDestination)
+			assertPublishedBundle(t, secondDestination)
+		})
+	}
+}
+
+func TestHTTPUploadInvalidArchiveFailsWithoutPublishing(t *testing.T) {
+	destination := t.TempDir()
+	cfg := httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "REPORTS_TOKEN",
+		stagingPath:  filepath.Join(t.TempDir(), "reports"),
+		destinations: []string{destination},
+	}}, 4, 1)
+	handler, err := newUploadHTTPHandler(context.Background(), cfg, uploadHTTPTestEnvironment(map[string]string{
+		"REPORTS_TOKEN": "reports-secret",
+	}))
+	if err != nil {
+		t.Fatalf("newUploadHTTPHandler() error = %v", err)
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	runID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("not a tar archive"))
+	record := waitForHTTPUploadStatus(t, server, runID, UploadStatusFailed)
+
+	if record.Error == "" {
+		t.Fatal("failed status error is empty")
+	}
+	if record.Report != nil {
+		t.Fatalf("failed staging report = %#v, want nil", record.Report)
+	}
+	assertDirectoryEmpty(t, destination)
+}
+
+func TestHTTPUploadSamePipelineRequestsSerialize(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	release := make(chan struct{})
+	started := make(chan struct{}, 1)
+	coordinator := newUploadCoordinator(ctx, httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "REPORTS_TOKEN",
+		stagingPath:  filepath.Join(t.TempDir(), "reports"),
+		destinations: []string{t.TempDir()},
+	}}, 4, 2), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        successfulUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			select {
+			case started <- struct{}{}:
+			default:
+			}
+			<-release
+			return RunReport{}, nil
+		},
+	})
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens:      map[string]string{"reports-secret": "reports"},
+		limits:      map[string]int64{"reports": 1024},
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("first"))
+	waitForRunStart(t, started)
+	first := waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning)
+	secondRunID := submitHTTPUpload(t, server, "reports-secret", ingest.ContentTypeTar, []byte("second"))
+	second := waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusQueued)
+
+	if first.PipelineID != "reports" || second.PipelineID != "reports" {
+		t.Fatalf("statuses = %#v %#v, want same pipeline", first, second)
+	}
+	if got := coordinator.RunningCount(); got != 1 {
+		t.Fatalf("running count = %d, want 1", got)
+	}
+	close(release)
+	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded)
+}
+
+func TestHTTPUploadDifferentPipelinesRunConcurrently(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	release := make(chan struct{})
+	started := make(chan string, 2)
+	coordinator := newUploadCoordinator(ctx, httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{
+		{
+			id:           "reports-one",
+			tokenEnv:     "REPORTS_ONE_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "reports-one"),
+			destinations: []string{t.TempDir()},
+		},
+		{
+			id:           "reports-two",
+			tokenEnv:     "REPORTS_TWO_TOKEN",
+			stagingPath:  filepath.Join(t.TempDir(), "reports-two"),
+			destinations: []string{t.TempDir()},
+		},
+	}, 4, 2), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        successfulUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			started <- options.PipelineID
+			<-release
+			return RunReport{}, nil
+		},
+	})
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens: map[string]string{
+			"one-secret": "reports-one",
+			"two-secret": "reports-two",
+		},
+		limits: map[string]int64{
+			"reports-one": 1024,
+			"reports-two": 1024,
+		},
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUpload(t, server, "one-secret", ingest.ContentTypeTar, []byte("first"))
+	secondRunID := submitHTTPUpload(t, server, "two-secret", ingest.ContentTypeTar, []byte("second"))
+	waitForStartedPipelines(t, started, "reports-one", "reports-two")
+	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusRunning)
+	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusRunning)
+
+	if got := coordinator.RunningCount(); got != 2 {
+		t.Fatalf("running count = %d, want 2", got)
+	}
+	close(release)
+	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+	waitForHTTPUploadStatus(t, server, secondRunID, UploadStatusSucceeded)
+}
+
+type httpUploadPipelineSpec struct {
+	id           string
+	tokenEnv     string
+	stagingPath  string
+	destinations []string
+}
+
+func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpec, queueSize, maxConcurrency int) config.Config {
+	t.Helper()
+	size := config.ByteSize(1024 * 1024)
+	retention := config.Duration(time.Minute)
+	cfg := config.Config{
+		Server: config.Server{HTTP: config.HTTPServer{
+			Bind:           config.DefaultHTTPBind,
+			StagingRoot:    t.TempDir(),
+			MaxUploadSize:  &size,
+			QueueSize:      queueSize,
+			MaxConcurrency: maxConcurrency,
+			Retention:      &retention,
+		}},
+	}
+	for _, spec := range pipelines {
+		pipeline := config.Pipeline{
+			ID: spec.id,
+			Source: config.Backend{
+				Backend: config.BackendHTTPUpload,
+				Upload: config.HTTPUpload{
+					TokenEnv:      spec.tokenEnv,
+					StagingPath:   spec.stagingPath,
+					MaxUploadSize: &size,
+				},
+			},
+		}
+		for index, destination := range spec.destinations {
+			pipeline.Destinations = append(pipeline.Destinations, config.Destination{
+				ID:      fmt.Sprintf("archive-%d", index+1),
+				Backend: config.BackendLocal,
+				Path:    destination,
+				Publish: &config.PublishPolicy{Source: true},
+			})
+		}
+		cfg.Pipelines = append(cfg.Pipelines, pipeline)
+	}
+	config.ApplyDefaults(&cfg)
+	return cfg
+}
+
+func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID {
+	t.Helper()
+	request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("NewRequest() error = %v", err)
+	}
+	request.Header.Set("Authorization", "Bearer "+token)
+	request.Header.Set("Content-Type", contentType)
+	response, err := server.Client().Do(request)
+	if err != nil {
+		t.Fatalf("POST /upload error = %v", err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusAccepted {
+		t.Fatalf("POST /upload status = %d, want %d", response.StatusCode, http.StatusAccepted)
+	}
+	var accepted uploadAcceptedResponse
+	if err := json.NewDecoder(response.Body).Decode(&accepted); err != nil {
+		t.Fatalf("decode accepted response: %v", err)
+	}
+	if accepted.RunID == "" || accepted.Status != UploadStatusAccepted {
+		t.Fatalf("accepted response = %#v, want run id and accepted status", accepted)
+	}
+	return accepted.RunID
+}
+
+func waitForHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID, status UploadStatus) UploadRunRecord {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	var latest UploadRunRecord
+	var latestStatus int
+	for time.Now().Before(deadline) {
+		latest, latestStatus = getHTTPUploadStatus(t, server, runID)
+		if latestStatus == http.StatusOK && latest.Status == status {
+			return latest
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for status %s; latest HTTP status=%d record=%#v", status, latestStatus, latest)
+	return UploadRunRecord{}
+}
+
+func getHTTPUploadStatus(t *testing.T, server *httptest.Server, runID UploadRunID) (UploadRunRecord, int) {
+	t.Helper()
+	response, err := server.Client().Get(server.URL + "/runs/" + string(runID))
+	if err != nil {
+		t.Fatalf("GET /runs error = %v", err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode != http.StatusOK {
+		return UploadRunRecord{}, response.StatusCode
+	}
+	var record UploadRunRecord
+	if err := json.NewDecoder(response.Body).Decode(&record); err != nil {
+		t.Fatalf("decode run status: %v", err)
+	}
+	return record, response.StatusCode
+}
+
+func bundleArchive(t *testing.T, compressed bool, opts testutil.BundleOptions) []byte {
+	t.Helper()
+	root := t.TempDir()
+	testutil.WriteSourceBundle(t, root, "", opts)
+	return tarDirectory(t, root, compressed)
+}
+
+func tarDirectory(t *testing.T, root string, compressed bool) []byte {
+	t.Helper()
+	var output bytes.Buffer
+	var writer io.WriteCloser = nopWriteCloser{writer: &output}
+	if compressed {
+		gzipWriter := gzip.NewWriter(&output)
+		writer = gzipWriter
+	}
+	tarWriter := tar.NewWriter(writer)
+	if err := filepath.WalkDir(root, func(filePath string, entry fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if entry.IsDir() {
+			return nil
+		}
+		relative, err := filepath.Rel(root, filePath)
+		if err != nil {
+			return err
+		}
+		data, err := os.ReadFile(filePath)
+		if err != nil {
+			return err
+		}
+		header := &tar.Header{
+			Name: filepath.ToSlash(relative),
+			Mode: 0o600,
+			Size: int64(len(data)),
+		}
+		if err := tarWriter.WriteHeader(header); err != nil {
+			return err
+		}
+		if _, err := tarWriter.Write(data); err != nil {
+			return err
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("walk bundle: %v", err)
+	}
+	if err := tarWriter.Close(); err != nil {
+		t.Fatalf("close tar: %v", err)
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatalf("close archive: %v", err)
+	}
+	return output.Bytes()
+}
+
+type nopWriteCloser struct {
+	writer io.Writer
+}
+
+func (writer nopWriteCloser) Write(data []byte) (int, error) {
+	return writer.writer.Write(data)
+}
+
+func (writer nopWriteCloser) Close() error {
+	return nil
+}
+
+func assertPublishedBundle(t *testing.T, destinationRoot string) {
+	t.Helper()
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nSunny.\n")
+	testutil.AssertFile(t, filepath.Join(destinationRoot, "summary.txt"), "Summary\n")
+	if _, err := os.Stat(filepath.Join(destinationRoot, storage.StateFileName)); err != nil {
+		t.Fatalf("destination state stat: %v", err)
+	}
+}
+
+func assertDirectoryEmpty(t *testing.T, root string) {
+	t.Helper()
+	entries, err := os.ReadDir(root)
+	if err != nil {
+		t.Fatalf("ReadDir() error = %v", err)
+	}
+	if len(entries) != 0 {
+		t.Fatalf("directory %s has %d entries, want empty", root, len(entries))
+	}
+}
+
+func waitForRunStart(t *testing.T, started <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-started:
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for run start")
+	}
+}
+
+func waitForStartedPipelines(t *testing.T, started <-chan string, want ...string) {
+	t.Helper()
+	remaining := map[string]bool{}
+	for _, pipelineID := range want {
+		remaining[pipelineID] = true
+	}
+	deadline := time.After(time.Second)
+	for len(remaining) > 0 {
+		select {
+		case pipelineID := <-started:
+			delete(remaining, pipelineID)
+		case <-deadline:
+			t.Fatalf("timed out waiting for pipelines to start; remaining=%v", remaining)
+		}
+	}
+}