fix: release gate reservation on retry to prevent queue wedge (#311) (#334)

* fix: release gate reservation on retry to prevent queue wedge (#311)

The queue-gate reservation stored in RedisSortedSetFlow.activeReleases was
released only by resultWorker (terminal results). The retry path (retryWorker)
re-enqueued the request without releasing, so every retry leaked a reservation:
inFlight ratcheted up until Budget() reached 0 and the queue stopped dispatching
entirely (LocalConcurrencyGate, no TTL), or the redis-quota counter over-admitted
when its key TTL reset while requests were still in flight.

- retryWorker now releases each retried request's reservation before re-enqueue
  (a re-dispatched request re-reserves via gate.Apply), ordered before
  flushRetryBatch's ZAdd so the reservation cannot be overwritten/orphaned.
- Dispatch-time store uses sync.Map.Swap and releases any prior closure
  defensively, so a lingering reservation is never orphaned.
- Add a regression test for the retry->release path (previously untested, which
  is why the leak stayed latent).

Pub/Sub is unaffected: it holds the reservation in a function-scoped defer that
runs on every return path including Nack/retry.

Closes #311

Signed-off-by: Shimi Bandiel <shimib@google.com>

* docs(release-notes): add fragment for #334 (gate reservation leak fix)

Signed-off-by: Shimi Bandiel <shimib@google.com>

---------

Signed-off-by: Shimi Bandiel <shimib@google.com>

Author: shimib

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)
 	}
 

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()

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).