From 13a1a28fc1e1f6f79fe065bbe8d98e73fe66e94e Mon Sep 17 00:00:00 2001 From: chruffins <23645059+chruffins@users.noreply.github.com> Date: Mon, 10 Aug 2026 19:40:11 +0000 Subject: [PATCH] Address pushes API review: push timeout, drop dead wait surface, simplify dedup --- cmd/api/api/pushes.go | 4 +- cmd/api/api/pushes_test.go | 3 - lib/imagepush/imagepush.go | 43 +++-- lib/imagepush/manager.go | 277 +++++++++---------------------- lib/imagepush/manager_test.go | 301 +++++++++------------------------- lib/imagepush/storage.go | 9 + 6 files changed, 196 insertions(+), 441 deletions(-) diff --git a/cmd/api/api/pushes.go b/cmd/api/api/pushes.go index 1a080369..1b813a85 100644 --- a/cmd/api/api/pushes.go +++ b/cmd/api/api/pushes.go @@ -134,11 +134,9 @@ func pushToOAPI(push imagepush.Push) oapi.Push { CreatedAt: push.CreatedAt, CompletedAt: push.CompletedAt, } - if push.Layers > 0 { + if push.Status == oapi.PushStatus(imagepush.StatusPushed) { layers := push.Layers out.Layers = &layers - } - if push.Bytes > 0 { bytes := push.Bytes out.Bytes = &bytes } diff --git a/cmd/api/api/pushes_test.go b/cmd/api/api/pushes_test.go index ec4daca8..bdfcb93f 100644 --- a/cmd/api/api/pushes_test.go +++ b/cmd/api/api/pushes_test.go @@ -45,9 +45,6 @@ func (f *fakePushManager) ListPushes(_ context.Context) ([]imagepush.Push, error return f.pushes, nil } -func (f *fakePushManager) WaitForPush(_ context.Context, _ string) error { return nil } - -func (f *fakePushManager) InProgressDigests() []string { return nil } func (f *fakePushManager) LiveCacheManifestDigests() []string { return nil } func TestCreatePush_MapsRequestAndCredentials(t *testing.T) { diff --git a/lib/imagepush/imagepush.go b/lib/imagepush/imagepush.go index 236602d5..dbcc2cfb 100644 --- a/lib/imagepush/imagepush.go +++ b/lib/imagepush/imagepush.go @@ -8,11 +8,13 @@ package imagepush import ( + "bytes" "context" "crypto/sha256" + "encoding/base64" "encoding/hex" "errors" - "strings" + "fmt" "time" "github.com/google/go-containerregistry/pkg/authn" @@ -24,6 +26,11 @@ const ( StatusPushing = "pushing" StatusPushed = "pushed" StatusFailed = "failed" + + // pushTimeout bounds a single registry export so a wedged registry cannot + // pin a queue slot forever; with a bounded concurrency pool one hung push + // would otherwise block every later push. + pushTimeout = 30 * time.Minute ) var ( @@ -71,7 +78,22 @@ func credFingerprint(c *authn.AuthConfig) string { if !credsPresent(c) { return "" } - sum := sha256.Sum256([]byte(strings.Join([]string{c.Username, c.Password, c.Auth, c.IdentityToken, c.RegistryToken}, "\x00"))) + // Normalize the precomputed base64 "user:pass" Auth shorthand into its + // username/password parts so the same login supplied either way hashes + // identically (AuthConfig.UnmarshalJSON already expands it, but a config + // built in code may not have gone through JSON). + username, password := c.Username, c.Password + if c.Auth != "" { + if decoded, err := base64.StdEncoding.DecodeString(c.Auth); err == nil { + if i := bytes.IndexByte(decoded, ':'); i >= 0 { + username, password = string(decoded[:i]), string(decoded[i+1:]) + } + } + } + // IdentityToken and RegistryToken are distinct auth modes (token/registry- + // scoped) and are kept as-is; they can carry a different identity than the + // basic-auth pair. + sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%s\x00%s", username, password, c.IdentityToken, c.RegistryToken))) return hex.EncodeToString(sum[:]) } @@ -95,12 +117,6 @@ type ImageResolver interface { GetImage(ctx context.Context, name string) (*images.Image, error) } -// StatusEvent represents a terminal status change for push notifications. -type StatusEvent struct { - Status string - Err error -} - // Manager orchestrates push jobs. type Manager interface { // CreatePush validates the request, persists a queued job, and enqueues it. @@ -113,13 +129,8 @@ type Manager interface { GetPush(ctx context.Context, id string) (*Push, error) // ListPushes returns all pushes, newest first. ListPushes(ctx context.Context) ([]Push, error) - // WaitForPush blocks until the push reaches a terminal state (pushed or - // failed) or the context is cancelled. - WaitForPush(ctx context.Context, id string) error - // InProgressDigests returns the manifest digests of queued and pushing - // jobs so the OCI cache GC can keep their blobs alive mid-push. - InProgressDigests() []string - // LiveCacheManifestDigests implements ocicachegc.RootsProvider by - // delegating to InProgressDigests. + // LiveCacheManifestDigests implements ocicachegc.RootsProvider: the + // manifest digests of queued and pushing jobs, so the OCI cache GC keeps + // their blobs alive mid-push. LiveCacheManifestDigests() []string } diff --git a/lib/imagepush/manager.go b/lib/imagepush/manager.go index 5f8c56cd..95f38662 100644 --- a/lib/imagepush/manager.go +++ b/lib/imagepush/manager.go @@ -32,9 +32,6 @@ type manager struct { mu sync.Mutex inflight map[string]inflightPush // key = pushKey(digest, target, insecure) - - subscriberMu sync.RWMutex - subscribers map[string][]chan StatusEvent // keyed by push ID } // NewManager creates a push manager. provider may be nil, in which case @@ -49,12 +46,11 @@ func NewManager(p *paths.Paths, resolver ImageResolver, provider registrypush.Pr } m := &manager{ - paths: p, - resolver: resolver, - provider: provider, - queue: queue.New(maxConcurrent), - inflight: make(map[string]inflightPush), - subscribers: make(map[string][]chan StatusEvent), + paths: p, + resolver: resolver, + provider: provider, + queue: queue.New(maxConcurrent), + inflight: make(map[string]inflightPush), } m.recoverInterruptedPushes() @@ -114,63 +110,56 @@ func (m *manager) CreatePush(ctx context.Context, req PushRequest) (*Push, error // a concurrent request for the same digest+target cannot slip in between // and create a duplicate job. The write is one small fsync'd file; keeping // it under the lock is what lets the dedup path hand back a durable record, - // and it only briefly stalls InProgressDigests — cheap next to the registry - // I/O that dominates a push. + // and it only briefly stalls the GC live-digest read — cheap next to the + // registry I/O that dominates a push. var meta *pushMetadata - for { - m.mu.Lock() - if existing, ok := m.inflight[key]; ok { - // Merge only when the in-flight job runs under the same credentials - // as the request. The manager never stores credential values, so it - // compares fingerprints: a request that borrowed credentials cannot - // merge into an anonymous in-flight push (its auth would be silently - // dropped), an anonymous request cannot merge into a credentialed one - // (it would silently inherit another caller's login), and two - // requests that borrowed different logins cannot merge either — one - // would run under the other caller's auth, and an instance can serve - // more than one principal. Surface the conflict instead so the caller - // can retry once the in-flight job completes or match its credentials. - if existing.credFingerprint != fingerprint { - m.mu.Unlock() - return nil, fmt.Errorf("%w: a push of %s to %s is already in flight with different credentials; retry once it completes or match its credentials", ErrCredentialConflict, img.Digest, dstRef.String()) - } - id := existing.id + m.mu.Lock() + if existing, ok := m.inflight[key]; ok { + // Merge only when the in-flight job runs under the same credentials + // as the request. The manager never stores credential values, so it + // compares fingerprints: a request that borrowed credentials cannot + // merge into an anonymous in-flight push (its auth would be silently + // dropped), an anonymous request cannot merge into a credentialed one + // (it would silently inherit another caller's login), and two + // requests that borrowed different logins cannot merge either — one + // would run under the other caller's auth, and an instance can serve + // more than one principal. Surface the conflict instead so the caller + // can retry once the in-flight job completes or match its credentials. + if existing.credFingerprint != fingerprint { m.mu.Unlock() - push, err := m.GetPush(ctx, id) - if errors.Is(err, ErrNotFound) { - // The job's terminal record could not be persisted and its - // directory was dropped; the queue completion hook releases the - // inflight entry moments later. Wait for that entry (not just the - // key) to go away, then retry the dedup: a concurrent waiter that - // got here first may already have registered a successor, which - // this retry merges into instead of surfacing a bare ErrNotFound - // from a create call. - if err := m.waitForInflightRelease(ctx, key, id); err != nil { - return nil, err - } - continue - } - return push, err - } - - meta = &pushMetadata{ - ID: cuid2.Generate(), - Status: StatusQueued, - Image: img.Name, - Digest: img.Digest, - Target: dstRef.String(), - Insecure: req.Insecure, - HadCredentials: credsPresent(req.Credentials), - CreatedAt: time.Now(), + return nil, fmt.Errorf("%w: a push of %s to %s is already in flight with different credentials; retry once it completes or match its credentials", ErrCredentialConflict, img.Digest, dstRef.String()) } - if err := writeMetadata(m.paths, meta); err != nil { - m.mu.Unlock() - return nil, fmt.Errorf("write initial metadata: %w", err) + id := existing.id + m.mu.Unlock() + push, err := m.GetPush(ctx, id) + if errors.Is(err, ErrNotFound) { + // The job's terminal record could not be persisted and its + // directory was dropped, so the in-flight entry points at nothing + // readable. This is a rare mid-finalization window; surface a + // clear retryable error rather than orchestrating this caller into + // a successor job that does not yet exist. A retry once the entry + // drops creates a fresh job. + return nil, fmt.Errorf("%w: push job %s is being finalized after a record write failure; retry", ErrNotFound, id) } - m.inflight[key] = inflightPush{id: meta.ID, digest: meta.Digest, credFingerprint: fingerprint} + return push, err + } + + meta = &pushMetadata{ + ID: cuid2.Generate(), + Status: StatusQueued, + Image: img.Name, + Digest: img.Digest, + Target: dstRef.String(), + Insecure: req.Insecure, + HadCredentials: credsPresent(req.Credentials), + CreatedAt: time.Now(), + } + if err := writeMetadata(m.paths, meta); err != nil { m.mu.Unlock() - break + return nil, fmt.Errorf("write initial metadata: %w", err) } + m.inflight[key] = inflightPush{id: meta.ID, digest: meta.Digest, credFingerprint: fingerprint} + m.mu.Unlock() metaCopy := *meta queuePos := m.queue.Enqueue(key, func() { @@ -185,9 +174,13 @@ func (m *manager) CreatePush(ctx context.Context, req PushRequest) (*Push, error } func (m *manager) executePush(ctx context.Context, meta *pushMetadata, provider registrypush.Provider) { - // Contain panics in the job goroutine: record a failed terminal and - // notify waiters instead of leaving the job stuck as pushing. The queue - // slot is released by its own deferred completion. + // Bound each export so a wedged registry cannot pin a queue slot forever. + ctx, cancel := context.WithTimeout(ctx, pushTimeout) + defer cancel() + + // Contain panics in the job goroutine: record a failed terminal instead + // of leaving the job stuck as pushing. The queue slot is released by its + // own deferred completion. defer func() { if r := recover(); r != nil { fmt.Fprintf(os.Stderr, "Warning: push %s to %s panicked: %v\n", meta.ID, meta.Target, r) @@ -196,10 +189,7 @@ func (m *manager) executePush(ctx context.Context, meta *pushMetadata, provider meta.Status = StatusFailed meta.Error = &errorMsg meta.CompletedAt = &now - if err := m.writeTerminal(meta); err != nil { - os.RemoveAll(m.paths.PushDir(meta.ID)) - } - m.notify(meta.ID, StatusFailed, fmt.Errorf("push panicked: %v", r)) + m.persistTerminal(meta) } }() @@ -225,25 +215,21 @@ func (m *manager) executePush(ctx context.Context, meta *pushMetadata, provider } meta.CompletedAt = &now + m.persistTerminal(meta) +} + +// persistTerminal writes a terminal status and, if that write fails, drops the +// record so GetPush/ListPushes do not surface a half-written job. Both the +// normal completion path and the panic handler use it so they agree on what +// "couldn't persist" means; the actual push outcome only reaches the log. +func (m *manager) persistTerminal(meta *pushMetadata) { if err := m.writeTerminal(meta); err != nil { - // The outcome cannot be recorded: drop the record and report the job - // as failed with the persistence problem, so WaitForPush and GetPush - // agree instead of diverging into success-then-not-found. The actual - // push outcome goes to the log. fmt.Fprintf(os.Stderr, "Warning: push %s to %s finished as %s but the job record could not be persisted: %v\n", meta.ID, meta.Target, strings.ToLower(meta.Status), err) os.RemoveAll(m.paths.PushDir(meta.ID)) persistErr := fmt.Errorf("job record could not be persisted: %w", err) errorMsg := persistErr.Error() meta.Status = StatusFailed meta.Error = &errorMsg - m.notify(meta.ID, StatusFailed, persistErr) - return - } - - if pushErr != nil { - m.notify(meta.ID, StatusFailed, pushErr) - } else { - m.notify(meta.ID, StatusPushed, nil) } } @@ -272,30 +258,6 @@ func (m *manager) releaseInflight(key string) func() { } } -// waitForInflightRelease blocks until the key's torn-down inflight entry is -// dropped — or replaced by a successor job a concurrent create registered -// first, which the caller then merges into by retrying the dedup. Waiting on -// the entry's id rather than the key's absence is what keeps a second waiter -// from parking until the successor finishes and then starting a duplicate -// push. The queue releases the key's active slot before it runs the -// completion hook, so once the torn-down entry is gone a fresh Enqueue for -// the key starts immediately. The caller's context bounds the wait. -func (m *manager) waitForInflightRelease(ctx context.Context, key, tornDownID string) error { - for { - m.mu.Lock() - existing, ok := m.inflight[key] - m.mu.Unlock() - if !ok || existing.id != tornDownID { - return nil - } - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(5 * time.Millisecond): - } - } -} - func (m *manager) GetPush(ctx context.Context, id string) (*Push, error) { if err := ctx.Err(); err != nil { return nil, err @@ -305,10 +267,7 @@ func (m *manager) GetPush(ctx context.Context, id string) (*Push, error) { return nil, err } - push := meta.toPush() - if meta.Status == StatusQueued { - push.QueuePosition = m.queue.GetPosition(pushKey(meta.Digest, meta.Target, meta.Insecure)) - } + push := m.toPushWithPosition(meta) return push, nil } @@ -323,68 +282,23 @@ func (m *manager) ListPushes(ctx context.Context) ([]Push, error) { pushes := make([]Push, 0, len(metas)) for _, meta := range metas { - push := meta.toPush() - if meta.Status == StatusQueued { - push.QueuePosition = m.queue.GetPosition(pushKey(meta.Digest, meta.Target, meta.Insecure)) - } - pushes = append(pushes, *push) + pushes = append(pushes, *m.toPushWithPosition(meta)) } return pushes, nil } -// WaitForPush blocks until the push reaches a terminal state (pushed or -// failed) or the context is cancelled. -func (m *manager) WaitForPush(ctx context.Context, id string) error { - push, err := m.GetPush(ctx, id) - if err != nil { - return err - } - - switch push.Status { - case StatusPushed: - return nil - case StatusFailed: - return pushError(push) - } - - ch := make(chan StatusEvent, 1) - m.subscribe(id, ch) - defer m.unsubscribe(id, ch) - - // Re-check after subscribing to close the race window. - push, err = m.GetPush(ctx, id) - if err != nil { - return err - } - switch push.Status { - case StatusPushed: - return nil - case StatusFailed: - return pushError(push) - } - - select { - case event := <-ch: - if event.Status == StatusPushed { - return nil - } - if event.Err != nil { - return fmt.Errorf("push failed: %w", event.Err) - } - return fmt.Errorf("push failed") - case <-ctx.Done(): - return ctx.Err() - } -} - -func pushError(push *Push) error { - if push.Error != nil { - return fmt.Errorf("push failed: %s", *push.Error) +// toPushWithPosition projects a stored record to its domain form and, for a +// queued job, enriches it with the live pending-queue position. GetPush and +// ListPushes share this so the projection cannot drift between them. +func (m *manager) toPushWithPosition(meta *pushMetadata) *Push { + push := meta.toPush() + if meta.Status == StatusQueued { + push.QueuePosition = m.queue.GetPosition(pushKey(meta.Digest, meta.Target, meta.Insecure)) } - return fmt.Errorf("push failed") + return push } -func (m *manager) InProgressDigests() []string { +func (m *manager) inProgressDigests() []string { m.mu.Lock() defer m.mu.Unlock() @@ -404,7 +318,7 @@ func (m *manager) InProgressDigests() []string { // LiveCacheManifestDigests implements ocicachegc.RootsProvider so in-flight // push digests are treated as live alongside the OCI layout index. func (m *manager) LiveCacheManifestDigests() []string { - return m.InProgressDigests() + return m.inProgressDigests() } func (m *manager) recoverInterruptedPushes() { @@ -460,7 +374,6 @@ func (m *manager) recoverInterruptedPushes() { // failRecovered marks a recovered job failed. If the status cannot be // persisted, the record is removed instead of being left permanently queued. -// Subscribers are notified so a WaitForPush racing the close does not hang. func (m *manager) failRecovered(meta *pushMetadata, reason string) { meta.Status = StatusFailed meta.Error = &reason @@ -470,40 +383,4 @@ func (m *manager) failRecovered(meta *pushMetadata, reason string) { fmt.Fprintf(os.Stderr, "Warning: dropping unrecoverable push record %s: %v\n", meta.ID, err) os.RemoveAll(m.paths.PushDir(meta.ID)) } - m.notify(meta.ID, StatusFailed, errors.New(reason)) -} - -func (m *manager) subscribe(id string, ch chan StatusEvent) { - m.subscriberMu.Lock() - defer m.subscriberMu.Unlock() - m.subscribers[id] = append(m.subscribers[id], ch) -} - -func (m *manager) unsubscribe(id string, ch chan StatusEvent) { - m.subscriberMu.Lock() - defer m.subscriberMu.Unlock() - - subs := m.subscribers[id] - for i, sub := range subs { - if sub == ch { - m.subscribers[id] = append(subs[:i], subs[i+1:]...) - break - } - } - if len(m.subscribers[id]) == 0 { - delete(m.subscribers, id) - } -} - -func (m *manager) notify(id, status string, err error) { - m.subscriberMu.RLock() - defer m.subscriberMu.RUnlock() - - event := StatusEvent{Status: status, Err: err} - for _, ch := range m.subscribers[id] { - select { - case ch <- event: - default: - } - } } diff --git a/lib/imagepush/manager_test.go b/lib/imagepush/manager_test.go index d77e9145..d88dac86 100644 --- a/lib/imagepush/manager_test.go +++ b/lib/imagepush/manager_test.go @@ -2,6 +2,7 @@ package imagepush import ( "context" + "encoding/base64" "errors" "fmt" "io" @@ -162,17 +163,32 @@ func testManager(t *testing.T, maxConcurrent int, provider registrypush.Provider return mgr, digest } +// waitTerminal polls GetPush until the push reaches a terminal (pushed or +// failed) state and returns it. It replaces the removed WaitForPush surface: +// pushes complete asynchronously and tests observe the result by polling. +func waitTerminal(t *testing.T, mgr Manager, id string) *Push { + t.Helper() + deadline := time.Now().Add(15 * time.Second) + for { + got, err := mgr.GetPush(context.Background(), id) + if err != nil { + t.Fatalf("GetPush %s: %v", id, err) + } + if got.Status == StatusPushed || got.Status == StatusFailed { + return got + } + if time.Now().After(deadline) { + t.Fatalf("push %s never reached a terminal state (status=%s)", id, got.Status) + } + time.Sleep(5 * time.Millisecond) + } +} + // mustPushed waits for the push to reach a terminal pushed state and returns // it, failing the test otherwise. func mustPushed(t *testing.T, mgr Manager, id string) *Push { t.Helper() - if err := mgr.WaitForPush(context.Background(), id); err != nil { - t.Fatalf("WaitForPush %s: %v", id, err) - } - got, err := mgr.GetPush(context.Background(), id) - if err != nil { - t.Fatalf("GetPush %s: %v", id, err) - } + got := waitTerminal(t, mgr, id) if got.Status != StatusPushed { t.Fatalf("push %s status = %s, want pushed (error: %v)", id, got.Status, got.Error) } @@ -226,8 +242,8 @@ func TestCreatePushEndToEnd(t *testing.T) { } // No in-flight digests once done. - if digests := mgr.InProgressDigests(); len(digests) != 0 { - t.Errorf("InProgressDigests = %v, want empty", digests) + if digests := mgr.(*manager).inProgressDigests(); len(digests) != 0 { + t.Errorf("inProgressDigests = %v, want empty", digests) } } @@ -250,14 +266,12 @@ func TestCreatePushDedupesInFlight(t *testing.T) { t.Errorf("duplicate push got new ID %s, want %s", second.ID, first.ID) } - if digests := mgr.InProgressDigests(); len(digests) != 1 || digests[0] != digest { - t.Errorf("InProgressDigests = %v, want [%s]", digests, digest) + if digests := mgr.(*manager).inProgressDigests(); len(digests) != 1 || digests[0] != digest { + t.Errorf("inProgressDigests = %v, want [%s]", digests, digest) } close(gate) - if err := mgr.WaitForPush(context.Background(), first.ID); err != nil { - t.Fatalf("WaitForPush: %v", err) - } + mustPushed(t, mgr, first.ID) } func TestCreatePushQueuesBehindConcurrencyLimit(t *testing.T) { @@ -306,12 +320,8 @@ func TestCreatePushQueuesBehindConcurrencyLimit(t *testing.T) { } close(gate) - if err := mgr.WaitForPush(context.Background(), first.ID); err != nil { - t.Fatalf("WaitForPush first: %v", err) - } - if err := mgr.WaitForPush(context.Background(), second.ID); err != nil { - t.Fatalf("WaitForPush second: %v", err) - } + mustPushed(t, mgr, first.ID) + mustPushed(t, mgr, second.ID) } func TestCreatePushRejectsInvalidRequests(t *testing.T) { @@ -377,15 +387,7 @@ func TestCreatePushFailureRecorded(t *testing.T) { t.Fatalf("CreatePush: %v", err) } - err = mgr.WaitForPush(context.Background(), push.ID) - if err == nil { - t.Fatal("WaitForPush should fail for a failed push") - } - - got, err := mgr.GetPush(context.Background(), push.ID) - if err != nil { - t.Fatalf("GetPush: %v", err) - } + got := waitTerminal(t, mgr, push.ID) if got.Status != StatusFailed { t.Errorf("status = %s, want failed", got.Status) } @@ -404,9 +406,7 @@ func TestListPushesNewestFirst(t *testing.T) { if err != nil { t.Fatalf("CreatePush a: %v", err) } - if err := mgr.WaitForPush(context.Background(), first.ID); err != nil { - t.Fatalf("WaitForPush a: %v", err) - } + mustPushed(t, mgr, first.ID) second, err := mgr.CreatePush(context.Background(), PushRequest{ Image: "myapp:v1", Target: host + "/export/b:v1", Insecure: true, @@ -414,9 +414,7 @@ func TestListPushesNewestFirst(t *testing.T) { if err != nil { t.Fatalf("CreatePush b: %v", err) } - if err := mgr.WaitForPush(context.Background(), second.ID); err != nil { - t.Fatalf("WaitForPush b: %v", err) - } + mustPushed(t, mgr, second.ID) pushes, err := mgr.ListPushes(context.Background()) if err != nil { @@ -495,14 +493,12 @@ func TestCreatePushDedupesConcurrently(t *testing.T) { } } - if digests := mgr.InProgressDigests(); len(digests) != 1 || digests[0] != digest { - t.Errorf("InProgressDigests = %v, want [%s]", digests, digest) + if digests := mgr.(*manager).inProgressDigests(); len(digests) != 1 || digests[0] != digest { + t.Errorf("inProgressDigests = %v, want [%s]", digests, digest) } close(gate) - if err := mgr.WaitForPush(context.Background(), ids[0]); err != nil { - t.Fatalf("WaitForPush: %v", err) - } + mustPushed(t, mgr, ids[0]) } func TestCreatePushCredentialConflict(t *testing.T) { @@ -539,12 +535,8 @@ func TestCreatePushCredentialConflict(t *testing.T) { close(gateA) close(gateB) - if err := mgr.WaitForPush(context.Background(), seeded.ID); err != nil { - t.Fatalf("WaitForPush seeded: %v", err) - } - if err := mgr.WaitForPush(context.Background(), seeded2.ID); err != nil { - t.Fatalf("WaitForPush seeded 2: %v", err) - } + mustPushed(t, mgr, seeded.ID) + mustPushed(t, mgr, seeded2.ID) // The conflicted requests must not have created duplicate jobs: only the // two seeds exist. @@ -592,131 +584,7 @@ func TestCreatePushCredentialMismatch(t *testing.T) { } close(gate) - if err := mgr.WaitForPush(context.Background(), seeded.ID); err != nil { - t.Fatalf("WaitForPush: %v", err) - } -} - -func TestCreatePushDedupSurvivesTornDownKey(t *testing.T) { - mgr, digest := testManager(t, 1, nil, nil) - host := openRegistry(t) - target := host + "/export/app:v1" - - dstRef, err := name.ParseReference(target, name.Insecure) - if err != nil { - t.Fatalf("ParseReference: %v", err) - } - key := pushKey(digest, dstRef.String(), true) - - // Simulate the persist-failure teardown window: the job's record is gone - // from disk but its inflight entry is still registered, released by the - // queue completion hook moments later. - m := mgr.(*manager) - m.mu.Lock() - m.inflight[key] = inflightPush{id: "ghost", digest: digest} - m.mu.Unlock() - go func() { - time.Sleep(50 * time.Millisecond) - m.mu.Lock() - delete(m.inflight, key) - m.mu.Unlock() - }() - - // The dedup path must wait out the torn-down key and create a fresh job, - // not surface ErrNotFound from a create. - push, err := mgr.CreatePush(context.Background(), PushRequest{Image: "myapp:v1", Target: target, Insecure: true}) - if err != nil { - t.Fatalf("CreatePush: %v", err) - } - if push.ID == "ghost" { - t.Fatal("CreatePush returned the torn-down job") - } - mustPushed(t, mgr, push.ID) -} - -func TestCreatePushDedupWaitersMergeIntoSuccessor(t *testing.T) { - mgr, digest := testManager(t, 1, nil, nil) - host := openRegistry(t) - target := host + "/export/app:v1" - - dstRef, err := name.ParseReference(target, name.Insecure) - if err != nil { - t.Fatalf("ParseReference: %v", err) - } - key := pushKey(digest, dstRef.String(), true) - - // Two concurrent creates racing the same torn-down key: one must create - // the successor job and the other must merge into it — not wait out the - // successor and then start a duplicate push. - m := mgr.(*manager) - m.mu.Lock() - m.inflight[key] = inflightPush{id: "ghost", digest: digest} - m.mu.Unlock() - go func() { - time.Sleep(50 * time.Millisecond) - m.mu.Lock() - delete(m.inflight, key) - m.mu.Unlock() - }() - - ids := make([]string, 2) - errs := make([]error, 2) - var wg sync.WaitGroup - for i := range ids { - wg.Add(1) - go func(i int) { - defer wg.Done() - push, err := mgr.CreatePush(context.Background(), PushRequest{Image: "myapp:v1", Target: target, Insecure: true}) - if push != nil { - ids[i] = push.ID - } - errs[i] = err - }(i) - } - wg.Wait() - - for i := range ids { - if errs[i] != nil { - t.Fatalf("CreatePush #%d: %v", i, errs[i]) - } - } - if ids[0] != ids[1] { - t.Errorf("concurrent creates got IDs %s and %s, want one shared successor job", ids[0], ids[1]) - } - mustPushed(t, mgr, ids[0]) - - pushes, err := mgr.ListPushes(context.Background()) - if err != nil { - t.Fatalf("ListPushes: %v", err) - } - if len(pushes) != 1 { - t.Errorf("len(pushes) = %d, want 1 (no duplicate after the successor)", len(pushes)) - } -} - -func TestWaitForPushCancellation(t *testing.T) { - mgr, _ := testManager(t, 1, nil, nil) - host, gate := gatedRegistry(t) - - push, err := mgr.CreatePush(context.Background(), PushRequest{Image: "myapp:v1", Target: host + "/export/app:v1", Insecure: true}) - if err != nil { - t.Fatalf("CreatePush: %v", err) - } - - ctx, cancel := context.WithCancel(context.Background()) - errCh := make(chan error, 1) - go func() { errCh <- mgr.WaitForPush(ctx, push.ID) }() - cancel() - if err := <-errCh; !errors.Is(err, context.Canceled) { - t.Errorf("WaitForPush err = %v, want context.Canceled", err) - } - - // Let the in-flight job finish so its writes land before the fixture's - // TempDir cleanup. - close(gate) - if err := mgr.WaitForPush(context.Background(), push.ID); err != nil { - t.Fatalf("WaitForPush after cancel: %v", err) - } + mustPushed(t, mgr, seeded.ID) } func TestInProgressDigestsDedupesAcrossTargets(t *testing.T) { @@ -735,8 +603,8 @@ func TestInProgressDigestsDedupesAcrossTargets(t *testing.T) { pushes = append(pushes, push.ID) } - if digests := mgr.InProgressDigests(); len(digests) != 1 || digests[0] != digest { - t.Errorf("InProgressDigests = %v, want [%s]", digests, digest) + if digests := mgr.(*manager).inProgressDigests(); len(digests) != 1 || digests[0] != digest { + t.Errorf("inProgressDigests = %v, want [%s]", digests, digest) } // Drain the gated jobs so their writes land before the fixture's TempDir @@ -766,9 +634,7 @@ func TestRecoveryDedupesSameKey(t *testing.T) { t.Fatalf("NewManager: %v", err) } - if err := mgr.WaitForPush(context.Background(), "older"); err != nil { - t.Fatalf("WaitForPush older: %v", err) - } + mustPushed(t, mgr, "older") got, err := mgr.GetPush(context.Background(), "newer") if err != nil { @@ -780,9 +646,9 @@ func TestRecoveryDedupesSameKey(t *testing.T) { if got.Error == nil || !strings.Contains(*got.Error, "duplicate of push job older") { t.Errorf("newer error = %v, want duplicate-of-older explanation", got.Error) } - // WaitForPush on the superseded job surfaces the failure rather than hanging. - if err := mgr.WaitForPush(context.Background(), "newer"); err == nil { - t.Error("WaitForPush on superseded job should fail") + // The superseded job surfaces the failure rather than hanging. + if got := waitTerminal(t, mgr, "newer"); got.Status != StatusFailed { + t.Errorf("superseded job status = %s, want failed", got.Status) } } @@ -802,15 +668,6 @@ func TestSequentialSameKeyPushesAllComplete(t *testing.T) { } } -func TestWaitForPushNotFound(t *testing.T) { - mgr, _ := testManager(t, 1, nil, nil) - - err := mgr.WaitForPush(context.Background(), "missing") - if !errors.Is(err, ErrNotFound) { - t.Errorf("err = %v, want ErrNotFound", err) - } -} - // erroringProvider always fails, proving a push that succeeds used the // request's borrowed credentials instead of the manager default. type erroringProvider struct{} @@ -867,9 +724,7 @@ func TestCredentialsNeverPersisted(t *testing.T) { if err != nil { t.Fatalf("CreatePush: %v", err) } - if err := mgr.WaitForPush(context.Background(), push.ID); err != nil { - t.Fatalf("WaitForPush: %v", err) - } + mustPushed(t, mgr, push.ID) data, err := os.ReadFile(p.PushMetadata(push.ID)) if err != nil { @@ -949,14 +804,12 @@ func TestCreatePushMissingBlobs(t *testing.T) { if err != nil { t.Fatalf("CreatePush: %v", err) } - err = mgr.WaitForPush(context.Background(), push.ID) - if err == nil { - t.Fatal("WaitForPush should fail when cache blobs are missing") + got := waitTerminal(t, mgr, push.ID) + if got.Status != StatusFailed { + t.Fatalf("status = %s, want failed (error: %v)", got.Status, got.Error) } - // Depending on timing the failure is observed via the live event (typed) - // or via persisted metadata (string), so accept both forms. - if !errors.Is(err, ocicache.ErrNotFound) && !strings.Contains(err.Error(), ocicache.ErrNotFound.Error()) { - t.Errorf("err = %v, want ocicache.ErrNotFound", err) + if got.Error == nil || !strings.Contains(*got.Error, ocicache.ErrNotFound.Error()) { + t.Errorf("error = %v, want ocicache.ErrNotFound", got.Error) } } @@ -987,14 +840,12 @@ func TestRecoveryFailsWhenBlobsReclaimed(t *testing.T) { t.Fatalf("NewManager: %v", err) } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - err = mgr.WaitForPush(ctx, "recovered-missing-blobs") - if err == nil { - t.Fatal("WaitForPush should fail when cache blobs were reclaimed") + got := waitTerminal(t, mgr, "recovered-missing-blobs") + if got.Status != StatusFailed { + t.Fatalf("status = %s, want failed (error: %v)", got.Status, got.Error) } - if !errors.Is(err, ocicache.ErrNotFound) && !strings.Contains(err.Error(), ocicache.ErrNotFound.Error()) { - t.Errorf("err = %v, want ocicache.ErrNotFound", err) + if got.Error == nil || !strings.Contains(*got.Error, ocicache.ErrNotFound.Error()) { + t.Errorf("error = %v, want ocicache.ErrNotFound", got.Error) } } @@ -1017,23 +868,13 @@ func TestExecutePushContainsPanic(t *testing.T) { t.Fatalf("CreatePush: %v", err) } - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - err = mgr.WaitForPush(ctx, push.ID) - if err == nil { - t.Fatal("WaitForPush should fail for a panicked push") - } - if !strings.Contains(err.Error(), "panicked") { - t.Errorf("err = %v, want panic explanation", err) - } - - got, err := mgr.GetPush(context.Background(), push.ID) - if err != nil { - t.Fatalf("GetPush: %v", err) - } + got := waitTerminal(t, mgr, push.ID) if got.Status != StatusFailed { t.Errorf("status = %s, want failed", got.Status) } + if got.Error == nil || !strings.Contains(*got.Error, "panicked") { + t.Errorf("error = %v, want panic explanation", got.Error) + } } func TestRecoveryTreatsInsecureAsDistinctKey(t *testing.T) { @@ -1090,3 +931,25 @@ func TestRecoverySweepsOrphanDirs(t *testing.T) { } mustPushed(t, mgr, "real") } + +func TestCredFingerprintNormalizesAuth(t *testing.T) { + // The same login supplied as Username/Password and as the precomputed + // base64 "user:pass" Auth shorthand must hash identically, so in-flight + // dedup does not report a false credential conflict between the two forms. + basic := &authn.AuthConfig{Username: "pusher", Password: "hunter2"} + shorthand := &authn.AuthConfig{ + Auth: base64.StdEncoding.EncodeToString([]byte("pusher:hunter2")), + } + + basicFp := credFingerprint(basic) + if basicFp == "" { + t.Fatal("basic-auth fingerprint should be non-empty") + } + shorthandFp := credFingerprint(shorthand) + if shorthandFp != basicFp { + t.Errorf("Auth-shorthand fingerprint %q != basic %q", shorthandFp, basicFp) + } + if credFingerprint(nil) != "" || credFingerprint(&authn.AuthConfig{}) != "" { + t.Error("anonymous configs should share the empty fingerprint") + } +} diff --git a/lib/imagepush/storage.go b/lib/imagepush/storage.go index a269f374..e231bf07 100644 --- a/lib/imagepush/storage.go +++ b/lib/imagepush/storage.go @@ -85,6 +85,15 @@ func writeMetadata(p *paths.Paths, meta *pushMetadata) error { return fmt.Errorf("rename metadata: %w", err) } + // Sync the directory so the rename itself is durable: the file is fsync'd + // and renamed above, but without a directory sync a crash right after the + // rename can still lose the directory entry. Best-effort — a directory + // sync failure is not worth failing the write over. + if dir, err := os.Open(dir); err == nil { + _ = dir.Sync() + _ = dir.Close() + } + return nil }