Add async upload coordination

docs/internal/app.md

@@ -79,6 +79,43 @@ 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, generates
+run IDs, tracks status records, stages accepted archives through
+`internal/ingest`, and executes the selected pipeline through
+`RunPipelineWithLocalSource`.
+
+Upload run IDs use:
+
+```text
+<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 staged. Execution is bounded by
+`server.http.max_concurrency`, and only one upload for a given pipeline may run
+at a time. Later 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.
+
 ## Coordination
 
 `PipelineRunCoordinator` wraps `RunPipeline` with in-memory admission control.
@@ -133,6 +170,7 @@ Run helpers are grouped by responsibility:
 - `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, queueing, status tracking, staging handoff, and staged-source execution.
 - `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`.

internal/app/upload_coordinator.go

@@ -0,0 +1,382 @@
+package app
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"sync"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+)
+
+const DefaultUploadMaxFileCount = 4096
+
+type UploadRunID string
+
+type UploadStatus string
+
+const (
+	UploadStatusAccepted  UploadStatus = "accepted"
+	UploadStatusQueued    UploadStatus = "queued"
+	UploadStatusRunning   UploadStatus = "running"
+	UploadStatusSucceeded UploadStatus = "succeeded"
+	UploadStatusFailed    UploadStatus = "failed"
+	UploadStatusExpired   UploadStatus = "expired"
+)
+
+type UploadRunRecord struct {
+	ID         UploadRunID  `json:"run_id"`
+	PipelineID string       `json:"pipeline_id"`
+	Status     UploadStatus `json:"status"`
+	AcceptedAt time.Time    `json:"accepted_at"`
+	StartedAt  *time.Time   `json:"started_at,omitempty"`
+	FinishedAt *time.Time   `json:"finished_at,omitempty"`
+	Report     *RunReport   `json:"report,omitempty"`
+	Error      string       `json:"error,omitempty"`
+	StagedRoot string       `json:"-"`
+}
+
+type UploadRequest struct {
+	PipelineID   string
+	ContentType  string
+	Body         io.Reader
+	DryRun       bool
+	Force        bool
+	MaxFileCount int
+}
+
+type UploadQueueFullError struct {
+	QueueSize int
+}
+
+func (err UploadQueueFullError) Error() string {
+	return fmt.Sprintf("upload queue is full with capacity %d", err.QueueSize)
+}
+
+func IsUploadQueueFull(err error) bool {
+	var full UploadQueueFullError
+	return errors.As(err, &full)
+}
+
+type UploadCoordinator struct {
+	ctx          context.Context
+	cfg          config.Config
+	stage        uploadStageFunc
+	run          uploadRunFunc
+	now          func() time.Time
+	randomSuffix func() (string, error)
+	retention    time.Duration
+
+	mu             sync.Mutex
+	signal         chan struct{}
+	queueSize      int
+	maxConcurrency int
+	runningCount   int
+	activePipeline map[string]bool
+	pending        []*uploadJob
+	records        map[UploadRunID]UploadRunRecord
+}
+
+type uploadStageFunc func(context.Context, ingest.StageOptions) (ingest.StagedBundle, error)
+
+type uploadRunFunc func(context.Context, config.Config, RunPipelineWithLocalSourceOptions) (RunReport, error)
+
+type uploadJob struct {
+	recordID UploadRunID
+	request  UploadRequest
+	pipeline config.Pipeline
+}
+
+type uploadCoordinatorHooks struct {
+	stage        uploadStageFunc
+	run          uploadRunFunc
+	now          func() time.Time
+	randomSuffix func() (string, error)
+}
+
+func NewUploadCoordinator(ctx context.Context, cfg config.Config) *UploadCoordinator {
+	return newUploadCoordinator(ctx, cfg, uploadCoordinatorHooks{})
+}
+
+func newUploadCoordinator(ctx context.Context, cfg config.Config, hooks uploadCoordinatorHooks) *UploadCoordinator {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	config.ApplyDefaults(&cfg)
+	stage := hooks.stage
+	if stage == nil {
+		stage = ingest.StageArchive
+	}
+	run := hooks.run
+	if run == nil {
+		run = runPipelineConfigWithLocalSource
+	}
+	now := hooks.now
+	if now == nil {
+		now = time.Now
+	}
+	randomSuffix := hooks.randomSuffix
+	if randomSuffix == nil {
+		randomSuffix = randomRunIDSuffix
+	}
+	coordinator := &UploadCoordinator{
+		ctx:            ctx,
+		cfg:            cfg,
+		stage:          stage,
+		run:            run,
+		now:            now,
+		randomSuffix:   randomSuffix,
+		retention:      cfg.Server.HTTP.Retention.AsDuration(),
+		signal:         make(chan struct{}, 1),
+		queueSize:      cfg.Server.HTTP.QueueSize,
+		maxConcurrency: cfg.Server.HTTP.MaxConcurrency,
+		activePipeline: map[string]bool{},
+		records:        map[UploadRunID]UploadRunRecord{},
+	}
+	go coordinator.dispatchLoop()
+	return coordinator
+}
+
+func (coordinator *UploadCoordinator) Submit(ctx context.Context, request UploadRequest) (UploadRunRecord, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return UploadRunRecord{}, err
+	}
+	if request.Body == nil {
+		return UploadRunRecord{}, fmt.Errorf("upload body is required")
+	}
+	pipeline, ok := findPipeline(coordinator.cfg, request.PipelineID)
+	if !ok {
+		return UploadRunRecord{}, PipelineNotFoundError{ID: request.PipelineID}
+	}
+	if pipeline.Source.Backend != config.BackendHTTPUpload {
+		return UploadRunRecord{}, fmt.Errorf("pipeline %s source backend %s is not configured for uploads", pipeline.ID, pipeline.Source.Backend)
+	}
+	runID, err := coordinator.newRunID(pipeline.ID)
+	if err != nil {
+		return UploadRunRecord{}, err
+	}
+
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	coordinator.expireLocked(coordinator.now().UTC())
+	if len(coordinator.pending) >= coordinator.queueSize {
+		return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
+	}
+	record := UploadRunRecord{
+		ID:         runID,
+		PipelineID: pipeline.ID,
+		Status:     UploadStatusAccepted,
+		AcceptedAt: coordinator.now().UTC(),
+	}
+	coordinator.records[runID] = record
+	coordinator.pending = append(coordinator.pending, &uploadJob{
+		recordID: runID,
+		request:  request,
+		pipeline: pipeline,
+	})
+	coordinator.notify()
+	return record, nil
+}
+
+func (coordinator *UploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	coordinator.expireLocked(coordinator.now().UTC())
+	record, ok := coordinator.records[runID]
+	return record, ok
+}
+
+func (coordinator *UploadCoordinator) Expire() []UploadRunRecord {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	return coordinator.expireLocked(coordinator.now().UTC())
+}
+
+func (coordinator *UploadCoordinator) QueueDepth() int {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	return len(coordinator.pending)
+}
+
+func (coordinator *UploadCoordinator) RunningCount() int {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	return coordinator.runningCount
+}
+
+func (coordinator *UploadCoordinator) newRunID(pipelineID string) (UploadRunID, error) {
+	suffix, err := coordinator.randomSuffix()
+	if err != nil {
+		return "", err
+	}
+	timestamp := coordinator.now().UTC().Format("20060102T150405Z")
+	return UploadRunID(pipelineID + "." + timestamp + "." + suffix), nil
+}
+
+func (coordinator *UploadCoordinator) dispatchLoop() {
+	for {
+		select {
+		case <-coordinator.ctx.Done():
+			return
+		case <-coordinator.signal:
+			for coordinator.startNext() {
+			}
+		}
+	}
+}
+
+func (coordinator *UploadCoordinator) startNext() bool {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	if coordinator.runningCount >= coordinator.maxConcurrency {
+		coordinator.markPendingQueuedLocked()
+		return false
+	}
+	index := -1
+	for candidateIndex, job := range coordinator.pending {
+		if coordinator.activePipeline[job.pipeline.ID] {
+			record := coordinator.records[job.recordID]
+			if record.Status == UploadStatusAccepted {
+				record.Status = UploadStatusQueued
+				coordinator.records[job.recordID] = record
+			}
+			continue
+		}
+		index = candidateIndex
+		break
+	}
+	if index < 0 {
+		return false
+	}
+
+	job := coordinator.pending[index]
+	coordinator.pending = append(coordinator.pending[:index], coordinator.pending[index+1:]...)
+	now := coordinator.now().UTC()
+	record := coordinator.records[job.recordID]
+	record.Status = UploadStatusRunning
+	record.StartedAt = &now
+	coordinator.records[job.recordID] = record
+	coordinator.runningCount++
+	coordinator.activePipeline[job.pipeline.ID] = true
+
+	go coordinator.runJob(job)
+	return true
+}
+
+func (coordinator *UploadCoordinator) markPendingQueuedLocked() {
+	for _, job := range coordinator.pending {
+		record := coordinator.records[job.recordID]
+		if record.Status == UploadStatusAccepted {
+			record.Status = UploadStatusQueued
+			coordinator.records[job.recordID] = record
+		}
+	}
+}
+
+func (coordinator *UploadCoordinator) runJob(job *uploadJob) {
+	record := coordinator.currentRecord(job.recordID)
+	maxFileCount := job.request.MaxFileCount
+	if maxFileCount <= 0 {
+		maxFileCount = DefaultUploadMaxFileCount
+	}
+	staged, err := coordinator.stage(coordinator.ctx, ingest.StageOptions{
+		Body:                job.request.Body,
+		ContentType:         job.request.ContentType,
+		PipelineStagingPath: job.pipeline.Source.Upload.StagingPath,
+		RunID:               string(record.ID),
+		MaxUploadSize:       int64(*job.pipeline.Source.Upload.MaxUploadSize),
+		MaxExtractedSize:    int64(*job.pipeline.Source.Upload.MaxUploadSize),
+		MaxFileCount:        maxFileCount,
+	})
+	if err == nil {
+		coordinator.setStagedRoot(job.recordID, staged.Root)
+		var report RunReport
+		report, err = coordinator.run(coordinator.ctx, coordinator.cfg, RunPipelineWithLocalSourceOptions{
+			PipelineID: job.pipeline.ID,
+			SourceRoot: staged.Root,
+			DryRun:     job.request.DryRun,
+			Force:      job.request.Force,
+		})
+		coordinator.complete(job, &report, err)
+		return
+	}
+	coordinator.complete(job, nil, err)
+}
+
+func (coordinator *UploadCoordinator) currentRecord(runID UploadRunID) UploadRunRecord {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	return coordinator.records[runID]
+}
+
+func (coordinator *UploadCoordinator) setStagedRoot(runID UploadRunID, root string) {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	record := coordinator.records[runID]
+	record.StagedRoot = root
+	coordinator.records[runID] = record
+}
+
+func (coordinator *UploadCoordinator) complete(job *uploadJob, report *RunReport, runErr error) {
+	coordinator.mu.Lock()
+	defer coordinator.mu.Unlock()
+	record := coordinator.records[job.recordID]
+	finishedAt := coordinator.now().UTC()
+	record.FinishedAt = &finishedAt
+	record.Report = report
+	if runErr != nil {
+		record.Status = UploadStatusFailed
+		record.Error = runErr.Error()
+	} else {
+		record.Status = UploadStatusSucceeded
+	}
+	coordinator.records[job.recordID] = record
+	coordinator.runningCount--
+	delete(coordinator.activePipeline, job.pipeline.ID)
+	coordinator.notify()
+}
+
+func (coordinator *UploadCoordinator) expireLocked(now time.Time) []UploadRunRecord {
+	var expired []UploadRunRecord
+	for runID, record := range coordinator.records {
+		if record.FinishedAt == nil || record.Status == UploadStatusExpired {
+			continue
+		}
+		if now.Before(record.FinishedAt.Add(coordinator.retention)) {
+			continue
+		}
+		if record.StagedRoot != "" {
+			_ = os.RemoveAll(record.StagedRoot)
+		}
+		record.Status = UploadStatusExpired
+		record.Report = nil
+		record.Error = ""
+		expired = append(expired, record)
+		delete(coordinator.records, runID)
+	}
+	return expired
+}
+
+func (coordinator *UploadCoordinator) notify() {
+	select {
+	case coordinator.signal <- struct{}{}:
+	default:
+	}
+}
+
+func randomRunIDSuffix() (string, error) {
+	var data [4]byte
+	if _, err := rand.Read(data[:]); err != nil {
+		return "", fmt.Errorf("generate run id suffix: %w", err)
+	}
+	return hex.EncodeToString(data[:]), nil
+}

