feat(redis): durable dequeue via claims for the sorted-set transport

[Abhinav-kodes]

What does this PR do?

The redis-sortedset transport now dequeues with a peek -> claim -> ack flow instead of a destructive ZPOPMIN:

• Atomic Lua claim: Moves a request out of the pending set under a lease (ownership token plus expiry index) before dispatch.
• Atomic ack & token fencing: Validates that the active lease owner token matches, pushes the result, and drops the claim atomically. Stale or expired owners whose lease has lapsed are fenced and cannot publish.
• Background reclaimer: Redelivers claims whose lease has expired back to the pending queue.
• Heartbeater: Periodically renews held leases every TTL / 3 (clamped between 1s and 30s) with token fencing. If renewal is rejected (e.g. lease lapsed or claimed elsewhere), the local handle is deleted immediately.
• ID reuse handling: In-memory claim handles are keyed by ID + RequestToken, isolating generation state and preventing old results from touching a new generation's claim.
• Unified shutdown & retry handling: Graceful shutdown relies on lease expiry without requiring a separate destructive sweep. Requests waiting in the retry queue retain their lease via fenced renewals.
• Sustained Redis outage handling: If Redis operations exhaust retries during a sustained outage on result/retry flush, the local claim handle is dropped so the claim can be redelivered via lease expiry once Redis recovers.
• Configuration: Tuned via transport-config JSON using claim_lease_ttl_seconds (default: 300, serves as crash-detection window with periodic heartbeats) and claim_reclaim_interval_ms (default: 15000), with non-negative validation.
• Compatibility: No producer or API changes. The pending member format remains unchanged, making rolling upgrades safe.

---

Why is this change needed?

Fixes #404. Dequeue was previously destructive (ZPOPMIN) and lacked an in-flight recovery mechanism. Between the initial pop and result push, requests existed solely in process memory. A SIGKILL, OOM crash, or node failure silently dropped all accepted requests held by that worker.

This change establishes at-least-once execution:

• Dequeue operates under a revocable lease; unacked requests are redelivered by surviving instances when a lease expires.
• Result publication is protected by atomic per-claim token fencing (owners[id]==claimToken), preventing stale/expired workers from publishing.
• The durability model, trade-offs, and Redis persistence requirements (AOF/replication) are documented in docs/guides/durable-dequeue.md.

---

How was this tested?

• Unit tests:
  • Claim races, stale tokens, token fencing, reclaim logic, heartbeats, and negative config validations (pkg/redis/claim_test.go, pkg/redis/options_test.go).
  • End-to-end sustained Redis outage recovery for both result and retry flush paths (TestSortedSetFlow_ResultSustainedOutage_DropsClaimHandle, TestSortedSetFlow_RetrySustainedOutage_DropsClaimHandle in pkg/redis/sortedset_impl_test.go).
• Integration tests:
  • Verified lease expiry and takeover after instance stops heartbeating (TestLeaseExpiry_TakeoverRedeliversClaims in test/integration/claim_expiry_loss_test.go).
• All tests pass with race detection enabled (go test -race).

---

Checklist

• Commits are signed off with git commit -s per DCO
• Code follows project contributing guidelines
• Tests pass locally with make test
• Linters pass with make lint
• Documentation is updated

---

Related Issues

• Fixes: [Bug]: Accepted Redis requests can be lost when an Async pod is hard-killed #404
• Related:
  • [Bug]: Async results can be lost with multiple Batch Processor replicas llm-d-batch-gateway#644
  • [Feature]: Resume in-progress batches after Batch Processor pod or node loss llm-d-batch-gateway#645
  • [Feature]: Add shared Batch Gateway + llm-d Async end-to-end coverage llm-d-batch-gateway#657 (shared durability contract)

---

Release Note

The redis-sortedset transport no longer loses accepted requests when a processor crashes or restarts. Dequeue now uses a lease claim model, allowing surviving instances to redeliver work after lease expiration while fencing stale owners from publishing results.

[shimib]

Thanks for working on this! The lease-claiming architecture is a great addition to the durability story.

