diff --git a/pkg/redis/sortedset_impl.go b/pkg/redis/sortedset_impl.go index 277b1d49..d34c472f 100644 --- a/pkg/redis/sortedset_impl.go +++ b/pkg/redis/sortedset_impl.go @@ -524,7 +524,13 @@ func (r *RedisSortedSetFlow) processMessages(ctx context.Context, msgChannel cha } if len(releases) > 0 { - r.activeReleases.Store(rview.ReqID(), releases) + // Defensive: never orphan a lingering reservation for this id — release + // any prior closure instead of silently overwriting it (see #311). + if prev, loaded := r.activeReleases.Swap(rview.ReqID(), releases); loaded { + if rels, ok := prev.([]pipeline.GateReleaseFunc); ok { + pipeline.ReleaseGateReleases(rels) + } + } } // Stamp ingestion time as the message enters the in-process buffer so the @@ -572,6 +578,23 @@ func (r *RedisSortedSetFlow) parseMessage(z redis.Z, logger logr.Logger) (*api.I func (r *RedisSortedSetFlow) retryWorker(ctx context.Context) { processMsg := func(processCtx context.Context, msg pipeline.RetryMessage) { batch := drainBatch(msg, r.retryChannel, maxBatchSize) + // #311: a retried request returns to the queue and is re-gated (and thus + // re-reserved) on its next dispatch, so its current gate reservation must + // be released here — otherwise inFlight ratchets up on every retry until + // Budget() reaches 0 and the queue stops dispatching entirely. Release + // BEFORE re-enqueue: a re-dispatch can only occur after flushRetryBatch's + // ZAdd, so releasing first prevents the reservation from being overwritten + // and orphaned. + for _, m := range batch { + if m.InternalRequest == nil || m.PublicRequest == nil { + continue + } + if val, ok := r.activeReleases.LoadAndDelete(m.PublicRequest.ReqID()); ok { + if rels, ok := val.([]pipeline.GateReleaseFunc); ok { + pipeline.ReleaseGateReleases(rels) + } + } + } r.flushRetryBatch(processCtx, batch) } diff --git a/pkg/redis/sortedset_impl_test.go b/pkg/redis/sortedset_impl_test.go index 0775b7df..e6f45a4e 100644 --- a/pkg/redis/sortedset_impl_test.go +++ b/pkg/redis/sortedset_impl_test.go @@ -495,6 +495,59 @@ func TestSortedSetFlow_RetryBackoff(t *testing.T) { } } +// TestSortedSetFlow_RetryReleasesReservation is a regression test for #311: a +// retried request returns to the queue and is re-gated (re-reserved) on its next +// dispatch, so the retry path must release its current queue-gate reservation. +// Before the fix, only resultWorker released reservations, so every retry leaked +// one — inFlight ratcheted up until Budget() hit 0 and the queue wedged. +func TestSortedSetFlow_RetryReleasesReservation(t *testing.T) { + s, rdb, ctx, cancel := setupTest(t) + defer s.Close() + defer rdb.Close() // nolint:errcheck + defer cancel() + + const reqID = "retry-release-1" + queue := "retry-release-queue" + + flow := &RedisSortedSetFlow{ + rdb: rdb, + retryChannel: make(chan pipeline.RetryMessage, 1), + pollInterval: 50 * time.Millisecond, + batchSize: 10, + gate: noopGate(), + } + + // Simulate a dispatched request holding a gate reservation (as the dequeue + // loop stores after gate.Apply). + releasedCh := make(chan struct{}, 1) + flow.activeReleases.Store(reqID, []pipeline.GateReleaseFunc{ + func() { releasedCh <- struct{}{} }, + }) + + go flow.retryWorker(ctx) + + flow.retryChannel <- pipeline.RetryMessage{ + EmbelishedRequestMessage: pipeline.EmbelishedRequestMessage{ + InternalRequest: api.NewInternalRequest( + api.InternalRouting{RetryCount: 1, RequestQueueName: queue}, + &api.RequestMessage{ID: reqID, Created: time.Now().Unix(), Deadline: 9999999999}, + ), + }, + BackoffDurationSeconds: 0, + } + + select { + case <-releasedCh: + // Reservation released on the retry path — correct. + case <-time.After(2 * time.Second): + t.Fatal("retry did not release the queue-gate reservation (#311 leak)") + } + + if _, held := flow.activeReleases.Load(reqID); held { + t.Error("reservation still present in activeReleases after retry") + } +} + func TestSortedSetFlow_ResultFIFO(t *testing.T) { s, rdb, ctx, cancel := setupTest(t) defer s.Close() diff --git a/release-notes.d/unreleased/334.md b/release-notes.d/unreleased/334.md new file mode 100644 index 00000000..613d002a --- /dev/null +++ b/release-notes.d/unreleased/334.md @@ -0,0 +1,7 @@ +--- +pr: 334 +url: https://github.com/llm-d/llm-d-async/pull/334 +author: shimib +date: 2026-07-21 +--- +Fixed a gate capacity reservation leak in the Redis sorted-set flow: retried requests never released their per-queue gate reservation, so `inFlight` ratcheted up on every retry until the queue stopped dispatching entirely (`local-max-concurrency`) or the `redis-quota` counter over-admitted on TTL reset. Retries now release the reservation before re-enqueue (and re-reserve on re-dispatch).