Queue image rebuilds behind in-flight builds instead of dropping them

lib/images/manager_test.go

@@ -613,3 +613,94 @@ func waitForReady(t *testing.T, mgr Manager, ctx context.Context, imageName stri
 
 	t.Fatal("Build did not complete within 60 seconds")
 }
+
+// TestDeleteAndRecreateDuringBuildTail exercises the race where a delete +
+// recreate lands between a build's ready signal and its queue-slot release:
+// the digest is still marked active in the build queue, and the recreated
+// image must still build and become ready.
+func TestDeleteAndRecreateDuringBuildTail(t *testing.T) {
+	// Use the pure-Go cpio exporter so the test needs no mkfs.erofs binary.
+	origFormat := DefaultImageFormat
+	DefaultImageFormat = FormatCpio
+	defer func() { DefaultImageFormat = origFormat }()
+
+	dataDir := t.TempDir()
+	p := paths.New(dataDir)
+
+	// Seed the OCI cache with a synthetic image so builds need no network.
+	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),
+	})))
+
+	client, err := newOCIClient(cacheDir)
+	require.NoError(t, err)
+
+	m := &manager{
+		paths:            p,
+		ociClient:        client,
+		queue:            NewBuildQueue(1),
+		readySubscribers: make(map[string][]chan StatusEvent),
+	}
+
+	ctx := context.Background()
+	const repo = "kernel.local/test/recreate-race"
+	const tag = "v1"
+	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 ev := <-events:
+		require.Equal(t, StatusReady, ev.Status)
+	case <-time.After(30 * time.Second):
+		t.Fatal("first build did not become ready")
+	}
+
+	// Hold the digest's queue slot as a build in its tail would: terminal
+	// signal sent, slot not yet released.
+	slotHeld := make(chan struct{})
+	releaseSlot := make(chan struct{})
+	m.queue.Enqueue(digestStr, CreateImageRequest{}, func() {
+		close(slotHeld)
+		<-releaseSlot
+	})
+	select {
+	case <-slotHeld:
+	case <-time.After(5 * time.Second):
+		t.Fatal("queue slot was not held")
+	}
+
+	// Delete and recreate while the slot is held: the recreate must queue
+	// behind the in-flight entry instead of being dropped.
+	require.NoError(t, m.DeleteImage(ctx, repo+":"+tag))
+	recreated, err := m.ImportLocalImage(ctx, repo, tag, digestStr)
+	require.NoError(t, err)
+	require.Equal(t, StatusPending, recreated.Status)
+	require.NotNil(t, recreated.QueuePosition)
+	close(releaseSlot)
+
+	// The recreated build must run and signal ready; before the fix it was
+	// swallowed by the still-active queue entry and never started.
+	select {
+	case ev := <-events:
+		require.Equal(t, StatusReady, ev.Status)
+	case <-time.After(10 * time.Second):
+		t.Fatal("recreated image never became ready")
+	}
+
+	// The ready signal precedes the tag symlink, so poll rather than GET once.
+	waitForReady(t, m, ctx, repo+":"+tag)
+}

lib/images/queue.go

@@ -28,16 +28,15 @@ func NewBuildQueue(maxConcurrent int) *BuildQueue {
 }
 
 // Enqueue adds a build to the queue. Returns queue position (0 if started immediately, >0 if queued).
-// If the image is already building or queued, returns its current position without re-enqueueing.
+// If the image is already queued, returns its current position without re-enqueueing.
+// If the image is still marked active, the build is queued behind the in-flight one
+// rather than dropped: the active build may have signalled readiness without having
+// released its slot yet, and dropping a fresh request here leaves recreated images
+// pending forever.
 func (q *BuildQueue) Enqueue(imageName string, req CreateImageRequest, startFn func()) int {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	// Check if already building (position 0, actively running)
-	if q.active[imageName] {
-		return 0
-	}
-
 	// Check if already in pending queue
 	for i, build := range q.pending {
 		if build.ImageName == imageName {
@@ -57,7 +56,7 @@ func (q *BuildQueue) Enqueue(imageName string, req CreateImageRequest, startFn f
 		StartFn:   wrappedFn,
 	}
 
-	if len(q.active) < q.maxConcurrent {
+	if len(q.active) < q.maxConcurrent && !q.active[imageName] {
 		q.active[imageName] = true
 		go wrappedFn()
 		return 0
@@ -85,10 +84,8 @@ func (q *BuildQueue) GetPosition(imageName string) *int {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
-	if q.active[imageName] {
-		return nil
-	}
-
+	// An image can be active (earlier build finishing) and pending at once;
+	// report the pending position, which is what a waiter observes.
 	for i, build := range q.pending {
 		if build.ImageName == imageName {
 			pos := i + 1

lib/images/queue_test.go

@@ -0,0 +1,80 @@
+package images
+
+import (
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestEnqueueWhileActiveQueuesBehindInFlightBuild covers the window between a
+// build's terminal signal and its queue-slot release: a fresh request for the
+// same image landing in that window must queue behind the in-flight build, not
+// be dropped. Dropping it leaves recreated images pending forever.
+func TestEnqueueWhileActiveQueuesBehindInFlightBuild(t *testing.T) {
+	q := NewBuildQueue(1)
+
+	started := make(chan struct{})
+	release := make(chan struct{})
+
+	pos := q.Enqueue("sha256:abc", CreateImageRequest{}, func() {
+		close(started)
+		<-release
+	})
+	require.Equal(t, 0, pos)
+	<-started
+
+	// First build is in its tail: still holding the active slot.
+	var secondRuns atomic.Int32
+	pos = q.Enqueue("sha256:abc", CreateImageRequest{}, func() {
+		secondRuns.Add(1)
+	})
+	assert.Equal(t, 1, pos, "second request should queue behind the in-flight build")
+	assert.Equal(t, 1, q.PendingCount())
+
+	// A duplicate request returns the existing pending position.
+	pos = q.Enqueue("sha256:abc", CreateImageRequest{}, func() {
+		secondRuns.Add(1)
+	})
+	assert.Equal(t, 1, pos)
+	assert.Equal(t, 1, q.PendingCount())
+
+	// While active and pending at once, GetPosition reports the pending slot.
+	queuePos := q.GetPosition("sha256:abc")
+	require.NotNil(t, queuePos)
+	assert.Equal(t, 1, *queuePos)
+
+	close(release)
+	require.Eventually(t, func() bool {
+		return secondRuns.Load() == 1 && q.QueueLength() == 0
+	}, 5*time.Second, time.Millisecond, "queued build must run after the in-flight build completes")
+}
+
+// TestEnqueueStartsImmediatelyAfterSlotFrees verifies the normal path is
+// unchanged: once the in-flight build fully completes, a new request starts
+// right away instead of queueing.
+func TestEnqueueStartsImmediatelyAfterSlotFrees(t *testing.T) {
+	q := NewBuildQueue(1)
+
+	done := make(chan struct{})
+	pos := q.Enqueue("sha256:abc", CreateImageRequest{}, func() {
+		close(done)
+	})
+	require.Equal(t, 0, pos)
+	<-done
+
+	require.Eventually(t, func() bool {
+		return q.QueueLength() == 0
+	}, 5*time.Second, time.Millisecond)
+
+	var ran atomic.Bool
+	pos = q.Enqueue("sha256:abc", CreateImageRequest{}, func() {
+		ran.Store(true)
+	})
+	assert.Equal(t, 0, pos)
+	require.Eventually(t, func() bool {
+		return ran.Load()
+	}, 5*time.Second, time.Millisecond)
+}