Add HTTP upload idempotency support

2026-06-04 14:03:26 +00:00
parent 1a402e6cfa
commit a15722571f
9 changed files with 603 additions and 36 deletions
--- a/docs/integrations/http-upload.md
+++ b/docs/integrations/http-upload.md
@@ -30,6 +30,14 @@ Returns `200 OK` when the server is running:

 Accepts one source bundle archive and returns after the archive is staged and validated.

+Producers may include:
+
+```text
+Idempotency-Key: <key>
+```
+
+Idempotency keys are scoped to the authenticated pipeline selected by the bearer token. Valid keys are non-empty ASCII strings up to 128 bytes using letters, digits, `.`, `_`, `-`, and `:`. Invalid keys return `400`.
+
 Accepted content types:

 - `application/x-tar`
@@ -44,8 +52,9 @@ Successful admission returns `202 Accepted`:

 Common error responses:

- `400`: pipeline query supplied, archive rejected, malformed archive, or invalid staged source bundle.
+- `400`: pipeline query supplied, invalid idempotency key, archive rejected, malformed archive, or invalid staged source bundle.
 - `401`: missing, empty, or unknown bearer token.
+- `409`: repeated idempotency key conflicts with another source manifest, or the same key is already being staged.
 - `413`: upload body exceeds the selected pipeline size limit.
 - `415`: unsupported content type.
 - `503`: upload queue is full.
@@ -56,6 +65,14 @@ Error bodies use:
 {"error":"<message>"}
 ```

+Retryable idempotency conflicts include:
+
+```json
+{"error":"upload idempotency key is already being processed","retryable":true}
+```
+
+When `Idempotency-Key` is omitted, upload admission preserves the raw HTTP behavior: every valid accepted upload receives its own run id. When a key is supplied, the server records the accepted run after archive staging and source bundle validation succeed. Reusing the same key for the same authenticated pipeline and the same normalized source manifest returns the original `202 Accepted` response and does not enqueue another run. Reusing the same key for a different normalized source manifest returns `409 Conflict`.
+
 ### `GET /runs/<run-id>`

 Returns an in-memory status record while retained:
@@ -95,6 +112,8 @@ The uploaded archive size and extracted bundle size are bounded by the selected

 Completed status records expire after `server.http.retention`; expiration removes committed staged bundle directories for completed uploads. Server restart clears queue state and status records.

+Idempotency records are memory-only, expire with completed upload status records, and are cleared by server restart.
+
 ## Boundaries

 The HTTP API does not expose pipeline selection by request parameter, TLS, public routing policy, or durable status storage. Put public access controls, TLS termination, and rate limiting in deployment infrastructure.
--- a/docs/internal/app.md
+++ b/docs/internal/app.md
@@ -36,17 +36,19 @@ Run workflows discover and validate source bundles through `internal/bundle`. De

 HTTP uploads stage and validate archives before enqueueing a pipeline run with a local staged source root.

+Upload idempotency is owned by the upload coordinator. Optional `Idempotency-Key` values are scoped to the authenticated pipeline. The coordinator reserves a key while staging is in progress, records the accepted run id with the validated source manifest identity after staging succeeds, returns the original accepted record for the same key and same manifest, and rejects the same key with a different manifest as a conflict.
+
 ## Skip And Resume Behavior

 Fan-out destinations are independent. A destination failure is recorded and does not prevent later destinations from being attempted. Dry-run builds plans and reports without destination writes, destination state writes, notifier calls, or SSH known-host persistence.

-HTTP upload status is in memory. Accepted jobs move through accepted, queued, running, succeeded, or failed states and expire after configured retention.
+HTTP upload status is in memory. Accepted jobs move through accepted, queued, running, succeeded, or failed states and expire after configured retention. Upload idempotency records are also memory-only, expire with the completed status record for their accepted run, and are cleared by process restart.

 ## Failure Behavior

 Runtime setup fails for config load, config validation, secret loading, or credential resolution errors. Source setup failures stop the affected run before destination planning. Destination open, planning, execution, and notification failures are recorded as destination failures where a partial result exists.

-HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Upload requests can fail during authentication, content-type validation, queue admission, archive staging, source validation, or later publish execution.
+HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Upload requests can fail during authentication, idempotency-key validation, content-type validation, idempotency conflict checks, queue admission, archive staging, source validation, or later publish execution.

 ## Tests To Inspect

@@ -63,4 +65,5 @@ HTTP upload startup fails if upload tokens are missing, empty, or duplicated. Up
 - Fan-out destinations remain independent after a destination-scoped failure.
 - Secret values are never printed; warnings may name variables only.
 - Upload admission stages and validates a bundle before returning a run id.
+- Idempotent upload retries compare normalized source manifest identity, not archive bytes.
 - Runtime backend registration remains app-owned.
--- a/docs/operations.md
+++ b/docs/operations.md
@@ -138,6 +138,16 @@ curl -X POST http://127.0.0.1:8080/upload \
  --data-binary @bundle.tar.gz
 ```