I've performed a deep dive into the implementation and verified the mechanics against our test suites. I found a few critical race/data-loss issues (items 1–5) and some minor regressions/nits.

I've pushed a fully tested reference branch with surgical fixes for the blocking items (1–4). You can pull or inspect the clean commits directly here: https://github.com/shimib/llm-d-async/tree/fix/claim-issues

---

🚨 Blocking / Critical Issues (Requesting Changes)

1. The global dedup marker breaks legitimate request-ID reuse (Silent Result Loss)

• Location: pkg/redis/claim.go:59-61
• Issue: terminalKey is result-terminal:<request ID>, and IDs are producer-supplied. The repo has an explicit contract that request IDs may be safely reused across submissions (e.g., redis_sortedset_producer.go:206-208 clears stale cancellation markers precisely for this reason). With the current PR, a second submission reusing an ID within resultDedupTTL (6h) has its terminal result suppressed as a "duplicate" in Lua, causing the client to poll forever.
• Fix: Key the marker on ID + RequestToken (result.Routing.RequestToken is freshly random per enqueue, and redelivered copies carry the same token JSON so crash dedup still works). If empty, fall back to ID. Also namespace the key per queue to prevent cross-queue collisions.

2. Heartbeat renews foreign leases indefinitely (Deadlock on Crash)

• Location: pkg/redis/claim.go:156-164
• Issue: RENEWCLAIM is not token-guarded. If Instance A fails/parks a request to retry, it keeps the ID in its local claimTokens map. The retry flusher moves it back to pending, and Instance B claims it (overwriting the owner token in Redis). A's heartbeater continues to call renewClaim every tick. Because the script does not check the owner token, A's stale heartbeats succeed, constantly updating the index. If B crashes mid-flight, the lease never lapses because healthy A is renewing it indefinitely, wedging the request until the deadline grace finally terminates it.
• Fix: Token-guard RENEWCLAIM (if HGET owners[id] ~= token then return -1), and have heartbeatClaims delete the local claimTokens and retryOwned handles upon receiving a mismatch (-1), preventing both deadlocks and memory leaks.

3. retryOwned is never cleared when a retried request is re-claimed

• Location: pkg/redis/claim.go:201-222
• Issue: claimRequest stores the new handle but doesn't delete retryOwned. If a request is claimed again on the same instance after a retry cycle, it remains marked as retryOwned. On graceful shutdown, sweepUnackedClaims will skip sweeping it, forcing it to wait out its full 5-minute lease on Redis before redelivery.
• Fix: Delete the ID from r.retryOwned upon successful claimRequest.

4. async_claim_depth reads double

• Location: pkg/redis/claim.go:326
• Issue: The claimed hash holds two fields per request (<id> and <id>:score), so HLen reports double the real claim count.
• Fix: Use ZCard(keys.idx) which accurately tracks unique active claims.

5. The deadline+30s lease cap races real results against DEADLINE_EXCEEDED

• Location: pkg/redis/claim.go:189-196
• Issue: Once a running request goes 30s past its original deadline, the heartbeat can no longer renew the lease (expiry is capped in the past). The reclaimer sweeps it and pushes a DEADLINE_EXCEEDED terminal record. The real result (which may still complete successfully before the worker request timeout) is then suppressed as a duplicate. On main, late successes were safely delivered.
• Fix: Set reclaimGraceAfterDeadline to comfortably exceed the worker's request timeout, or exempt active, in-flight (non-retryOwned) renewals from the deadline cap.

---

⚠️ Non-Blocking & Follow-ups

6. Double pending copies on retry queue crash

If a crash occurs while a request sits in the retry queue, the reclaimer will redeliver the original claimed payload (its lease lapses) while the retry-queue copy independently re-enters pending via the flusher. The dedup marker bounds the damage, but this should be added to the tradeoffs section of durable-dequeue.md.

7. Transient cross-instance quota double-count

gate.Apply runs before claimRequest, so multiple instances peeking the same head entry will both reserve quota before one loses the claim and releases it. This is a defensible tradeoff to avoid claiming churn, but deserves a brief code comment indicating it is deliberate.

