diff --git a/task_buffer.go b/task_buffer.go index 8b4ba02..7f178e0 100644 --- a/task_buffer.go +++ b/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-- diff --git a/task_buffer_test.go b/task_buffer_test.go index db263b0..a86c2e5 100644 --- a/task_buffer_test.go +++ b/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}) }) }