diff --git a/pkg/appview/storage/proxy_blob_store.go b/pkg/appview/storage/proxy_blob_store.go index c1e3c12..2eee1e6 100644 --- a/pkg/appview/storage/proxy_blob_store.go +++ b/pkg/appview/storage/proxy_blob_store.go @@ -16,6 +16,7 @@ import ( "atcr.io/pkg/atproto" "github.com/distribution/distribution/v3" "github.com/distribution/distribution/v3/registry/api/errcode" + "github.com/google/uuid" "github.com/opencontainers/go-digest" ) @@ -42,6 +43,25 @@ var ( globalUploadsMu sync.RWMutex ) +// newWriterID names a blob upload. A variable so tests can force a collision +// against the guard in Create. +// +// This used to be fmt.Sprintf("upload-%d", time.Now().UnixNano()), which is not +// unique: Go's wall clock is coarser than the spacing between goroutines, so +// two uploads opened at the same instant read the same nanosecond often enough +// to matter (roughly one release in five, measured on a tsc clocksource box). +// Docker and crane POST the config blob and several layers concurrently, so +// every multi blob push rolled those dice. The loser's writer replaced the +// winner's in globalUploads, both clients' PATCHes then resumed the same +// writer, and distribution rejected the second with a 416 "upload resumed at +// wrong offset", which the client reports as RANGE_INVALID. +// +// The UUID is hex and hyphens, so the ID stays URL safe: it travels in the +// upload URL and inside distribution's _state token. +var newWriterID = func() string { + return "upload-" + uuid.NewString() +} + // The transport and client below are package-level on purpose. RoutingRepository // (and therefore ProxyBlobStore) is built fresh on every registry request, so a // per-instance transport was thrown away after a single request: nothing was ever @@ -378,8 +398,7 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo } } - // Generate unique writer ID - writerID := fmt.Sprintf("upload-%d", time.Now().UnixNano()) + writerID := newWriterID() now := time.Now() writer := &ProxyBlobWriter{ @@ -395,8 +414,17 @@ func (p *ProxyBlobStore) Create(ctx context.Context, options ...distribution.Blo requestCtx: ctx, } - // Store in global uploads map for resume support + // Store in global uploads map for resume support. + // + // Refuse to overwrite: with a UUID an occupied key cannot be chance, so + // something is badly wrong and silently dropping the sitting writer would + // be worse than failing this Create. The sitting writer keeps its buffer + // budget and its hold side multipart, and its own client keeps resuming it. globalUploadsMu.Lock() + if _, taken := globalUploads[writer.id]; taken { + globalUploadsMu.Unlock() + return nil, fmt.Errorf("upload id %s is already in use", writer.id) + } globalUploads[writer.id] = writer globalUploadsMu.Unlock() diff --git a/pkg/appview/storage/proxy_blob_store_test.go b/pkg/appview/storage/proxy_blob_store_test.go index aa74bff..1cb9449 100644 --- a/pkg/appview/storage/proxy_blob_store_test.go +++ b/pkg/appview/storage/proxy_blob_store_test.go @@ -21,6 +21,7 @@ import ( "atcr.io/pkg/auth" "github.com/distribution/distribution/v3" "github.com/distribution/distribution/v3/registry/api/errcode" + "github.com/google/uuid" "github.com/opencontainers/go-digest" ) @@ -779,7 +780,7 @@ func (m *mockS3Server) part(n int) []byte { // newMockHoldServer creates a mock hold service func newMockHoldServer(t *testing.T, s3URL string) *mockHoldServer { m := &mockHoldServer{ - UploadID: "test-upload-id-" + fmt.Sprintf("%d", time.Now().UnixNano()), + UploadID: "test-upload-id-" + uuid.NewString(), } m.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { @@ -3066,3 +3067,139 @@ func TestWrite_NeverOvershootsTheThreshold(t *testing.T) { }) } } + +// TestCreate_ConcurrentIDsAreUnique pins that two uploads opened at the same +// instant never get the same ID. The old nanosecond clock ID collided here: +// Docker and crane POST the config blob and several layers together, both +// writers landed on one key in globalUploads, and the client that lost the race +// got a 416 from the resume that followed. +// +// Note this test does not reliably fail against the old code. The collision +// needs true simultaneity and the mock path desynchronises the goroutines, so +// the overwrite guard itself is covered separately by +// TestCreate_RefusesDuplicateID. +func TestCreate_ConcurrentIDsAreUnique(t *testing.T) { + isolateUploads(t) + + s3Server := newMockS3Server(t, true) + defer s3Server.Close() + + holdServer := newMockHoldServer(t, s3Server.URL) + defer holdServer.Close() + + store := createTestProxyBlobStore(t, holdServer.URL) + + const ( + writers = 3 + rounds = 300 + ) + + var mu sync.Mutex + seen := make(map[string]int, writers*rounds) + created := make([]distribution.BlobWriter, 0, writers*rounds) + + for range rounds { + // One barrier per round, so all three Creates are released together + // rather than trickling past each other. + var release, finished sync.WaitGroup + release.Add(1) + finished.Add(writers) + + for range writers { + go func() { + defer finished.Done() + release.Wait() + + w, err := store.Create(context.Background()) + if err != nil { + t.Errorf("Create() failed: %v", err) + return + } + + mu.Lock() + seen[w.ID()]++ + created = append(created, w) + mu.Unlock() + }() + } + + release.Done() + finished.Wait() + } + + t.Cleanup(func() { + for _, w := range created { + w.Cancel(context.Background()) + } + }) + + if len(created) != writers*rounds { + t.Fatalf("Expected %d writers, got %d", writers*rounds, len(created)) + } + + for id, n := range seen { + if n != 1 { + t.Errorf("ID %s handed out %d times, expected once", id, n) + } + } + if len(seen) != writers*rounds { + t.Errorf("Expected %d distinct IDs, got %d", writers*rounds, len(seen)) + } + + globalUploadsMu.RLock() + tracked := len(globalUploads) + globalUploadsMu.RUnlock() + if tracked != writers*rounds { + t.Errorf("Expected %d writers tracked in globalUploads, got %d", writers*rounds, tracked) + } +} + +// TestCreate_RefusesDuplicateID covers the overwrite guard directly, by forcing +// newWriterID to hand out an ID that is already taken. With a UUID that cannot +// happen by chance, so Create must fail loudly and leave the sitting writer +// alone rather than evict it and strand its client on a 416. +func TestCreate_RefusesDuplicateID(t *testing.T) { + isolateUploads(t) + + s3Server := newMockS3Server(t, true) + defer s3Server.Close() + + holdServer := newMockHoldServer(t, s3Server.URL) + defer holdServer.Close() + + store := createTestProxyBlobStore(t, holdServer.URL) + + first, err := store.Create(context.Background()) + if err != nil { + t.Fatalf("Create() failed: %v", err) + } + defer first.Cancel(context.Background()) + + prev := newWriterID + newWriterID = func() string { return first.ID() } + t.Cleanup(func() { newWriterID = prev }) + + second, err := store.Create(context.Background()) + if err == nil { + second.Cancel(context.Background()) + t.Fatal("Expected Create() to refuse an ID that is already in use") + } + if second != nil { + t.Errorf("Expected nil writer alongside the error, got %v", second) + } + if !strings.Contains(err.Error(), "already in use") { + t.Errorf("Expected an 'already in use' error, got %v", err) + } + + globalUploadsMu.RLock() + sitting := globalUploads[first.ID()] + tracked := len(globalUploads) + globalUploadsMu.RUnlock() + + if sitting != first { + t.Error("The original writer should still be the one in globalUploads") + } + if tracked != 1 { + t.Errorf("Expected 1 writer tracked in globalUploads, got %d", tracked) + } +}