fix(redis): enforce retry backoff via the retry queue

pkg/redis/options.go

@@ -79,7 +79,11 @@ func (c *PubSubConfig) Validate() error {
 // SortedSetConfig is the transport config for the Redis sorted-set flow.
 // It is parsed from JSON provided via --transport-config or --transport-config-file.
 type SortedSetConfig struct {
-	URL             string                 `json:"url,omitempty"`
+	URL string `json:"url,omitempty"`
+	// RetryQueueName is the sorted set holding backoff retries, scored by
+	// retry-due time. Retries re-enter their request queue (with the original
+	// deadline score) only once due, so backoff is actually enforced.
+	RetryQueueName  string                 `json:"retry_queue_name,omitempty"`
 	ResultQueueName string                 `json:"result_queue_name,omitempty"`
 	PollIntervalMs  int                    `json:"poll_interval_ms,omitempty"`
 	BatchSize       int                    `json:"batch_size,omitempty"`
@@ -111,6 +115,9 @@ func (c *SortedSetConfig) ApplyEnvOverrides() {
 }
 
 func (c *SortedSetConfig) ApplyDefaults() {
+	if c.RetryQueueName == "" {
+		c.RetryQueueName = "retry-sortedset"
+	}
 	if c.ResultQueueName == "" {
 		c.ResultQueueName = "result-list"
 	}

pkg/redis/sortedset_impl.go

@@ -67,6 +67,7 @@ type RedisSortedSetFlow struct {
 	resultChannel           chan api.ResultMessage
 	pollInterval            time.Duration
 	batchSize               int
+	retryQueueName          string
 	activeReleases          sync.Map
 	gate                    pipeline.Gate
 	gateFactory             pipeline.GateFactory
@@ -116,6 +117,7 @@ func NewRedisSortedSetFlow(cfg SortedSetConfig, workerPools []pipeline.WorkerPoo
 		resultChannel:          make(chan api.ResultMessage, resultChannelBuffer),
 		pollInterval:           time.Duration(cfg.PollIntervalMs) * time.Millisecond,
 		batchSize:              cfg.BatchSize,
+		retryQueueName:         cfg.RetryQueueName,
 		defaultResultQueueName: cfg.ResultQueueName,
 		workerPools:            workerPools,
 		gateFactory:            gateFactory,
@@ -226,6 +228,9 @@ func (r *RedisSortedSetFlow) Start(ctx context.Context) {
 			r.requestWorker(consumeCtx, ch.channel.Channel, ch.queueName, ch.queueID)
 		}(ch)
 	}
+	r.consumeWg.Add(1)
+	go func() { defer r.consumeWg.Done(); r.retryMover(consumeCtx) }()
+
 	r.drainWg.Add(2)
 	go func() { defer r.drainWg.Done(); r.retryWorker(drainCtx) }()  // #nosec G118
 	go func() { defer r.drainWg.Done(); r.resultWorker(drainCtx) }() // #nosec G118
@@ -575,15 +580,24 @@ func (r *RedisSortedSetFlow) flushRetryBatch(ctx context.Context, batch []pipeli
 		if queueName == "" {
 			queueName = r.defaultRequestQueueName
 		}
+		// Preserve the origin queue in the envelope so the retry mover can
+		// re-enter the message into the right queue once it is due.
+		msg.RequestQueueName = queueName
 		bytes, err := json.Marshal(msg.InternalRequest)
 		if err != nil {
 			logger.V(logutil.DEFAULT).Error(err, "Failed to marshal retry")
 			continue
 		}
 
+		// Score is the retry-due time. The retry queue is drained by the
+		// retry mover strictly at or after this time, so the backoff is
+		// enforced. Retries must NOT be ZADDed into the request queue
+		// directly: there the score means deadline and ZPopMin would pop a
+		// future-scored retry immediately (and ahead of all fresh traffic,
+		// since now+backoff sorts below any realistic deadline).
 		retryScore := float64(time.Now().Unix()) + msg.BackoffDurationSeconds
 		entries = append(entries, retryEntry{
-			queue: queueName,
+			queue: r.retryQueue(),
 			value: redis.Z{Score: retryScore, Member: string(bytes)},
 		})
 	}
@@ -691,3 +705,63 @@ func (r *RedisSortedSetFlow) marshalResult(msg api.ResultMessage) string {
 	fallbackBytes, _ := json.Marshal(fallback)
 	return string(fallbackBytes)
 }
+
+// retryQueue returns the retry queue name, defaulting for flows constructed
+// directly (tests) without ApplyDefaults.
+func (r *RedisSortedSetFlow) retryQueue() string {
+	if r.retryQueueName == "" {
+		return "retry-sortedset"
+	}
+	return r.retryQueueName
+}
+
+// retryMover re-enters due retries into their request queues. Retries wait in
+// the retry queue scored by retry-due time; once due, they return to their
+// origin queue with the message's original deadline as the score, restoring
+// earliest-deadline-first ordering among fresh traffic.
+func (r *RedisSortedSetFlow) retryMover(ctx context.Context) {
+	logger := log.FromContext(ctx)
+	ticker := time.NewTicker(r.pollInterval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			now := float64(time.Now().Unix())
+			members, err := r.rdb.ZRangeArgs(ctx, redis.ZRangeArgs{
+				Key: r.retryQueue(), ByScore: true,
+				Start: "-inf", Stop: fmt.Sprintf("%f", now),
+				Count: int64(r.batchSize), Offset: 0,
+			}).Result()
+			if err != nil || len(members) == 0 {
+				continue
+			}
+			for _, member := range members {
+				// ZRem guards against double-move: only the remover that wins
+				// the removal re-enters the message.
+				removed, err := r.rdb.ZRem(ctx, r.retryQueue(), member).Result()
+				if err != nil || removed == 0 {
+					continue
+				}
+				var ir api.InternalRequest
+				if err := json.Unmarshal([]byte(member), &ir); err != nil || ir.PublicRequest == nil {
+					logger.V(logutil.DEFAULT).Error(err, "Failed to parse due retry, dropping", "member", member[:min(len(member), 120)])
+					continue
+				}
+				queueName := ir.RequestQueueName
+				if queueName == "" {
+					queueName = r.defaultRequestQueueName
+				}
+				if err := r.rdb.ZAdd(ctx, queueName, redis.Z{
+					Score:  float64(ir.PublicRequest.ReqDeadline()),
+					Member: member,
+				}).Err(); err != nil {
+					logger.V(logutil.DEFAULT).Error(err, "Failed to re-enter due retry", "queue", queueName)
+					// Put it back in the retry queue so it is not lost.
+					_ = r.rdb.ZAdd(ctx, r.retryQueue(), redis.Z{Score: now, Member: member}).Err()
+				}
+			}
+		}
+	}
+}

pkg/redis/sortedset_impl_test.go

@@ -489,14 +489,42 @@ func TestSortedSetFlow_RetryBackoff(t *testing.T) {
 	flow.retryChannel <- retryMsg
 	time.Sleep(100 * time.Millisecond)
 
-	results, _ := rdb.ZRangeWithScores(ctx, queue, 0, -1).Result()
+	// The retry parks in the retry queue scored by its due time, NOT in the
+	// request queue (where the score means deadline and ZPopMin would pop a
+	// future-scored retry immediately).
+	if n, _ := rdb.ZCard(ctx, queue).Result(); n != 0 {
+		t.Fatalf("Expected request queue empty before backoff elapses, got %d entries", n)
+	}
+	results, _ := rdb.ZRangeWithScores(ctx, flow.retryQueue(), 0, -1).Result()
 	if len(results) != 1 {
-		t.Fatalf("Expected 1 retry message, got %d", len(results))
+		t.Fatalf("Expected 1 message in retry queue, got %d", len(results))
 	}
-
 	expectedScore := float64(time.Now().Unix()) + 2.0
 	if results[0].Score < expectedScore-1 || results[0].Score > expectedScore+1 {
-		t.Errorf("Retry score incorrect: expected ~%f, got %f", expectedScore, results[0].Score)
+		t.Errorf("Retry due-time score incorrect: expected ~%f, got %f", expectedScore, results[0].Score)
+	}
+
+	// Once due, the mover re-enters it into the request queue with the
+	// original deadline as the score.
+	go flow.retryMover(ctx)
+	s.FastForward(3 * time.Second)
+	deadlineCheck := time.After(3 * time.Second)
+	for {
+		entries, _ := rdb.ZRangeWithScores(ctx, queue, 0, -1).Result()
+		if len(entries) == 1 {
+			if entries[0].Score != 9999999999 {
+				t.Errorf("Re-entered retry should carry deadline score, got %f", entries[0].Score)
+			}
+			if n, _ := rdb.ZCard(ctx, flow.retryQueue()).Result(); n != 0 {
+				t.Errorf("Retry queue should be empty after move, got %d", n)
+			}
+			break
+		}
+		select {
+		case <-deadlineCheck:
+			t.Fatal("timed out waiting for due retry to re-enter request queue")
+		case <-time.After(20 * time.Millisecond):
+		}
 	}
 }
 
@@ -1192,7 +1220,7 @@ func TestSortedSetFlow_RetryWorkerDrainsOnShutdown(t *testing.T) {
 		t.Fatal("retryWorker did not stop after context cancellation")
 	}
 
-	count, err := rdb.ZCard(ctx, queue).Result()
+	count, err := rdb.ZCard(ctx, flow.retryQueue()).Result()
 	if err != nil {
 		t.Fatalf("ZCard error: %v", err)
 	}
@@ -1242,7 +1270,7 @@ func TestSortedSetFlow_RetryBatchAfterFailure(t *testing.T) {
 
 	deadline := time.Now().Add(2 * time.Second)
 	for time.Now().Before(deadline) {
-		count, err := rdb.ZCard(ctx, queue).Result()
+		count, err := rdb.ZCard(ctx, flow.retryQueue()).Result()
 		if err == nil && count == 1 {
 			return
 		}
@@ -1679,10 +1707,11 @@ func TestNewRedisSortedSetFlow_DefaultsWorkerPoolIDInConfigMap(t *testing.T) {
 }
 
 // TestNewRedisSortedSetFlow_RetryFallsBackToFirstQueue is a regression test: a
-// retry message that carries no RequestQueueName must be re-enqueued to a real
+// retry message that carries no RequestQueueName must be destined for a real
 // queue key, not "". NewRedisSortedSetFlow seeds defaultRequestQueueName from
-// the first configured queue; without that seed flushRetryBatch ZADDs the retry
-// to the empty key "" and the message is lost.
+// the first configured queue; without that seed flushRetryBatch would stamp ""
+// into the parked retry's envelope and the mover would re-enter it on the
+// empty key, losing the message.
 func TestNewRedisSortedSetFlow_RetryFallsBackToFirstQueue(t *testing.T) {
 	s := miniredis.RunT(t)
 	defer s.Close()
@@ -1717,8 +1746,18 @@ func TestNewRedisSortedSetFlow_RetryFallsBackToFirstQueue(t *testing.T) {
 
 	flow.flushRetryBatch(context.Background(), []pipeline.RetryMessage{retryMsg})
 
-	if n, _ := flow.rdb.ZCard(context.Background(), primaryQueue).Result(); n != 1 {
-		t.Errorf("Expected retry re-enqueued to %q (ZCARD=1), got ZCARD=%d", primaryQueue, n)
+	// The retry parks in the retry queue with the seeded queue name stamped
+	// into its envelope, so the mover re-enters it on a real key.
+	members, _ := flow.rdb.ZRange(context.Background(), flow.retryQueue(), 0, -1).Result()
+	if len(members) != 1 {
+		t.Fatalf("Expected 1 parked retry in %q, got %d", flow.retryQueue(), len(members))
+	}
+	var ir api.InternalRequest
+	if err := json.Unmarshal([]byte(members[0]), &ir); err != nil {
+		t.Fatalf("Failed to parse parked retry: %v", err)
+	}
+	if ir.RequestQueueName != primaryQueue {
+		t.Errorf("Parked retry envelope queue = %q, want %q", ir.RequestQueueName, primaryQueue)
 	}
 	if n, _ := flow.rdb.ZCard(context.Background(), "").Result(); n != 0 {
 		t.Errorf("Retry landed on the empty key \"\" (ZCARD=%d); default request queue was not seeded", n)

release-notes.d/unreleased/390.md

@@ -0,0 +1,10 @@
+---
+pr: 390
+url: https://github.com/llm-d/llm-d-async/pull/390
+author: BenjaminBraunDev
+date: 2026-08-04
+---
+Sorted-set retries now honor their exponential backoff by parking in the retry
+queue (`retry_queue_name` in the sorted-set transport config, default
+`retry-sortedset`) until due, instead of immediately re-entering the request
+queue ahead of fresh traffic.