diff --git a/lib/images/credentials_test.go b/lib/images/credentials_test.go index 059dac6a..46212ba7 100644 --- a/lib/images/credentials_test.go +++ b/lib/images/credentials_test.go @@ -113,11 +113,30 @@ func TestBorrowedCredentialsExpireWhileQueued(t *testing.T) { return m.inflightPulls[digest].credentials == nil }, time.Second, time.Millisecond) - credentials, _, expired := m.borrowedAuth(digest) + credentials, _, expired, stale := m.borrowedAuth(digest, inflight) assert.True(t, expired) + assert.False(t, stale) assert.Nil(t, credentials) } +func TestBorrowedAuthRejectsReplacedInflightPull(t *testing.T) { + m := &manager{inflightPulls: make(map[string]*inflightImagePull)} + const digest = "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + first := m.registerInflightPull(digest, &authn.AuthConfig{Username: "first"}) + second := m.registerInflightPull(digest, &authn.AuthConfig{Username: "second"}) + defer m.releaseInflightPull(digest, second)() + + credentials, _, expired, stale := m.borrowedAuth(digest, first) + assert.Nil(t, credentials) + assert.False(t, expired) + assert.True(t, stale) + + credentials, _, expired, stale = m.borrowedAuth(digest, second) + assert.Equal(t, "second", credentials.Username) + assert.False(t, expired) + assert.False(t, stale) +} + func TestRecoverInterruptedCredentialedConversionFromCache(t *testing.T) { p := paths.New(t.TempDir()) img := createTestDockerImage(t) diff --git a/lib/images/manager.go b/lib/images/manager.go index 8e3e84ad..c1a0da0a 100644 --- a/lib/images/manager.go +++ b/lib/images/manager.go @@ -14,12 +14,15 @@ import ( "time" "github.com/google/go-containerregistry/pkg/authn" + "github.com/google/uuid" "github.com/kernel/hypeman/lib/paths" "github.com/kernel/hypeman/lib/queue" "github.com/kernel/hypeman/lib/tags" "go.opentelemetry.io/otel/metric" ) +var errStaleBuild = errors.New("stale image build") + const ( StatusPending = "pending" StatusPulling = "pulling" @@ -297,6 +300,9 @@ func (m *manager) registerInflightPull(digest string, credentials *authn.AuthCon if m.inflightPulls == nil { m.inflightPulls = make(map[string]*inflightImagePull) } + if previous := m.inflightPulls[digest]; previous != nil && previous.timer != nil { + previous.timer.Stop() + } inflight := &inflightImagePull{ fingerprint: credentialFingerprint(credentials), credentials: credentials, @@ -334,17 +340,20 @@ func (m *manager) releaseInflightPull(digest string, inflight *inflightImagePull } } -func (m *manager) borrowedAuth(digest string) (*authn.AuthConfig, time.Time, bool) { +func (m *manager) borrowedAuth(digest string, expected *inflightImagePull) (*authn.AuthConfig, time.Time, bool, bool) { m.createMu.Lock() defer m.createMu.Unlock() inflight := m.inflightPulls[digest] + if expected != nil && inflight != expected { + return nil, time.Time{}, false, true + } if inflight == nil || inflight.credentialsExpireAt.IsZero() { - return nil, time.Time{}, false + return nil, time.Time{}, false, false } if inflight.credentials == nil || time.Now().After(inflight.credentialsExpireAt) { - return nil, inflight.credentialsExpireAt, true + return nil, inflight.credentialsExpireAt, true, false } - return inflight.credentials, inflight.credentialsExpireAt, false + return inflight.credentials, inflight.credentialsExpireAt, false, false } func (m *manager) createAndQueueImage(ref *ResolvedRef, req CreateImageRequest, requestedPlatform Platform) (*Image, error) { @@ -365,6 +374,7 @@ func (m *manager) createAndQueueImage(ref *ResolvedRef, req CreateImageRequest, Status: StatusPending, Request: &storedReq, BorrowedAuth: req.Credentials != nil, + BuildID: uuid.New().String(), Tags: tags.Clone(req.Tags), CreatedAt: time.Now(), } @@ -377,10 +387,14 @@ func (m *manager) createAndQueueImage(ref *ResolvedRef, req CreateImageRequest, // Keep borrowed credentials outside the queued closure so their lifetime is // bounded even when this job waits behind another pull. inflight := m.registerInflightPull(ref.Digest(), req.Credentials) - queuePos := m.queue.Enqueue(ref.Digest(), func() { - credentials, deadline, expired := m.borrowedAuth(ref.Digest()) + buildID := meta.BuildID + queuePos := m.queue.EnqueueSuccessor(ref.Digest(), func() { + credentials, deadline, expired, stale := m.borrowedAuth(ref.Digest(), inflight) + if stale { + return + } if expired { - m.updateStatusByDigest(ref, StatusFailed, ErrBorrowedCredentialsExpired) + m.updateStatusByDigest(ref, StatusFailed, ErrBorrowedCredentialsExpired, buildID) return } ctx := context.Background() @@ -389,7 +403,7 @@ func (m *manager) createAndQueueImage(ref *ResolvedRef, req CreateImageRequest, ctx, cancel = context.WithDeadline(ctx, deadline) defer cancel() } - m.buildImage(ctx, ref, credentials) + m.buildImage(ctx, ref, credentials, buildID) }, m.releaseInflightPull(ref.Digest(), inflight)) img := meta.toImage() @@ -399,7 +413,7 @@ func (m *manager) createAndQueueImage(ref *ResolvedRef, req CreateImageRequest, return img, nil } -func (m *manager) buildImage(ctx context.Context, ref *ResolvedRef, credentials *authn.AuthConfig) { +func (m *manager) buildImage(ctx context.Context, ref *ResolvedRef, credentials *authn.AuthConfig, buildID string) { buildStart := time.Now() buildStatus := "failed" buildDir := m.paths.SystemBuild(ref.String()) @@ -409,7 +423,7 @@ func (m *manager) buildImage(ctx context.Context, ref *ResolvedRef, credentials }() if err := os.MkdirAll(buildDir, 0755); err != nil { - m.updateStatusByDigest(ref, StatusFailed, fmt.Errorf("create build dir: %w", err)) + m.updateStatusByDigest(ref, StatusFailed, fmt.Errorf("create build dir: %w", err), buildID) return } @@ -420,7 +434,7 @@ func (m *manager) buildImage(ctx context.Context, ref *ResolvedRef, credentials m.recordImageBuildPhase(ctx, ref.Digest(), "cleanup", time.Since(start), phaseStatus(err), "not_applicable") }() - m.updateStatusByDigest(ref, StatusPulling, nil) + m.updateStatusByDigest(ref, StatusPulling, nil, buildID) // Pull by the digest-pinned reference, not the tag: a digest ref fetches the // exact manifest regardless of the platform passed downstream, so a @@ -431,7 +445,7 @@ func (m *manager) buildImage(ctx context.Context, ref *ResolvedRef, credentials result, err := m.ociClient.pullAndExportWithAuth(ctx, pullRef, ref.Digest(), tempDir, credentials) m.recordPullResultMetrics(ctx, ref.Digest(), result) if err != nil { - m.updateStatusByDigest(ref, StatusFailed, fmt.Errorf("pull and export: %w", err)) + m.updateStatusByDigest(ref, StatusFailed, fmt.Errorf("pull and export: %w", err), buildID) m.recordPullMetrics(ctx, "failed") return } @@ -449,7 +463,7 @@ func (m *manager) buildImage(ctx context.Context, ref *ResolvedRef, credentials } } - m.updateStatusByDigest(ref, StatusConverting, nil) + m.updateStatusByDigest(ref, StatusConverting, nil, buildID) diskPath := digestPath(m.paths, ref.Repository(), ref.DigestHex()) // Use default image format (erofs on Linux, ext4 on Darwin) @@ -457,30 +471,32 @@ func (m *manager) buildImage(ctx context.Context, ref *ResolvedRef, credentials diskSize, err := ExportRootfs(tempDir, diskPath, DefaultImageFormat) m.recordImageBuildPhase(ctx, ref.Digest(), "filesystem_export", time.Since(convertStart), phaseStatus(err), "not_applicable") if err != nil { - m.updateStatusByDigest(ref, StatusFailed, fmt.Errorf("convert to %s: %w", DefaultImageFormat, err)) + m.updateStatusByDigest(ref, StatusFailed, fmt.Errorf("convert to %s: %w", DefaultImageFormat, err), buildID) return } finalizeStart := time.Now() - err = m.finalizeImage(ref, result, diskSize) + err = m.finalizeImage(ref, result, diskSize, buildID) m.recordImageBuildPhase(ctx, ref.Digest(), "finalize", time.Since(finalizeStart), phaseStatus(err), "not_applicable") if err != nil { - m.updateStatusByDigest(ref, StatusFailed, err) + if errors.Is(err, errStaleBuild) { + return + } + m.updateStatusByDigest(ref, StatusFailed, err, buildID) return } buildStatus = "success" } -func (m *manager) finalizeImage(ref *ResolvedRef, result *pullResult, diskSize int64) error { - // Read current metadata to preserve request info. +func (m *manager) finalizeImage(ref *ResolvedRef, result *pullResult, diskSize int64, buildID string) error { + m.createMu.Lock() + defer m.createMu.Unlock() + + // Read current metadata to preserve request info and reject stale builds. meta, err := readMetadata(m.paths, ref.Repository(), ref.DigestHex()) - if err != nil { - meta = &imageMetadata{ - Name: ref.String(), - Digest: ref.Digest(), - CreatedAt: time.Now(), - } + if err != nil || meta.BuildID != buildID { + return errStaleBuild } // The pulled image config is the source of truth for the platform. @@ -557,19 +573,15 @@ func (m *manager) recordImageBuildPhase(ctx context.Context, digest, phase strin ) } -func (m *manager) updateStatusByDigest(ref *ResolvedRef, status string, err error) { +func (m *manager) updateStatusByDigest(ref *ResolvedRef, status string, err error, buildID string) { + m.createMu.Lock() + defer m.createMu.Unlock() + meta, readErr := readMetadata(m.paths, ref.Repository(), ref.DigestHex()) - if readErr != nil { - // Create new metadata if it doesn't exist - meta = &imageMetadata{ - Name: ref.String(), - Digest: ref.Digest(), - Status: status, - CreatedAt: time.Now(), - } - } else { - meta.Status = status + if readErr != nil || meta.BuildID != buildID { + return } + meta.Status = status if err != nil { errorMsg := err.Error() @@ -578,7 +590,8 @@ func (m *manager) updateStatusByDigest(ref *ResolvedRef, status string, err erro writeMetadata(m.paths, ref.Repository(), ref.DigestHex(), meta) - // Notify subscribers of terminal status + // Notify while holding createMu so a delete/recreate cannot race the + // metadata write and receive a terminal event for the old build. if status == StatusReady || status == StatusFailed { m.notifyReady(ref.DigestHex(), status, err) } @@ -608,11 +621,12 @@ func (m *manager) RecoverInterruptedBuilds() { } ref := NewResolvedRef(normalized, meta.Digest) if meta.BorrowedAuth && (meta.Status != StatusConverting || !m.ociClient.existsInLayout(digestToLayoutTag(meta.Digest))) { - m.updateStatusByDigest(ref, StatusFailed, ErrBorrowedCredentialsExpired) + m.updateStatusByDigest(ref, StatusFailed, ErrBorrowedCredentialsExpired, meta.BuildID) continue } + buildID := meta.BuildID m.queue.Enqueue(meta.Digest, func() { - m.buildImage(context.Background(), ref, nil) + m.buildImage(context.Background(), ref, nil, buildID) }, nil) } } diff --git a/lib/images/manager_test.go b/lib/images/manager_test.go index 974000e0..89d3fb4f 100644 --- a/lib/images/manager_test.go +++ b/lib/images/manager_test.go @@ -602,6 +602,123 @@ func TestImportLocalImageFromOCICache(t *testing.T) { t.Logf("Disk path verified: %s (%d bytes)", diskPath, diskStat.Size()) } +func TestDeleteAndRecreateDuringBuildTail(t *testing.T) { + origFormat := DefaultImageFormat + DefaultImageFormat = FormatCpio + defer func() { DefaultImageFormat = origFormat }() + + dataDir := t.TempDir() + p := paths.New(dataDir) + mgr, err := NewManager(p, 1, nil) + require.NoError(t, err) + m := mgr.(*manager) + + ctx := context.Background() + const repo = "kernel.local/test/recreate-race" + const tag = "v1" + + testImg := createTestDockerImage(t) + imgDigest, err := testImg.Digest() + require.NoError(t, err) + digestStr := imgDigest.String() + + cacheDir := p.SystemOCICache() + layoutPath, err := layout.Write(cacheDir, empty.Index) + require.NoError(t, err) + require.NoError(t, layoutPath.AppendImage(testImg, layout.WithAnnotations(map[string]string{ + "org.opencontainers.image.ref.name": digestToLayoutTag(digestStr), + }))) + + digestHex := digestToLayoutTag(digestStr) + events := make(chan StatusEvent, 2) + m.subscribeToReady(digestHex, events) + defer m.unsubscribeFromReady(digestHex, events) + + _, err = m.ImportLocalImage(ctx, repo, tag, digestStr) + require.NoError(t, err) + select { + case event := <-events: + require.Equal(t, StatusReady, event.Status) + case <-time.After(30 * time.Second): + t.Fatal("first build did not become ready") + } + firstMeta, err := readMetadata(p, repo, digestHex) + require.NoError(t, err) + require.NotEmpty(t, firstMeta.BuildID) + + slotHeld := make(chan struct{}) + releaseSlot := make(chan struct{}) + m.queue.EnqueueSuccessor(digestStr, func() { + close(slotHeld) + <-releaseSlot + }, nil) + select { + case <-slotHeld: + case <-time.After(5 * time.Second): + t.Fatal("queue slot was not held") + } + + // Delete by digest so the test does not depend on the tag symlink being + // created after the ready notification. + require.NoError(t, m.DeleteImage(ctx, repo+"@"+digestStr)) + recreated, err := m.ImportLocalImage(ctx, repo, tag, digestStr) + require.NoError(t, err) + require.Equal(t, StatusPending, recreated.Status) + require.NotNil(t, recreated.QueuePosition) + require.Equal(t, 1, *recreated.QueuePosition) + + currentMeta, err := readMetadata(p, repo, digestHex) + require.NoError(t, err) + require.NotEqual(t, firstMeta.BuildID, currentMeta.BuildID) + require.NoError(t, m.DeleteImage(ctx, repo+"@"+digestStr)) + recreatedAgain, err := m.ImportLocalImage(ctx, repo, tag, digestStr) + require.NoError(t, err) + require.Equal(t, StatusPending, recreatedAgain.Status) + require.NotNil(t, recreatedAgain.QueuePosition) + require.Equal(t, 1, *recreatedAgain.QueuePosition) + latestMeta, err := readMetadata(p, repo, digestHex) + require.NoError(t, err) + require.NotEqual(t, currentMeta.BuildID, latestMeta.BuildID) + currentMeta = latestMeta + + waitCtx, cancelWait := context.WithCancel(ctx) + defer cancelWait() + waitResult := make(chan error, 1) + go func() { + waitResult <- m.WaitForReady(waitCtx, repo+"@"+digestStr) + }() + select { + case err := <-waitResult: + t.Fatalf("recreated image completed before successor ran: %v", err) + case <-time.After(100 * time.Millisecond): + } + normalized, err := ParseNormalizedRef(repo + "@" + digestStr) + require.NoError(t, err) + staleRef := NewResolvedRef(normalized, digestStr) + m.updateStatusByDigest(staleRef, StatusFailed, errors.New("stale build"), firstMeta.BuildID) + staleResult, _, _, err := m.ociClient.extractOCIImageDetails(digestHex) + require.NoError(t, err) + require.ErrorIs(t, m.finalizeImage(staleRef, &pullResult{Metadata: staleResult}, 1, firstMeta.BuildID), errStaleBuild) + currentMeta, err = readMetadata(p, repo, digestHex) + require.NoError(t, err) + require.Equal(t, StatusPending, currentMeta.Status) + require.Nil(t, currentMeta.Error) + + close(releaseSlot) + select { + case event := <-events: + require.Equal(t, StatusReady, event.Status) + case <-time.After(30 * time.Second): + t.Fatal("recreated image did not become ready") + } + select { + case err := <-waitResult: + require.NoError(t, err) + case <-time.After(30 * time.Second): + t.Fatal("WaitForReady did not observe the successor build") + } +} + // waitForReady waits for an image build to complete func waitForReady(t *testing.T, mgr Manager, ctx context.Context, imageName string) { for i := 0; i < 600; i++ { diff --git a/lib/images/storage.go b/lib/images/storage.go index 02a53d60..8764400c 100644 --- a/lib/images/storage.go +++ b/lib/images/storage.go @@ -29,6 +29,7 @@ type imageMetadata struct { WorkingDir string `json:"working_dir,omitempty"` CreatedAt time.Time `json:"created_at"` BorrowedAuth bool `json:"borrowed_auth,omitempty"` + BuildID string `json:"build_id,omitempty"` } func (m *imageMetadata) toImage() *Image { diff --git a/lib/queue/queue.go b/lib/queue/queue.go index 3cd580c1..6e6c55c1 100644 --- a/lib/queue/queue.go +++ b/lib/queue/queue.go @@ -39,17 +39,24 @@ func New(maxConcurrent int) *Queue { // is torn down only once the queue is done with it), but only when this call // actually started the job — a dedup'd enqueue does not run it. func (q *Queue) Enqueue(key string, startFn func(), done func()) int { + return q.enqueue(key, startFn, done, false, false) +} + +// EnqueueSuccessor queues a new job behind an active job with the same key. +// If a successor is already pending, it is replaced by the new job. This is +// useful when the caller has replaced the persisted work for a key while the +// previous job is finishing. +func (q *Queue) EnqueueSuccessor(key string, startFn func(), done func()) int { + return q.enqueue(key, startFn, done, true, true) +} + +func (q *Queue) enqueue(key string, startFn func(), done func(), allowActive, replacePending bool) int { q.mu.Lock() defer q.mu.Unlock() - if q.active[key] { + if q.active[key] && !allowActive { return 0 } - for i, j := range q.pending { - if j.key == key { - return i + 1 - } - } wrappedFn := func() { // complete runs first (last-registered defer runs first), so done @@ -61,7 +68,16 @@ func (q *Queue) Enqueue(key string, startFn func(), done func()) int { startFn() } - if len(q.active) < q.maxConcurrent { + for i, j := range q.pending { + if j.key == key { + if replacePending { + q.pending[i] = job{key: key, startFn: wrappedFn} + } + return i + 1 + } + } + + if len(q.active) < q.maxConcurrent && !q.active[key] { q.active[key] = true // The key enters the active set before the goroutine launches, both // under the lock, so a concurrent Enqueue always observes the slot and @@ -77,17 +93,23 @@ func (q *Queue) Enqueue(key string, startFn func(), done func()) int { } // complete removes the key from the active set and starts the next pending -// job if there is capacity. +// job if there is capacity. Pending successors for an active key are skipped +// until that key completes. func (q *Queue) complete(key string) { q.mu.Lock() delete(q.active, key) var next *job - if len(q.pending) > 0 && len(q.active) < q.maxConcurrent { - nextJob := q.pending[0] - q.pending = q.pending[1:] - q.active[nextJob.key] = true - next = &nextJob + if len(q.active) < q.maxConcurrent { + for i, pending := range q.pending { + if q.active[pending.key] { + continue + } + q.pending = append(q.pending[:i], q.pending[i+1:]...) + q.active[pending.key] = true + next = &pending + break + } } q.mu.Unlock() @@ -96,15 +118,12 @@ func (q *Queue) complete(key string) { } } -// GetPosition returns nil if the key is active or unknown, otherwise its -// 1-based position in the pending queue. +// GetPosition returns nil if the key is unknown or active without a +// successor, otherwise its 1-based position in the pending queue. func (q *Queue) GetPosition(key string) *int { q.mu.Lock() defer q.mu.Unlock() - if q.active[key] { - return nil - } for i, j := range q.pending { if j.key == key { pos := i + 1 diff --git a/lib/queue/queue_test.go b/lib/queue/queue_test.go index 95b838b2..be12f68c 100644 --- a/lib/queue/queue_test.go +++ b/lib/queue/queue_test.go @@ -103,6 +103,107 @@ func TestDedupesByKey(t *testing.T) { } } +// TestEnqueueSuccessor queues a replacement behind an active job without +// changing the normal active-key deduplication behavior. +func TestEnqueueSuccessor(t *testing.T) { + q := New(1) + release := make(chan struct{}) + started := make(chan struct{}, 1) + successorStarted := make(chan struct{}) + replacementDone := make(chan struct{}) + + q.Enqueue("same", func() { + started <- struct{}{} + <-release + }, nil) + <-started + + pos := q.EnqueueSuccessor("same", func() { + close(successorStarted) + }, nil) + if pos != 1 { + t.Fatalf("successor position = %d, want 1", pos) + } + if pos := q.GetPosition("same"); pos == nil || *pos != 1 { + t.Fatalf("GetPosition(same) = %v, want 1", pos) + } + + duplicatePos := q.EnqueueSuccessor("same", func() { + close(successorStarted) + }, func() { + close(replacementDone) + }) + if duplicatePos != 1 { + t.Fatalf("duplicate successor position = %d, want 1", duplicatePos) + } + + select { + case <-successorStarted: + t.Fatal("successor started before its predecessor was released") + default: + } + + close(release) + select { + case <-replacementDone: + case <-time.After(5 * time.Second): + t.Fatal("replacement successor did not run after active job completed") + } + select { + case <-successorStarted: + default: + t.Fatal("replacement successor did not start") + } +} + +func TestEnqueueSuccessorSkipsActiveKey(t *testing.T) { + q := New(2) + releaseA := make(chan struct{}) + releaseB := make(chan struct{}) + successorDone := make(chan struct{}) + startedA := make(chan struct{}, 1) + startedB := make(chan struct{}, 1) + startedC := make(chan struct{}, 1) + + q.Enqueue("a", func() { + startedA <- struct{}{} + <-releaseA + }, nil) + <-startedA + q.EnqueueSuccessor("a", func() {}, func() { + close(successorDone) + }) + q.Enqueue("b", func() { + startedB <- struct{}{} + <-releaseB + }, nil) + <-startedB + cDone := make(chan struct{}) + q.Enqueue("c", func() { + startedC <- struct{}{} + }, func() { + close(cDone) + }) + + close(releaseB) + select { + case <-startedC: + case <-time.After(5 * time.Second): + t.Fatal("unblocked pending job did not start") + } + select { + case <-cDone: + case <-time.After(5 * time.Second): + t.Fatal("unblocked pending job did not complete") + } + close(releaseA) + select { + case <-successorDone: + case <-time.After(5 * time.Second): + t.Fatal("successor did not run after its active job completed") + } +} + // TestDoneRunsAfterKeyReleased guards the completion-hook ordering: done must // fire only after the key has left the active set, otherwise a caller's // "job finished" bookkeeping would race a concurrent re-enqueue of the key.