8. Result flushing batching regression

flushResultBatch went from a single pipelined LPUSH block to serial EVALSHA round-trips per result. Under batch-workloads, this serial execution introduces a heavy performance regression. Pipelining these script calls using a Redis Pipeliner is highly recommended.

9. Deadline-expired metric double-counts

In sortedset_impl.go, RecordExceededDeadlineReq and the "Deadline expired" log fire before the claim is won, meaning every peeking instance will double-count it. Move these statements after claimed == true.

10. Flag validation

In compat.go, durations are truncated to seconds (ClaimLeaseTTL / time.Second). Sub-second durations (e.g. 500ms) truncate to 0 and silently fallback to the 5m default. Negative durations also pass through unchecked. Durations should be validated to be $\ge 1s$ during options load.

11. Docs / Rolling upgrade

The claim that "rolling upgrade is safe" is true but worth a clarifying note: old instances still run ZPOPMIN destructively during the mixed-fleet window, meaning the old loss mode will persist until the upgrade completely finishes.

[hhzhang16]

I think this needs a smaller first slice. The hard-kill bug requires a claim, an owner-fenced heartbeat, an owner-fenced terminal ack, and expiry-based redelivery. This patch additionally introduces terminal-marker GC, a separate graceful-shutdown handback protocol, retry-specific ownership state, three tuning knobs on multiple config surfaces, and three new metrics. Those extra mechanisms are where several of the correctness problems are coming from.

Could we use RequestToken as the request-generation identity and the claim token as the ownership fence, then allow ACK to push a result only when the ownership token still matches? That removes the need for terminal marker and dedup TTL entirely as stale owners simply cannot publish. For the initial fix, graceful shutdown can stop heartbeating and let the same lease-recovery path handle unacked work. Immediate shutdown handback, additional tuning, and richer observability can follow in future MRs once the core hard-kill path is proven.

[hhzhang16]

The retry-queue write before this section of code has no failure path after retryRedisOp exhausts its attempts. The retry message is consumed, while its claim handle remains live and continues heartbeating, which can permanently orphan the request. Please stop renewal/release ownership on persistence failure, and test a sustained outage rather than only recovery within the three attempts.

[Abhinav-kodes]

Fixed in commit 5e1b4b2

When retryRedisOp exhausts all retry attempts in flushRetryBatch (or flushResultBatch), the error path now iterates through the entries and deletes their handles from r.claimTokens. This halts heartbeat renewals, allowing the claim lease to expire in Redis so that reclaimExpiredClaims can safely redeliver the request rather than leaving it permanently orphaned.

[hhzhang16]

This renewal needs to verify the ownership token. After lease expiry and takeover, the old replica still has its local handle and calls this every heartbeat

[Abhinav-kodes]

Implemented in commits 0fb95b7 and d15a037.

renewClaimScript now guards renewal with HGET owners vs token. If a lease was taken over, the script returns -1 (or 0 if no claim exists). In heartbeatClaims, any result where res != 1 immediately deletes the local claimHandle from claimTokens, stopping all future renewals and preventing stale extensions.

[hhzhang16]

Is there a use case/need for all three top-level knobs?

[shimib]

100% agreed, @hhzhang16. This is an incredibly elegant pivot.

Shifting to a token-guarded ACK and a "smaller first slice" is a massive win. It completely eliminates the complex peripheral machinery where several of the correctness issues I flagged are cropping up:

