fix: preserve FIFO order in overflow buffer

task_buffer.go

@@ -54,9 +54,9 @@ func (b *chunkedTaskBuffer) Close() {
 	b.overflowClosed = true
 }
 
-// PushAndForward appends task to the buffer, then tries to forward one
-// buffered task through tryForward. If forwarding fails, the task is pushed
-// back to the buffer tail.
+// PushAndForward appends task to the buffer, then tries to forward the oldest
+// buffered task through tryForward. A failed forward leaves that task at the
+// head, preserving FIFO order while the handoff channel is saturated.
 func (b *chunkedTaskBuffer) PushAndForward(task Task, tryForward func(Task) bool) taskBufferPushResult {
 	b.taskMu.Lock()
 	defer b.taskMu.Unlock()
@@ -70,15 +70,50 @@ func (b *chunkedTaskBuffer) PushAndForward(task Task, tryForward func(Task) bool
 
 	b.pushTail(task)
 
-	if t, ok := b.popHead(); ok {
-		if !tryForward(t) {
-			b.pushTail(t)
-		}
+	if t, ok := b.peekHead(); ok && tryForward(t) {
+		// popHead cannot fail here: taskMu is held and peekHead just observed
+		// the same head task. Remove only after a successful handoff so a
+		// failed handoff never moves an older task behind newer submissions.
+		b.popHead()
 	}
 
 	return taskBufferAccepted
 }
 
+// peekHead returns the task at the head of the buffer without removing it.
+// Must be called with taskMu held.
+func (b *chunkedTaskBuffer) peekHead() (Task, bool) {
+	if b.headChunk == nil {
+		return nil, false
+	}
+	if b.headIdx >= taskChunkSize {
+		// Advance to next chunk; recycle the exhausted one.
+		next := b.headChunk.next
+		b.headChunk.next = nil
+		b.chunkPool.Put(b.headChunk)
+		b.headChunk = next
+		b.headIdx = 0
+		if b.headChunk == nil {
+			b.tailChunk = nil
+			b.tailIdx = 0
+			return nil, false
+		}
+	}
+
+	// Head caught up to tail in the same chunk - queue drained.
+	if b.headChunk == b.tailChunk && b.headIdx >= b.tailIdx {
+		b.headChunk.next = nil
+		b.chunkPool.Put(b.headChunk)
+		b.headChunk = nil
+		b.tailChunk = nil
+		b.headIdx = 0
+		b.tailIdx = 0
+		return nil, false
+	}
+
+	return b.headChunk.tasks[b.headIdx], true
+}
+
 func (b *chunkedTaskBuffer) PopBatch(dst []Task) int {
 	b.taskMu.Lock()
 	defer b.taskMu.Unlock()
@@ -121,35 +156,10 @@ func (b *chunkedTaskBuffer) pushTail(task Task) {
 // Returns nil, false if the buffer is empty.
 // Must be called with taskMu held.
 func (b *chunkedTaskBuffer) popHead() (Task, bool) {
-	if b.headChunk == nil {
+	task, ok := b.peekHead()
+	if !ok {
 		return nil, false
 	}
-	if b.headIdx >= taskChunkSize {
-		// Advance to next chunk; recycle the exhausted one.
-		next := b.headChunk.next
-		b.headChunk.next = nil
-		b.chunkPool.Put(b.headChunk)
-		b.headChunk = next
-		b.headIdx = 0
-		if b.headChunk == nil {
-			b.tailChunk = nil
-			b.tailIdx = 0
-			return nil, false
-		}
-	}
-
-	// Head caught up to tail in the same chunk - queue drained.
-	if b.headChunk == b.tailChunk && b.headIdx >= b.tailIdx {
-		b.headChunk.next = nil
-		b.chunkPool.Put(b.headChunk)
-		b.headChunk = nil
-		b.tailChunk = nil
-		b.headIdx = 0
-		b.tailIdx = 0
-		return nil, false
-	}
-
-	task := b.headChunk.tasks[b.headIdx]
 	b.headChunk.tasks[b.headIdx] = nil // help GC
 	b.headIdx++
 	b.chunkLen--

task_buffer_test.go

@@ -84,33 +84,6 @@ func assertTaskIDs(tb testing.TB, tasks []Task, want []int) {
 	}
 }
 
-// assertTaskIDSet checks task conservation without making order part of the
-// contract. It is used when a failed forward is placed back into the buffer.
-func assertTaskIDSet(tb testing.TB, tasks []Task, want ...int) {
-	tb.Helper()
-
-	if len(tasks) != len(want) {
-		tb.Fatalf("got %d tasks, want %d", len(tasks), len(want))
-	}
-
-	wantCounts := make(map[int]int, len(want))
-	for _, id := range want {
-		wantCounts[id]++
-	}
-	for _, task := range tasks {
-		id := bufferTaskID(tb, task)
-		if wantCounts[id] == 0 {
-			tb.Fatalf("unexpected or duplicate task id %d", id)
-		}
-		wantCounts[id]--
-	}
-	for id, count := range wantCounts {
-		if count != 0 {
-			tb.Fatalf("task id %d is missing %d occurrence(s)", id, count)
-		}
-	}
-}
-
 // TestChunkedTaskBufferPushAndForward covers the successful fast path and the
 // failed-forward path, both with and without an existing backlog.
 func TestChunkedTaskBufferPushAndForward(t *testing.T) {
@@ -192,7 +165,7 @@ func TestChunkedTaskBufferPushAndForward(t *testing.T) {
 		assertTaskIDs(t, drainTaskBuffer(buffer), []int{2, 3})
 	})
 
-	t.Run("backlog forward failure preserves every task", func(t *testing.T) {
+	t.Run("backlog forward failure preserves FIFO order", func(t *testing.T) {
 		buffer := newChunkedTaskBuffer()
 		fillTaskBuffer(t, buffer, 1, 2)
 		callbackCalls := 0
@@ -213,9 +186,9 @@ func TestChunkedTaskBufferPushAndForward(t *testing.T) {
 		if forwardedID != 1 {
 			t.Fatalf("forwarded task id = %d, want oldest task 1", forwardedID)
 		}
-		// A failed task must remain available, but its exact requeue position is
-		// an implementation detail and is intentionally not asserted here.
-		assertTaskIDSet(t, drainTaskBuffer(buffer), 1, 2, 3)
+		// taskQueue is still full, so the oldest task must stay at the head
+		// rather than being requeued behind newer tasks.
+		assertTaskIDs(t, drainTaskBuffer(buffer), []int{1, 2, 3})
 	})
 }