+For safe producer retries, include an idempotency key that is stable for the producer operation:
+
+```sh
+curl -X POST http://127.0.0.1:8080/upload \
+  -H "Authorization: Bearer $DISTRIBUTOR_EXAMPLE_UPLOAD_TOKEN" \
+  -H "Content-Type: application/gzip" \
+  -H "Idempotency-Key: producer.run.20260604T120000Z" \
+  --data-binary @bundle.tar.gz
+```
+
 Accepted uploads return after the archive is staged and validated:

 ```json
@@ -154,6 +164,8 @@ Status values are `accepted`, `queued`, `running`, `succeeded`, and `failed`. Co

 Upload admission is bounded by `server.http.queue_size`. Publication concurrency is bounded by `server.http.max_concurrency`, and the coordinator does not run two uploads for the same pipeline at the same time.

+`Idempotency-Key` is optional for raw HTTP clients. When present, it is scoped to the authenticated pipeline. Reusing the same key with the same normalized source manifest returns the original accepted run response and does not enqueue another run. Reusing the key with a different source manifest returns `409 Conflict`. If another request with the same key is still being staged before its manifest is known, the server returns a retryable `409 Conflict`. Idempotency records are memory-only and expire with completed upload status records.
+
 The upload server accepts `application/x-tar`, `application/gzip`, and `application/x-gzip`. Archives are extracted into a temporary staging directory, must contain exactly one root-level `manifest.json`, and must validate as one complete source bundle before a run id is issued. Per-source `max_upload_size` bounds both uploaded archive size and extracted bundle size. The implementation also caps extracted file count.

 The default bind address is private loopback. Put TLS, public routing, rate limiting, and external access policy in a reverse proxy or deployment layer.
--- a/docs/troubleshooting.md
+++ b/docs/troubleshooting.md
@@ -451,6 +451,26 @@ Safe fix: send one valid tar or tar.gz source bundle archive with `Content-Type:

 Reference: [Operations](operations.md#http-upload-operation).

+## Upload Idempotency Conflict
+
+Symptom: `POST /upload` returns `409`.
+
+Likely cause: the request reused an `Idempotency-Key` for the same authenticated pipeline with a different source manifest, or another request with the same key is still being staged before its manifest is known.
+
+Diagnostic:
+
+```sh
+curl -i -X POST http://127.0.0.1:8080/upload \
+  -H "Authorization: Bearer $DISTRIBUTOR_UPLOAD_TOKEN" \
+  -H "Content-Type: application/gzip" \
+  -H "Idempotency-Key: <key>" \
+  --data-binary @bundle.tar.gz
+```
+
+Safe fix: if the response includes `"retryable":true`, retry the same upload later with the same key. Otherwise, inspect the producer operation and use the same key only for the same source bundle.
+
+Reference: [HTTP Upload API Contract](integrations/http-upload.md#post-upload).
+
 ## Upload Status Is Missing

 Symptom: `GET /runs/<run_id>` returns `404`.
--- a/internal/app/upload_coordinator.go
+++ b/internal/app/upload_coordinator.go
@@ -13,6 +13,7 @@ import (

 	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+	sourcebundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
 )

 const DefaultUploadMaxFileCount = 4096
@@ -43,12 +44,13 @@ type UploadRunRecord struct {
 }

 type UploadRequest struct {
-	PipelineID   string
-	ContentType  string
-	Body         io.Reader
-	DryRun       bool
-	Force        bool
-	MaxFileCount int
+	PipelineID     string
+	ContentType    string
+	Body           io.Reader
+	IdempotencyKey string
+	DryRun         bool
+	Force          bool
+	MaxFileCount   int
 }

 type UploadQueueFullError struct {
@@ -64,6 +66,22 @@ func IsUploadQueueFull(err error) bool {
 	return errors.As(err, &full)
 }

+type UploadIdempotencyConflictError struct {
+	Retryable bool
+}
+
+func (err UploadIdempotencyConflictError) Error() string {
+	if err.Retryable {
+		return "upload idempotency key is already being processed"
+	}
+	return "upload idempotency key conflicts with a different source manifest"
+}
+
+func IsUploadIdempotencyConflict(err error) bool {
+	var conflict UploadIdempotencyConflictError
+	return errors.As(err, &conflict)
+}
+
 type UploadCoordinator struct {
 	ctx          context.Context
 	cfg          config.Config
@@ -82,6 +100,7 @@ type UploadCoordinator struct {
 	activePipeline map[string]bool
 	pending        []*uploadJob
 	records        map[UploadRunID]UploadRunRecord
+	idempotency    map[uploadIdempotencyScope]uploadIdempotencyRecord
 }

 type uploadStageFunc func(context.Context, ingest.StageOptions) (ingest.StagedBundle, error)
@@ -95,6 +114,17 @@ type uploadJob struct {
 	stagedRoot string
 }

+type uploadIdempotencyScope struct {
+	PipelineID string
+	Key        string
+}
+
+type uploadIdempotencyRecord struct {
+	RunID    UploadRunID
+	Manifest sourcebundle.Manifest
+	Pending  bool
+}
+
 type uploadCoordinatorHooks struct {
 	stage        uploadStageFunc
 	run          uploadRunFunc
@@ -140,6 +170,7 @@ func newUploadCoordinator(ctx context.Context, cfg config.Config, hooks uploadCo
 		maxConcurrency: cfg.Server.HTTP.MaxConcurrency,
 		activePipeline: map[string]bool{},
 		records:        map[UploadRunID]UploadRunRecord{},
+		idempotency:    map[uploadIdempotencyScope]uploadIdempotencyRecord{},
 	}
 	go coordinator.dispatchLoop()
 	return coordinator
@@ -169,14 +200,26 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 	if err := ingest.ValidateContentType(request.ContentType); err != nil {
 		return UploadRunRecord{}, err
 	}
+	scope, hasKey := uploadRequestIdempotencyScope(pipeline.ID, request.IdempotencyKey)

 	coordinator.mu.Lock()
 	coordinator.expireLocked(coordinator.now().UTC())
-	if coordinator.queueFullLocked() {
+	existingIdempotency, hasExistingIdempotency := coordinator.idempotency[scope]
+	if hasKey && hasExistingIdempotency && existingIdempotency.Pending {
 		coordinator.mu.Unlock()
-		return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
+		return UploadRunRecord{}, UploadIdempotencyConflictError{Retryable: true}
+	}
+	needsReservation := !hasKey || !hasExistingIdempotency
+	if needsReservation {
+		if coordinator.queueFullLocked() {
+			coordinator.mu.Unlock()
+			return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
+		}
+		coordinator.reservedCount++
+		if hasKey {
+			coordinator.idempotency[scope] = uploadIdempotencyRecord{Pending: true}
+		}
 	}
-	coordinator.reservedCount++
 	coordinator.mu.Unlock()

 	staged, err := coordinator.stage(ctx, ingest.StageOptions{
@@ -189,13 +232,36 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 		MaxFileCount:        uploadMaxFileCount(request.MaxFileCount),
 	})
 	if err != nil {
-		coordinator.releaseReservation()
+		if needsReservation {
+			coordinator.releaseReservation(scope, hasKey)
+		}
 		return UploadRunRecord{}, err
 	}

 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
-	coordinator.reservedCount--
+	if needsReservation {
+		coordinator.reservedCount--
+	}
+	if hasKey {
+		existingIdempotency, hasExistingIdempotency = coordinator.idempotency[scope]
+		if hasExistingIdempotency && !existingIdempotency.Pending {
+			if uploadManifestsEqual(existingIdempotency.Manifest, staged.Manifest) {
+				_ = os.RemoveAll(staged.Root)
+				record, ok := coordinator.records[existingIdempotency.RunID]
+				if !ok {
+					return UploadRunRecord{}, fmt.Errorf("idempotency record references missing run")
+				}
+				return record, nil
+			}
+			_ = os.RemoveAll(staged.Root)
+			return UploadRunRecord{}, UploadIdempotencyConflictError{}
+		}
+		if !hasExistingIdempotency && !needsReservation && coordinator.queueFullLocked() {
+			_ = os.RemoveAll(staged.Root)
+			return UploadRunRecord{}, UploadQueueFullError{QueueSize: coordinator.queueSize}
+		}
+	}
 	record := UploadRunRecord{
 		ID:         runID,
 		PipelineID: pipeline.ID,
@@ -204,6 +270,12 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 		StagedRoot: staged.Root,
 	}
 	coordinator.records[runID] = record
+	if hasKey {
+		coordinator.idempotency[scope] = uploadIdempotencyRecord{
+			RunID:    runID,
+			Manifest: staged.Manifest,
+		}
+	}
 	coordinator.pending = append(coordinator.pending, &uploadJob{
 		recordID:   runID,
 		request:    request,
@@ -214,6 +286,13 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload
 	return record, nil
 }

+func uploadRequestIdempotencyScope(pipelineID, key string) (uploadIdempotencyScope, bool) {
+	if key == "" {
+		return uploadIdempotencyScope{}, false
+	}
+	return uploadIdempotencyScope{PipelineID: pipelineID, Key: key}, true
+}
+
 func (coordinator *UploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) {
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
@@ -326,10 +405,15 @@ func (coordinator *UploadCoordinator) runJob(job *uploadJob) {
 	coordinator.complete(job, &report, err)
 }

-func (coordinator *UploadCoordinator) releaseReservation() {
+func (coordinator *UploadCoordinator) releaseReservation(scope uploadIdempotencyScope, hasKey bool) {
 	coordinator.mu.Lock()
 	defer coordinator.mu.Unlock()
 	coordinator.reservedCount--
+	if hasKey {
+		if record, ok := coordinator.idempotency[scope]; ok && record.Pending {
+			delete(coordinator.idempotency, scope)
+		}
+	}
 }

 func (coordinator *UploadCoordinator) queueFullLocked() bool {
@@ -379,10 +463,31 @@ func (coordinator *UploadCoordinator) expireLocked(now time.Time) []UploadRunRec
 		record.Error = ""
 		expired = append(expired, record)
 		delete(coordinator.records, runID)
+		for scope, idempotencyRecord := range coordinator.idempotency {
+			if idempotencyRecord.RunID == runID {
+				delete(coordinator.idempotency, scope)
+			}
+		}
 	}
 	return expired
 }

+func uploadManifestsEqual(a, b sourcebundle.Manifest) bool {
+	if a.SchemaVersion != b.SchemaVersion ||
+		a.ID != b.ID ||
+		a.Digest != b.Digest ||
+		!a.Created.Equal(b.Created) ||
+		len(a.Files) != len(b.Files) {
+		return false
+	}
+	for index := range a.Files {
+		if a.Files[index] != b.Files[index] {
+			return false
+		}
+	}
+	return true
+}
+
 func (coordinator *UploadCoordinator) notify() {
 	select {
 	case coordinator.signal <- struct{}{}:
--- a/internal/app/upload_coordinator_test.go
+++ b/internal/app/upload_coordinator_test.go
@@ -15,6 +15,7 @@ import (

 	"gitea.maximumdirect.net/eric/distributor/internal/config"
 	"gitea.maximumdirect.net/eric/distributor/internal/ingest"
+	sourcebundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
 )

 func TestUploadCoordinatorGeneratesRunIDAndAcceptedStatus(t *testing.T) {
@@ -254,6 +255,233 @@ func TestUploadCoordinatorExpiresCompletedRecordsAndStagingDirectories(t *testin
 	}
 }

+func TestUploadCoordinatorIdempotencyReturnsOriginalRunForSameManifest(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	var runCount atomic.Int64
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        manifestUploadStage,
+		run: func(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
+			runCount.Add(1)
+			return RunReport{}, nil
+		},
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "producer.retry:20260603",
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+
+	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "producer.retry:20260603",
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID != first.ID {
+		t.Fatalf("second run id = %q, want original %q", second.ID, first.ID)
+	}
+	if got := runCount.Load(); got != 1 {
+		t.Fatalf("run count = %d, want 1", got)
+	}
+}
+
+func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        manifestUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("one"),
+		IdempotencyKey: "same-key",
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+
+	_, err = coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("two"),
+		IdempotencyKey: "same-key",
+	})
+	if err == nil || !IsUploadIdempotencyConflict(err) {
+		t.Fatalf("second Submit() error = %v, want idempotency conflict", err)
+	}
+}
+
+func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports-one", "reports-two"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        manifestUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports-one",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("one"),
+		IdempotencyKey: "shared-key",
+	})
+	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("two"),
+		IdempotencyKey: "shared-key",
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID == first.ID {
+		t.Fatalf("run ids matched across pipelines: %q", second.ID)
+	}
+}
+
+func TestUploadCoordinatorWithoutIdempotencyKeyAcceptsDuplicateBodies(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage:        manifestUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same")})
+	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("same")})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID == first.ID {
+		t.Fatalf("second run id = %q, want distinct run", second.ID)
+	}
+}
+
+func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	coordinator := newUploadCoordinator(ctx, uploadCoordinatorConfig(t, uploadCoordinatorConfigOptions{
+		pipelineIDs: []string{"reports"},
+	}), uploadCoordinatorHooks{
+		randomSuffix: uploadTestSuffixes("00000001", "00000002"),
+		stage: func(ctx context.Context, opts ingest.StageOptions) (ingest.StagedBundle, error) {
+			close(entered)
+			<-release
+			return manifestUploadStage(ctx, opts)
+		},
+		run: successfulUploadRun,
+	})
+	firstErr := make(chan error, 1)
+	go func() {
+		_, err := coordinator.Submit(context.Background(), UploadRequest{
+			PipelineID:     "reports",
+			ContentType:    ingest.ContentTypeTar,
+			Body:           strings.NewReader("same"),
+			IdempotencyKey: "in-flight",
+		})
+		firstErr <- err
+	}()
+	<-entered
+
+	var reads atomic.Int64
+	_, 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
+		}),
+		IdempotencyKey: "in-flight",
+	})
+	var conflict UploadIdempotencyConflictError
+	if err == nil || !errors.As(err, &conflict) || !conflict.Retryable {
+		t.Fatalf("second Submit() error = %v, want retryable idempotency conflict", err)
+	}
+	if got := reads.Load(); got != 0 {
+		t.Fatalf("retryable conflict body reads = %d, want 0", got)
+	}
+	close(release)
+	if err := <-firstErr; err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+}
+
+func TestUploadCoordinatorIdempotencyExpiresWithCompletedStatus(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", "00000002", "00000003"),
+		stage:        manifestUploadStage,
+		run:          successfulUploadRun,
+	})
+
+	first, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "expires",
+	})
+	if err != nil {
+		t.Fatalf("first Submit() error = %v", err)
+	}
+	waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded)
+
+	clock.Advance(2 * time.Second)
+	coordinator.Expire()
+
+	second, err := coordinator.Submit(context.Background(), UploadRequest{
+		PipelineID:     "reports",
+		ContentType:    ingest.ContentTypeTar,
+		Body:           strings.NewReader("same"),
+		IdempotencyKey: "expires",
+	})
+	if err != nil {
+		t.Fatalf("second Submit() error = %v", err)
+	}
+	if second.ID == first.ID {
+		t.Fatalf("second run id = %q, want new run after expiry", second.ID)
+	}
+}
+
 type readerFunc func([]byte) (int, error)

 func (fn readerFunc) Read(data []byte) (int, error) {
@@ -268,6 +496,37 @@ func successfulUploadStage(ctx context.Context, opts ingest.StageOptions) (inges
 	return ingest.StagedBundle{Root: root}, nil
 }

+func manifestUploadStage(ctx context.Context, opts ingest.StageOptions) (ingest.StagedBundle, error) {
+	data, err := io.ReadAll(opts.Body)
+	if err != nil {
+		return ingest.StagedBundle{}, err
+	}
+	root := filepath.Join(opts.PipelineStagingPath, opts.RunID)
+	if err := os.MkdirAll(root, 0o755); err != nil {
+		return ingest.StagedBundle{}, err
+	}
+	return ingest.StagedBundle{
+		Root:     root,
+		Manifest: uploadTestManifest(string(data)),
+	}, nil
+}
+
+func uploadTestManifest(id string) sourcebundle.Manifest {
+	created := time.Date(2026, 6, 3, 12, 0, 0, 0, time.UTC)
+	file := sourcebundle.ManifestFile{
+		Path:   "report.md",
+		SHA256: sourcebundle.FileDigest([]byte(id)),
+		Size:   int64(len(id)),
+	}
+	return sourcebundle.Manifest{
+		SchemaVersion: sourcebundle.SchemaVersion,
+		ID:            id,
+		Created:       created,
+		Files:         []sourcebundle.ManifestFile{file},
+		Digest:        sourcebundle.BundleDigest([]sourcebundle.ManifestFile{file}),
+	}
+}
+
 func successfulUploadRun(ctx context.Context, cfg config.Config, options RunPipelineWithLocalSourceOptions) (RunReport, error) {
 	return RunReport{}, nil
 }
--- a/internal/app/upload_http.go
+++ b/internal/app/upload_http.go
@@ -29,9 +29,12 @@ type uploadAcceptedResponse struct {
 }

 type httpErrorResponse struct {
-	Error string `json:"error"`
+	Error     string `json:"error"`
+	Retryable bool   `json:"retryable,omitempty"`
 }

+const idempotencyKeyHeader = "Idempotency-Key"
+
 func newUploadHTTPHandler(ctx context.Context, cfg config.Config, environment config.Environment) (http.Handler, error) {
 	config.ApplyDefaults(&cfg)
 	tokens, err := resolveUploadTokens(cfg, environment)
@@ -98,14 +101,16 @@ func (handler uploadHTTPHandler) handleUpload(w http.ResponseWriter, r *http.Req
 		writeHTTPError(w, http.StatusUnsupportedMediaType, "unsupported content type")
 		return
 	}
-	if !handler.coordinator.CanAccept() {
-		writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
+	idempotencyKey, err := uploadIdempotencyKey(r.Header)
+	if err != nil {
+		writeHTTPError(w, http.StatusBadRequest, "invalid idempotency key")
 		return
 	}
 	record, err := handler.coordinator.Submit(r.Context(), UploadRequest{
-		PipelineID:  pipelineID,
-		ContentType: contentType,
-		Body:        r.Body,
+		PipelineID:     pipelineID,
+		ContentType:    contentType,
+		Body:           r.Body,
+		IdempotencyKey: idempotencyKey,
 	})
 	if err != nil {
 		writeUploadSubmitError(w, err)
@@ -144,10 +149,44 @@ func (handler uploadHTTPHandler) authenticate(header string) (string, bool) {
 	return pipelineID, ok
 }

+func uploadIdempotencyKey(header http.Header) (string, error) {
+	values := header.Values(idempotencyKeyHeader)
+	if len(values) == 0 {
+		return "", nil
+	}
+	if len(values) != 1 {
+		return "", fmt.Errorf("idempotency key must appear at most once")
+	}
+	key := values[0]
+	if key == "" {
+		return "", fmt.Errorf("idempotency key is required when header is present")
+	}
+	if len(key) > 128 {
+		return "", fmt.Errorf("idempotency key must be at most 128 bytes")
+	}
+	for index := 0; index < len(key); index++ {
+		character := key[index]
+		if character >= 'a' && character <= 'z' ||
+			character >= 'A' && character <= 'Z' ||
+			character >= '0' && character <= '9' ||
+			character == '.' ||
+			character == '_' ||
+			character == '-' ||
+			character == ':' {
+			continue
+		}
+		return "", fmt.Errorf("idempotency key contains unsupported character")
+	}
+	return key, nil
+}
+
 func writeUploadSubmitError(w http.ResponseWriter, err error) {
+	var idempotencyConflict UploadIdempotencyConflictError
 	switch {
 	case IsUploadQueueFull(err):
 		writeHTTPError(w, http.StatusServiceUnavailable, "upload queue is full")
+	case errors.As(err, &idempotencyConflict):
+		writeHTTPErrorRetryable(w, http.StatusConflict, idempotencyConflict.Error(), idempotencyConflict.Retryable)
 	case errors.Is(err, ingest.ErrUploadTooLarge):
 		writeHTTPError(w, http.StatusRequestEntityTooLarge, "upload exceeds maximum size")
 	case errors.Is(err, ingest.ErrUnsupportedContentType):
@@ -158,7 +197,11 @@ func writeUploadSubmitError(w http.ResponseWriter, err error) {
 }

 func writeHTTPError(w http.ResponseWriter, status int, message string) {
-	writeJSON(w, status, httpErrorResponse{Error: message})
+	writeHTTPErrorRetryable(w, status, message, false)
+}
+
+func writeHTTPErrorRetryable(w http.ResponseWriter, status int, message string, retryable bool) {
+	writeJSON(w, status, httpErrorResponse{Error: message, Retryable: retryable})
 }

 func writeJSON(w http.ResponseWriter, status int, value any) {
--- a/internal/app/upload_http_integration_test.go
+++ b/internal/app/upload_http_integration_test.go
@@ -97,6 +97,62 @@ func TestHTTPUploadInvalidArchiveIsRejectedWithoutRunID(t *testing.T) {
 	assertDirectoryEmpty(t, destination)
 }

+func TestHTTPUploadIdempotencyReturnsOriginalRunForSameBundle(t *testing.T) {
+	destination := t.TempDir()
+	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "REPORTS_TOKEN",
+		stagingPath:  filepath.Join(t.TempDir(), "reports"),
+		destinations: []string{destination},
+	}}, 4, 1))
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens:      map[string]string{"reports-secret": "reports"},
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{}))
+	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+
+	secondRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeGzip, "same-key", bundleArchive(t, true, testutil.BundleOptions{}))
+	if secondRunID != firstRunID {
+		t.Fatalf("second run id = %q, want original %q", secondRunID, firstRunID)
+	}
+	if got := coordinator.QueueDepth(); got != 0 {
+		t.Fatalf("queue depth = %d, want no duplicate run queued", got)
+	}
+}
+
+func TestHTTPUploadIdempotencyReturnsConflictForDifferentBundle(t *testing.T) {
+	destination := t.TempDir()
+	coordinator := NewUploadCoordinator(context.Background(), httpUploadIntegrationConfig(t, []httpUploadPipelineSpec{{
+		id:           "reports",
+		tokenEnv:     "REPORTS_TOKEN",
+		stagingPath:  filepath.Join(t.TempDir(), "reports"),
+		destinations: []string{destination},
+	}}, 4, 1))
+	handler := uploadHTTPHandler{
+		coordinator: coordinator,
+		tokens:      map[string]string{"reports-secret": "reports"},
+	}
+	server := httptest.NewServer(handler)
+	defer server.Close()
+
+	firstRunID := submitHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{}))
+	waitForHTTPUploadStatus(t, server, firstRunID, UploadStatusSucceeded)
+
+	status, body := postHTTPUploadWithKey(t, server, "reports-secret", ingest.ContentTypeTar, "same-key", bundleArchive(t, false, testutil.BundleOptions{
+		ID: "weather.daily.brentwood.2026-05-31",
+	}))
+	if status != http.StatusConflict {
+		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusConflict, body)
+	}
+	if strings.Contains(body, "reports-secret") {
+		t.Fatalf("conflict response exposed token: %s", body)
+	}
+}
+
 func TestHTTPUploadOversizedArchiveIsRejectedWithoutRunID(t *testing.T) {
 	destination := t.TempDir()
 	stagingPath := filepath.Join(t.TempDir(), "reports")
@@ -278,6 +334,17 @@ func httpUploadIntegrationConfig(t *testing.T, pipelines []httpUploadPipelineSpe
 func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) UploadRunID {
 	t.Helper()
 	status, responseBody := postHTTPUpload(t, server, token, contentType, body)
+	return decodeAcceptedHTTPUpload(t, status, responseBody)
+}
+
+func submitHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, contentType, key string, body []byte) UploadRunID {
+	t.Helper()
+	status, responseBody := postHTTPUploadWithKey(t, server, token, contentType, key, body)
+	return decodeAcceptedHTTPUpload(t, status, responseBody)
+}
+
+func decodeAcceptedHTTPUpload(t *testing.T, status int, responseBody string) UploadRunID {
+	t.Helper()
 	if status != http.StatusAccepted {
 		t.Fatalf("POST /upload status = %d, want %d; body = %s", status, http.StatusAccepted, responseBody)
 	}
@@ -292,6 +359,11 @@ func submitHTTPUpload(t *testing.T, server *httptest.Server, token, contentType
 }

 func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType string, body []byte) (int, string) {
+	t.Helper()
+	return postHTTPUploadWithKey(t, server, token, contentType, "", body)
+}
+
+func postHTTPUploadWithKey(t *testing.T, server *httptest.Server, token, contentType, key string, body []byte) (int, string) {
 	t.Helper()
 	request, err := http.NewRequest(http.MethodPost, server.URL+"/upload", bytes.NewReader(body))
 	if err != nil {
@@ -299,6 +371,9 @@ func postHTTPUpload(t *testing.T, server *httptest.Server, token, contentType st
 	}
 	request.Header.Set("Authorization", "Bearer "+token)
 	request.Header.Set("Content-Type", contentType)
+	if key != "" {
+		request.Header.Set("Idempotency-Key", key)
+	}
 	response, err := server.Client().Do(request)
 	if err != nil {
 		t.Fatalf("POST /upload error = %v", err)
--- a/internal/app/upload_http_test.go
+++ b/internal/app/upload_http_test.go
@@ -97,7 +97,6 @@ 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)
@@ -116,6 +115,7 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 	request := httptest.NewRequest(http.MethodPost, "/upload", strings.NewReader("archive"))
 	request.Header.Set("Authorization", "Bearer valid-token")
 	request.Header.Set("Content-Type", "application/x-tar")
+	request.Header.Set("Idempotency-Key", "producer.retry:20260603")

 	handler.ServeHTTP(recorder, request)

@@ -125,6 +125,9 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {
 	if submitted.PipelineID != "reports" {
 		t.Fatalf("submitted pipeline = %q, want reports", submitted.PipelineID)
 	}
+	if submitted.IdempotencyKey != "producer.retry:20260603" {
+		t.Fatalf("submitted idempotency key = %q, want producer.retry:20260603", submitted.IdempotencyKey)
+	}
 	var response uploadAcceptedResponse
 	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
 		t.Fatalf("decode response: %v", err)
@@ -139,7 +142,7 @@ func TestUploadHTTPHandlerAuthenticatesAndAcceptsUpload(t *testing.T) {

 func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
 	handler := uploadHTTPHandler{
-		coordinator: fakeUploadCoordinator{canAccept: true},
+		coordinator: fakeUploadCoordinator{},
 		tokens:      map[string]string{"valid-token": "reports"},
 	}

@@ -160,34 +163,56 @@ func TestUploadHTTPHandlerRejectsUnauthorizedRequests(t *testing.T) {
 	}
 }

-func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *testing.T) {
+func TestUploadHTTPHandlerRejectsUnsupportedContentTypeInvalidKeyAndPipelineID(t *testing.T) {
 	tests := []struct {
 		name        string
-		canAccept   bool
 		url         string
 		contentType string
+		keyValues   []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:        "full queue",
-			canAccept:   false,
+			name:        "invalid key syntax",
 			url:         "/upload",
 			contentType: "application/x-tar",
-			body:        &countingReader{reader: strings.NewReader("archive")},
-			wantStatus:  http.StatusServiceUnavailable,
+			keyValues:   []string{"bad key"},
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
+		},
+		{
+			name:        "empty key",
+			url:         "/upload",
+			contentType: "application/x-tar",
+			keyValues:   []string{""},
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
+		},
+		{
+			name:        "too long key",
+			url:         "/upload",
+			contentType: "application/x-tar",
+			keyValues:   []string{strings.Repeat("a", 129)},
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
+		},
+		{
+			name:        "multiple keys",
+			url:         "/upload",
+			contentType: "application/x-tar",
+			keyValues:   []string{"one", "two"},
+			body:        strings.NewReader("archive"),
+			wantStatus:  http.StatusBadRequest,
 		},
 		{
 			name:        "submitted pipeline id",
-			canAccept:   true,
 			url:         "/upload?pipeline_id=reports",
 			contentType: "application/x-tar",
 			body:        strings.NewReader("archive"),
@@ -198,7 +223,6 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
 		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
@@ -210,15 +234,15 @@ func TestUploadHTTPHandlerRejectsUnsupportedOversizedFullQueueAndPipelineID(t *t
 			request := httptest.NewRequest(http.MethodPost, tt.url, tt.body)
 			request.Header.Set("Authorization", "Bearer valid-token")
 			request.Header.Set("Content-Type", tt.contentType)
+			for _, value := range tt.keyValues {
+				request.Header.Add("Idempotency-Key", value)
+			}

 			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)
-			}
 		})
 	}
 }