1. Eliminates Issue [Tracking] Async Processor tasks #1 (Result Loss on ID Reuse) naturally: Removing the result-terminal:* keys and the associated dedup TTL entirely avoids the silent suppression of legitimate ID re-submissions. If there are no persistent terminal keys, there is nothing left to poison a subsequent run.
2. Hardens the Heartbeat/Lease Safety: Fencing both the ACK and RENEWCLAIM scripts by the ownership token guarantees that a hijacked or stale lease can never block progress, renew foreign claims (Issue Flow-control metrics integration? #2), or write duplicate records. Stale owners simply fail the fence check and can be cleaned up cleanly on the next tick.
3. Simplifies Graceful Shutdown (Issue 🌱 Add dependabot for automated dependency updates #3): Dropping the custom SWEEPHANDBACK logic in favor of letting healthy survivors naturally reclaim the lapsed leases is extremely robust. It relies on a single, proven recovery path rather than maintaining a separate shutdown-only protocol.

By token-guarding ACK, the core scripts collapse into something beautiful and simple. For example, ackResultScript can be reduced to just this:

-- KEYS: claimed, owners, idx, resultList
-- ARGV: id, resultJSON, token, listTTL
if redis.call('HGET', KEYS[2], ARGV[1]) ~= ARGV[3] then
  return 0 -- Stale owner: blocked from writing result
end
redis.call('LPUSH', KEYS[4], ARGV[2])
if tonumber(ARGV[4]) > 0 then
  redis.call('EXPIRE', KEYS[4], tonumber(ARGV[4]))
end
redis.call('HDEL', KEYS[1], ARGV[1])
redis.call('HDEL', KEYS[1], ARGV[1] .. ':score')
redis.call('HDEL', KEYS[2], ARGV[1])
redis.call('ZREM', KEYS[3], ARGV[1])
return 1 -- Successful owner-guarded ack

I strongly endorse this simplified direction for the initial fix. It makes the core hard-kill path rock-solid and much easier to review and merge!

[Abhinav-kodes]

Thanks @shimib and @hhzhang16 for the detailed reviews - very helpful.

I agree the first slice got too big and the extra parts are where the bugs came from. I will update this PR to the smaller slice you both suggest:

• keep only peek -> claim -> heartbeat (token fenced) -> ack (token fenced) -> reclaim
• use RequestToken as generation ID and claim token as owner fence, so a stale owner simply cannot publish, this removes the global dedup marker, dedup TTL/GC and the duplicate metric
• make renew fenced and clean up local handles on mismatch
• remove the shutdown sweep and retryOwned for now and let lease expiry handle graceful case too, and keep tuning to defaults for now

This will fix the blocking items 1-5 structurally instead of patching them one by one. Non-blocking 6-11 will be addressed separately or noted in docs. I have seen your reference branch @shimib, thanks for sharing - I will use it as reference but implement the smaller slice here and update the PR shortly.

[Abhinav-kodes]

@shimib , @hhzhang16
I have pushed the shrunken slice

Known limitations for this small slice (already in docs):

1. Graceful shutdown now waits for the lease to expire (5 minutes by default). There is no immediate handback; it can be added in a follow-up, so restarts are a little slower.
2. If a crash occurs while a request is in the retry queue, one extra pending copy may be created, but only one result will be published because the ownership token fences the other.
3. There is no dead-letter queue at the moment; redelivery is bounded only by the request deadline, after which a deadline-exceeded result is produced.
4. During a rolling upgrade, old pods still use ZPOPMIN and can still lose requests until all pods are updated. Redis persistence (AOF/replication) is required.

[hhzhang16]

Thanks for reducing the scope! I think there's even more to reduce. I don't fully see the point of the graceful-with-workers integration test when it comes to this MR's scope. There's also the question of the top-level flags/compat wiring and some other details I'm not sure are fully necessary right now. Also, could you update the MR's description?

[hhzhang16]

This is still allowing stale owners -- if the winning owner already acked and deleted the claim, a stale owner would see no owner and proceed to publish another result. I think that this would be better solved by requiring an exact owner-token match with missing owners also returning 0

[Abhinav-kodes]

Fixed in commit 9509245 (fix(redis): require exact owner match in ackResultScript without empty-token bypass).

ackResultScript now strictly requires an exact owner-token match (if not owner or owner ~= ARGV[3] or ARGV[3] == "" then return 0 end), fencing any missing owners or empty tokens. Test cases have been updated to register explicit claim tokens rather than relying on an empty token bypass.

[hhzhang16]

I don't think the requestToken comments can be resolved because this is still using ReqID() (same in line 206). This is kind of dangerous because a same-flow ID reuse can let an old result borrow the new generation's claim token and can ack the wrong claim

[Abhinav-kodes]

This is protected across both layers in commits 0fb95b7, d15a037, and 9509245:

• In-memory token isolation (claimKey): In Go memory, claimTokens is keyed by claimKey(reqID, reqToken) (reqID + "\x00" + reqToken"). When an older execution (Gen 1) completes, ackResult uses result.Routing.RequestToken and looks up only "req-1\x00gen-1". It cannot load Gen 2's token "req-1\x00gen-2".

• Redis-level fencing (ackResultScript): In Redis, when Gen 2 is claimed, keys.owners[reqID] is updated to Gen 2's token. When Gen 1 attempts to ack with its old token token1 (or ""), ackResultScript evaluates if not owner or owner ~= token or token == "" then return 0 end and strictly fences it (returns 0), neither publishing the stale result nor dropping Gen 2's claim.

[hhzhang16]

Local handles should be removed for any result other than 1 (including 0)

[Abhinav-kodes]

Thanks @hhzhang16 and @shimib for the review!

I have updated the PR with all the requested changes:

1. Contexts in Start(): Start() now uses detached contexts so workers and heartbeats do not exit early on shutdown signal. Added StopHeartbeatForTest() hook to simulate hard kill in tests.
2. ackResultScript: Removed the empty token bypass. It now strictly requires an exact owner and token match. Updated direct-push unit tests to create test claims first.
3. Negative values validation: Added checks in SortedSetConfig.Validate() to reject negative numbers for TTLs, intervals, and batch size.
4. Compat shim: Removed hardcoded claim defaults from compat.go and let LoadSortedSetConfig apply them.
5. Hard kill test: Narrowed down to a single test for claim expiry and takeover, and renamed the file to claim_expiry_loss_test.go.
6. Docs and release note: Removed stale references to :score field and updated release notes to describe owner fencing.

All tests (make test, race detector, integration tests) and make lint pass cleanly. Please check again when you get time!

[shimib]

@hhzhang16 waiting for you LGTM to merge

[hhzhang16]

Thanks for the updates, this is looking much cleaner. Most of my earlier threads can be resolved (I seem unable to resolve them though, reached out to llm-d about it). No new comments, but the ones I still see as needing to be addressed or clarified are:

• Redis claim state is still keyed only by request ID rather than request ID + token
• Result persistence through a sustained Redis outage still needs coverage
• Retry persistence through a sustained Redis outage still needs coverage
• The hard-kill test still leaves the original flow running
• The exactly-once/collapsed-to-one-execution guarantee still looks stronger than what the fencing provides

I’m fine with narrowing the documented guarantee or scope and tracking follow-ups where appropriate. Once these are addressed or clarified, I should be able to approve.

[Abhinav-kodes]

@hhzhang16

I've replied and clarified each of the 5 points in their respective threads above.

In the latest commit 382838e, I added dedicated unit tests in pkg/redis/sortedset_impl_test.go for sustained Redis outages (verifying local claim handles are dropped when retries exhaust, so heartbeats stop and requests can be redelivered) and slightly adjusted the doc wording on delivery guarantees.

All tests are passing. Please let me know if this is good to merge!

[hhzhang16]

Thanks, I took another look at the latest update and while it's looking a lot better, I’m not quite ready to approve yet. The remaining items I see are:

• The docs and PR description still overstate the terminal-delivery guarantee. Fencing is per claim; it does not by itself guarantee one terminal record per request generation.
• The sustained-outage tests verify that the local claim handle is removed, but don’t exercise recovery through lease expiry and redelivery.
• The hard-kill test still leaves Flow A’s consumer/reclaimer alive and doesn’t clean up either flow. I’m fine with treating this as a lease-expiry/takeover test if it’s described and cleaned up accordingly.
• The lease TTL guidance should be corrected: healthy claims are renewed, so the TTL shouldn’t need to exceed the longest inference plus drain time.

Also, the hosted workflows haven’t run on the latest commit yet since it needs an approval from a maintainer @lioraron

[Abhinav-kodes]

Thanks, I took another look at the latest update and while it's looking a lot better, I’m not quite ready to approve yet. The remaining items I see are:

* [The docs and PR description still overstate the terminal-delivery guarantee](https://github.com/llm-d/llm-d-async/pull/412#discussion_r3883811024). Fencing is per claim; it does not by itself guarantee one terminal record per request generation.

* [The sustained-outage tests](https://github.com/llm-d/llm-d-async/pull/412#discussion_r3866416708) verify that the local claim handle is removed, but don’t exercise recovery through lease expiry and redelivery.

* [The hard-kill test](https://github.com/llm-d/llm-d-async/pull/412#discussion_r3883806682) still leaves Flow A’s consumer/reclaimer alive and doesn’t clean up either flow. I’m fine with treating this as a lease-expiry/takeover test if it’s described and cleaned up accordingly.

* [The lease TTL guidance](https://github.com/llm-d/llm-d-async/blob/382838e9227772b8fc74040ce491d84bfc5a33de/docs/guides/durable-dequeue.md#L69-L74) should be corrected: healthy claims are renewed, so the TTL shouldn’t need to exceed the longest inference plus drain time.

Also, the hosted workflows haven’t run on the latest commit yet since it needs an approval from a maintainer @lioraron

Thanks @hhzhang16 for the review! I've updated the PR addressing all 4 points:

1. Delivery Guarantee Docs: Clarified in docs/guides/durable-dequeue.md that fencing is per claim lease (owners[id]==claimToken) and does not by itself guarantee single-generation uniqueness across separate redeliveries.
2. Lease TTL Guidance: Corrected the documentation to note that because active claims are renewed via heartbeats, the lease TTL only needs to exceed heartbeat intervals rather than the full inference duration.
3. Sustained Outage Recovery Tests: Extended both result and retry sustained outage unit tests to exercise full recovery through Redis reconnection, lease expiry, and reclaimer redelivery back to the pending queue.
4. Lease Expiry / Takeover Test: Renamed to TestLeaseExpiry_TakeoverRedeliversClaims, updated the description, and added full lifecycle context cancellation and flow shutdown cleanups for both flows.

[hhzhang16]

The specific local token-borrowing issue is fixed, but the broader durable identity concern is still there. Redis still keys the claimed payload, owner, and expiry index only by ReqID, while RequestToken is used only in the process-local map. Two active submissions with the same ID can therefore overwrite each other’s durable claim state. Batch Gateway’s UUID IDs avoid this, but the Producer path does not enforce one active generation per ID.

WRT the test: Flow A’s reclaimer is still running during the takeover. The test can therefore pass with A redelivering the expired claim and C consuming it, rather than proving recovery with the original owner fully inert. Also, Flow C’s consumer is not stopped during cleanup and the worker wait groups are never waited. The smallest sufficient fix here is to stop all of A’s flow loops before starting C and fully clean up both flows/workers. There also seems to be some missing cleanup in the test -- it never calls flowC.StopConsuming() or waits for wgA/wgC

[Abhinav-kodes]

Thanks @hhzhang16 for pointing these out, I have updated both items in the latest commit:

1. Redis Claim Identity: Redis claim state (claimed, owners, idx) is now keyed by claimKey(reqID, reqToken) instead of reqID alone. If two submissions share the same ReqID with different RequestTokens, they have separate keys in Redis and won't overwrite each other. Added a new unit test TestClaimRequest_MultipleGenerationsSameReqID_DoNotOverwrite to cover this.
2. Takeover Test & Cleanup:
   • Flow A now calls both flowA.StopConsuming() and flowA.Shutdown() before Flow C starts, so Flow A's reclaimer, consumer, and heartbeater are completely stopped/inert during the takeover.
   • Added full cleanup for Flow C (flowC.StopConsuming(), flowC.Shutdown()), unblocked the test server, and added wg.Wait() for both wgA and wgC.