internal/app/upload_coordinator_test.go

@@ -0,0 +1,373 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/distributor/internal/config"
+	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+)
+
+func TestUploadCoordinatorGeneratesRunIDAndAcceptedStatus(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	clock := newUploadTestClock(time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC))
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"weather-daily"},
+	}), uploadCoordinatorHooks{
+		now:          clock.Now,
+		randomSuffix: uploadTestSuffixes("ab12cd34"),
+		stage:        successfulUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	record, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:  "weather-daily",
+		ContentType: ingest.ContentTypeTar,
+		Body:        strings.NewReader("archive"),
+	})
+	if err != nil {
+		t.Fatalf("Submit() error = %v", err)
+	}
+	if got, want := record.ID, UploadRunID("weather-daily.20260603T120000Z.ab12cd34"); got != want {
+		t.Fatalf("run id = %q, want %q", got, want)
+	}
+	if got, want := record.Status, UploadStatusAccepted; got != want {
+		t.Fatalf("initial status = %q, want %q", got, want)
+	}
+	waitForUploadStatus(t, coordinator, record.ID, UploadStatusSucceeded)
+}
+
+func TestUploadCoordinatorRejectsFullQueueBeforeReadingBody(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	release := make(chan struct{})
+	var reads atomic.Int64
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs:    []string{"reports"},
+		queueSize:      1,
+		maxConcurrency: 1,
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002", "00000003"),
+		stage:        successfulUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			<-release
+			return RunReport{}, nil
+		},
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:  "reports",
+		ContentType: ingest.ContentTypeTar,
+		Body:        strings.NewReader("first"),
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusRunning)
+
+	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:  "reports",
+		ContentType: ingest.ContentTypeTar,
+		Body:        strings.NewReader("second"),
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, second.ID, UploadStatusQueued)
+
+	_, err = coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:  "reports",
+		ContentType: ingest.ContentTypeTar,
+		Body: readerFunc(func(data []byte) (int, error) {
+			reads.Add(1)
+			return 0, io.EOF
+		}),
+	})
+	if err == nil || !IsUploadQueueFull(err) {
+		t.Fatalf("third Submit() error = %v, want full queue", err)
+	}
+	if got := reads.Load(); got != 0 {
+		t.Fatalf("rejected body reads = %d, want 0", got)
+	}
+	close(release)
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+	waitForUploadStatus(t, coordinator, second.ID, UploadStatusSucceeded)
+}
+
+func TestUploadCoordinatorSerializesSamePipelineUploads(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	release := make(chan struct{})
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs:    []string{"reports"},
+		queueSize:      4,
+		maxConcurrency: 2,
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        successfulUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			<-release
+			return RunReport{}, nil
+		},
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("first")})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	second, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("second")})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusRunning)
+	waitForUploadStatus(t, coordinator, second.ID, UploadStatusQueued)
+	if got := coordinator.RunningCount(); got != 1 {
+		t.Fatalf("running count = %d, want 1", got)
+	}
+	close(release)
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+	waitForUploadStatus(t, coordinator, second.ID, UploadStatusSucceeded)
+}
+
+func TestUploadCoordinatorRunsDifferentPipelinesConcurrentlyUpToLimit(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	release := make(chan struct{})
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs:    []string{"reports-one", "reports-two"},
+		queueSize:      4,
+		maxConcurrency: 2,
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        successfulUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			<-release
+			return RunReport{}, nil
+		},
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports-one", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("first")})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	second, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports-two", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("second")})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusRunning)
+	waitForUploadStatus(t, coordinator, second.ID, UploadStatusRunning)
+	if got := coordinator.RunningCount(); got != 2 {
+		t.Fatalf("running count = %d, want 2", got)
+	}
+	close(release)
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+	waitForUploadStatus(t, coordinator, second.ID, UploadStatusSucceeded)
+}
+
+func TestUploadCoordinatorRecordsFailureDetails(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	runErr := errors.New("publish failed")
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001"),
+		stage:        successfulUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			return RunReport{DryRun: options.DryRun}, runErr
+		},
+	})
+
+	record, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:  "reports",
+		ContentType: ingest.ContentTypeTar,
+		Body:        strings.NewReader("archive"),
+		DryRun:      true,
+	})
+	if err != nil {
+		t.Fatalf("Submit() error = %v", err)
+	}
+	failed := waitForUploadStatus(t, coordinator, record.ID, UploadStatusFailed)
+	if failed.Error != runErr.Error() {
+		t.Fatalf("error = %q, want %q", failed.Error, runErr.Error())
+	}
+	if failed.Report == nil || !failed.Report.DryRun {
+		t.Fatalf("report = %#v, want retained dry-run report", failed.Report)
+	}
+}
+
+func TestUploadCoordinatorExpiresCompletedRecordsAndStagingDirectories(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	clock := newUploadTestClock(time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC))
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+		retention:   time.Second,
+	}), uploadCoordinatorHooks{
+		now:          clock.Now,
+		randomSuffix: uploadTestSuffixes("00000001"),
+		stage:        successfulUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			return RunReport{DryRun: true}, nil
+		},
+	})
+
+	record, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:  "reports",
+		ContentType: ingest.ContentTypeTar,
+		Body:        strings.NewReader("archive"),
+	})
+	if err != nil {
+		t.Fatalf("Submit() error = %v", err)
+	}
+	succeeded := waitForUploadStatus(t, coordinator, record.ID, UploadStatusSucceeded)
+	if succeeded.Report == nil || !succeeded.Report.DryRun {
+		t.Fatalf("report = %#v, want retained dry-run report", succeeded.Report)
+	}
+	if _, err := os.Stat(succeeded.StagedRoot); err != nil {
+		t.Fatalf("staged root stat before expiry = %v", err)
+	}
+
+	clock.Advance(2 * time.Second)
+	expired := coordinator.Expire()
+	if got, want := len(expired), 1; got != want {
+		t.Fatalf("expired count = %d, want %d", got, want)
+	}
+	if expired[0].Status != UploadStatusExpired || expired[0].Report != nil || expired[0].Error != "" {
+		t.Fatalf("expired record = %#v, want expired without report/error", expired[0])
+	}
+	if _, ok := coordinator.Status(record.ID); ok {
+		t.Fatal("Status() ok = true after expiry, want removed status")
+	}
+	if _, err := os.Stat(succeeded.StagedRoot); !os.IsNotExist(err) {
+		t.Fatalf("staged root stat after expiry = %v, want not exist", err)
+	}
+}
+
+type readerFunc func([]byte) (int, error)
+
+func (fn readerFunc) Read(data []byte) (int, error) {
+	return fn(data)
+}
+
+func successfulUploadStage(ctx context.Context, opts ingest.StageOptions) (ingest.StagedBundle, error) {
+	root := filepath.Join(opts.PipelineStagingPath, opts.RunID)
+	if err := os.MkdirAll(root, 0o755); err != nil {
+		return ingest.StagedBundle{}, err
+	}
+	return ingest.StagedBundle{Root: root}, nil
+}
+
+func successfulUploadRun(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+	return RunReport{}, nil
+}
+
+type uploadCoordinatorConfigOptions struct {
+	pipelineIDs    []string
+	queueSize      int
+	maxConcurrency int
+	retention      time.Duration
+}
+
+func uploadCoordinatorConfig(t *testing.T, opts uploadCoordinatorConfigOptions) config.Config {
+	t.Helper()
+	queueSize := opts.queueSize
+	if queueSize == 0 {
+		queueSize = 4
+	}
+	maxConcurrency := opts.maxConcurrency
+	if maxConcurrency == 0 {
+		maxConcurrency = 1
+	}
+	retentionValue := opts.retention
+	if retentionValue == 0 {
+		retentionValue = time.Minute
+	}
+	retention := config.Duration(retentionValue)
+	maxUploadSize := config.ByteSize(1024)
+	cfg := config.Config{
+		Server: config.Server{HTTP: config.HTTPServer{
+			StagingRoot:    t.TempDir(),
+			MaxUploadSize:  &maxUploadSize,
+			QueueSize:      queueSize,
+			MaxConcurrency: maxConcurrency,
+			Retention:      &retention,
+		}},
+	}
+	for _, pipelineID := range opts.pipelineIDs {
+		cfg.Pipelines = append(cfg.Pipelines, config.Pipeline{
+			ID: pipelineID,
+			Source: config.Backend{
+				Backend: config.BackendHTTPUpload,
+				Upload:  config.HTTPUpload{TokenEnv: strings.ToUpper(strings.ReplaceAll(pipelineID, "-", "_")) + "_TOKEN"},
+			},
+			Destinations: []config.Destination{{
+				ID:      "archive",
+				Backend: config.BackendLocal,
+				Path:    t.TempDir(),
+			}},
+		})
+	}
+	return cfg
+}
+
+type uploadTestClock struct {
+	mu  sync.Mutex
+	now time.Time
+}
+
+func newUploadTestClock(now time.Time) *uploadTestClock {
+	return &uploadTestClock{now: now}
+}
+
+func (clock *uploadTestClock) Now() time.Time {
+	clock.mu.Lock()
+	defer clock.mu.Unlock()
+	return clock.now
+}
+
+func (clock *uploadTestClock) Advance(duration time.Duration) {
+	clock.mu.Lock()
+	defer clock.mu.Unlock()
+	clock.now = clock.now.Add(duration)
+}
+
+func uploadTestSuffixes(values ...string) func() (string, error) {
+	var mu sync.Mutex
+	index := 0
+	return func() (string, error) {
+		mu.Lock()
+		defer mu.Unlock()
+		if index >= len(values) {
+			return fmt.Sprintf("%08d", index+1), nil
+		}
+		value := values[index]
+		index++
+		return value, nil
+	}
+}
+
+func waitForUploadStatus(t *testing.T, coordinator *UploadCoordinator, runID UploadRunID, status UploadStatus) UploadRunRecord {
+	t.Helper()
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		record, ok := coordinator.Status(runID)
+		if ok && record.Status == status {
+			return record
+		}
+		time.Sleep(time.Millisecond)
+	}
+	record, ok := coordinator.Status(runID)
+	t.Fatalf("timed out waiting for status %s; latest ok=%t record=%#v", status, ok, record)
+	return UploadRunRecord{}
+}