From 1c5d7198e32fe74ad629049b8621429abbda7e01 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 8 Jun 2026 04:34:29 +0000 Subject: [PATCH] Scope upload idempotency by token --- internal/app/upload_coordinator.go | 7 +- internal/app/upload_coordinator_test.go | 97 +++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 3 deletions(-) diff --git a/internal/app/upload_coordinator.go b/internal/app/upload_coordinator.go index 6d90832..0ea24c3 100644 --- a/internal/app/upload_coordinator.go +++ b/internal/app/upload_coordinator.go @@ -116,6 +116,7 @@ type uploadJob struct { } type uploadIdempotencyScope struct { + TokenID string PipelineID string Key string } @@ -201,7 +202,7 @@ 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) + scope, hasKey := uploadRequestIdempotencyScope(request.TokenID, pipeline.ID, request.IdempotencyKey) coordinator.mu.Lock() coordinator.expireLocked(coordinator.now().UTC()) @@ -287,11 +288,11 @@ func (coordinator *UploadCoordinator) Submit(ctx context.Context, request Upload return record, nil } -func uploadRequestIdempotencyScope(pipelineID, key string) (uploadIdempotencyScope, bool) { +func uploadRequestIdempotencyScope(tokenID, pipelineID, key string) (uploadIdempotencyScope, bool) { if key == "" { return uploadIdempotencyScope{}, false } - return uploadIdempotencyScope{PipelineID: pipelineID, Key: key}, true + return uploadIdempotencyScope{TokenID: tokenID, PipelineID: pipelineID, Key: key}, true } func (coordinator *UploadCoordinator) Status(runID UploadRunID) (UploadRunRecord, bool) { diff --git a/internal/app/upload_coordinator_test.go b/internal/app/upload_coordinator_test.go index 6060616..05cd24c 100644 --- a/internal/app/upload_coordinator_test.go +++ b/internal/app/upload_coordinator_test.go @@ -271,6 +271,7 @@ func TestUploadCoordinatorIdempotencyReturnsOriginalRunForSameManifest(t *testin }) first, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same"), @@ -282,6 +283,7 @@ func TestUploadCoordinatorIdempotencyReturnsOriginalRunForSameManifest(t *testin waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded) second, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same"), @@ -310,6 +312,7 @@ func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(t *testing.T) }) first, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("one"), @@ -321,6 +324,7 @@ func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(t *testing.T) waitForUploadStatus(t, coordinator, first.ID, UploadStatusSucceeded) _, err = coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("two"), @@ -331,6 +335,42 @@ func TestUploadCoordinatorIdempotencyConflictsForDifferentManifest(t *testing.T) } } +func TestUploadCoordinatorIdempotencyIsScopedByToken(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{ + TokenID: "reporter-a", + PipelineID: "reports", + 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{ + TokenID: "reporter-b", + PipelineID: "reports", + 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 tokens: %q", second.ID) + } +} + func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -343,6 +383,7 @@ func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) { }) first, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports-one", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("one"), @@ -352,6 +393,7 @@ func TestUploadCoordinatorIdempotencyIsScopedByPipeline(t *testing.T) { t.Fatalf("first Submit() error = %v", err) } second, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports-two", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("two"), @@ -408,6 +450,7 @@ func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *tes firstErr := make(chan error, 1) go func() { _, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same"), @@ -419,6 +462,7 @@ func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *tes var reads atomic.Int64 _, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: readerFunc(func(data []byte) (int, error) { @@ -440,6 +484,57 @@ func TestUploadCoordinatorIdempotencyReturnsRetryableConflictWhileStaging(t *tes } } +func TestUploadCoordinatorIdempotencyPendingScopeIncludesToken(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var calls atomic.Int64 + 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) { + if calls.Add(1) == 1 { + close(entered) + <-release + } + return manifestUploadStage(ctx, opts) + }, + run: successfulUploadRun, + }) + firstErr := make(chan error, 1) + go func() { + _, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", + PipelineID: "reports", + ContentType: ingest.ContentTypeTar, + Body: strings.NewReader("same"), + IdempotencyKey: "in-flight", + }) + firstErr <- err + }() + <-entered + + second, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-b", + PipelineID: "reports", + ContentType: ingest.ContentTypeTar, + Body: strings.NewReader("same"), + IdempotencyKey: "in-flight", + }) + if err != nil { + t.Fatalf("second Submit() error = %v", err) + } + if second.ID == "" { + t.Fatal("second run id is empty, want accepted run") + } + 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() @@ -455,6 +550,7 @@ func TestUploadCoordinatorIdempotencyExpiresWithCompletedStatus(t *testing.T) { }) first, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same"), @@ -469,6 +565,7 @@ func TestUploadCoordinatorIdempotencyExpiresWithCompletedStatus(t *testing.T) { coordinator.Expire() second, err := coordinator.Submit(context.Background(), UploadRequest{ + TokenID: "reporter-a", PipelineID: "reports", ContentType: ingest.ContentTypeTar, Body: strings.NewReader("same"),