@@ -228,9 +252,13 @@ func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
 		name       string
 		err        error
 		wantStatus int
+		wantBody   string
 	}{
 		{name: "oversized", err: ingest.ErrUploadTooLarge, wantStatus: http.StatusRequestEntityTooLarge},
 		{name: "unsupported", err: ingest.ErrUnsupportedContentType, wantStatus: http.StatusUnsupportedMediaType},
+		{name: "full queue", err: UploadQueueFullError{QueueSize: 1}, wantStatus: http.StatusServiceUnavailable},
+		{name: "idempotency conflict", err: UploadIdempotencyConflictError{}, wantStatus: http.StatusConflict, wantBody: "different source manifest"},
+		{name: "idempotency in progress", err: UploadIdempotencyConflictError{Retryable: true}, wantStatus: http.StatusConflict, wantBody: `"retryable":true`},
 		{name: "malformed", err: errors.New("malformed archive"), wantStatus: http.StatusBadRequest},
 	}
 	for _, tt := range tests {
@@ -254,6 +282,9 @@ func TestUploadHTTPHandlerMapsSubmitErrors(t *testing.T) {
 			if recorder.Code != tt.wantStatus {
 				t.Fatalf("status = %d, want %d; body = %q", recorder.Code, tt.wantStatus, recorder.Body.String())
 			}
+			if tt.wantBody != "" && !strings.Contains(recorder.Body.String(), tt.wantBody) {
+				t.Fatalf("body = %q, want substring %q", recorder.Body.String(), tt.wantBody)
+			}
 		})
 	}
 }