Rework work queue around invocation-scoped leases

[pditommaso]

Summary

• make lib-data-workqueue concurrency invocation-scoped: a permit and thread are held only while one handler call executes
• leave non-terminal Redis entries in the Stream PEL and reclaim them after visibility timeout, with weighted fairness between new intake and retries
• heartbeat only active handler invocations, including handlers longer than visibility/max-processing time, without timeout-driven overlap
• keep lib-data-workqueue-redis at-least-once and backed only by Redis Stream data structures
• move near-exactly-once coordination into lib-cmd-queue-redis with renewable command-attempt guards, atomic stale-writer fencing, terminal-state checks, and stable command IDs
• add SUBMITTING so interrupted external submission is retried with the same idempotency key
• make command submission idempotent by ID while allowing retries to repair the state-write/queue-offer gap

Fixes #96.

Design and compatibility

This intentionally changes the WorkQueue SPI by separating new delivery (receiveNew) from expired PEL reclaim (reclaim). concurrency() now limits active handler invocations rather than admitted/live task lifecycles.

lib-data-workqueue guarantees at-least-once delivery. It does not claim exactly-once side effects. lib-cmd-queue-redis adds the state/ownership layer needed for near-exactly-once processing under normal operation. Command handlers must use command.id() as the idempotency key for external side effects because no local protocol can make an uncooperative remote API exactly once across process failure.

SUBMITTING is a new serialized command status and therefore needs consideration during mixed-version rollout.

Adversarial review

The review specifically exercised:

• more non-terminal tasks than the concurrency limit, verifying intake does not freeze
• multi-threaded handler execution while preserving the configured invocation bound
• handler runtime longer than Stream visibility timeout and max-processing-time, verifying no overlapping call
• two application contexts sharing real Redis, duplicate command submission, and a handler longer than claim timeout
• dead-owner lease expiry and takeover
• stale owner renewal and state writes
• state-created/queue-offer repair through idempotent resubmission
• executor rejection, close while active, ack failure, cancellation, and duplicate delivery paths

Two races found during review were corrected:

1. Redis lease renewal now checks the current Stream PEL owner before touching the entry, so a stale worker cannot claim it back from a peer.
2. Command ownership verification and state persistence are now one atomic state-provider operation, so a lease expiring between a read and write cannot let a stale attempt overwrite the new owner.

Validation

./gradlew :lib-data-store-state-redis:check :lib-data-workqueue:check :lib-data-workqueue-redis:check :lib-cmd-queue-redis:check
BUILD SUCCESSFUL

The suite includes unit tests and Testcontainers-backed Redis end-to-end validation.

[pditommaso]

Adversarial review

Reviewed the full diff plus the surrounding code it depends on (AbstractStateStore, RedisWorkQueue.claimMessage, RedisWorkQueueConfig, sched's consumer config). The direction is right — invocation-scoped permits and Stream-owned retry are simpler than the in-process re-poll scheduler. But the safety-valve removal and the second lease layer introduce worse failure modes than the ones they close.

Blocking

1. A successful command is discarded and re-executed when the attempt guard lapses.
CommandServiceImpl.java:341 — saveOwned(newState, owner) for the terminal result sits inside the try, and catch (Exception e) at :345 treats the resulting IllegalStateException as a handler error: logs "Command processing errored, will retry", bumps errorsCount, returns false.

Scenario: handler runs 6 min, attemptLease is 5 min, the renewer thread misses (Redis blip, or the single renewer thread is saturated — see #10). execute() returns SUCCEEDED. saveOwned fails, the success is dropped, the entry goes back to the PEL, and execute() runs again on the next delivery. The guard detects the lost lease but converts a completed side effect into a re-execution, while logging it as a handler bug. The heartbeat already logged "ownership lost while handler is active" and did nothing about it.

Fix: handle ownership loss distinctly from a handler throw — check saveOwned's boolean, log "result discarded, ownership lost", return false without recordError. Don't let it enter the handler-error path.

2. A hung handler now stalls the queue permanently, with no recovery.
max-processing-time no longer releases anything (AbstractWorkQueue.java:577 just warns, once, via the warned set). The permit is released only in finishAttempt, i.e. only when accept() returns. So one handler blocked forever consumes its permit forever; with concurrency()==1 (the default) the dispatcher's slots.tryAcquire() fails and the entire queue stops, permanently, behind a single log line. Previously the valve freed both permit and lease.

Compounding it: close() (:594) no longer releases leases, and finishAttempt only shuts the heartbeat down once active is empty — so a hung handler keeps the entry pinned in the PEL indefinitely in a live JVM (Micronaut context close, tests, embedded use). No peer can ever take it.

The "releasing the lease would cause overlap" argument is sound, but it doesn't justify holding the permit. Decouple them: past maxProcessingTime, release the semaphore permit while keeping the entry in active for heartbeating. One boolean on InFlight so finishAttempt doesn't double-release. Intake survives; overlap still can't happen.

3. submit() can silently drop a command.
CommandServiceImpl.java:120 — if create() returns false and findById() returns empty (state TTL'd out, or evicted between the two calls), nothing is enqueued and submit() still returns the id as if accepted. The whole point of this block is repairing a lost offer; this branch does the opposite.

4. cancel() now fails whenever a handler is active.
:144 — cancel competes for the same attempt guard and returns false if it can't get it. The case cancel exists for — a long-running command — is exactly the case where the guard is held. And false is now indistinguishable from "already terminal" / "unknown id". No test covers it. If serializing cancel against the handler is the goal, it needs a cancel-requested flag the handler path observes, not a silent denial.

5. Dead-owner recovery latency is now max(visibilityTimeout, attemptLease).
attemptLease() defaults to a flat 5m with no relationship to getVisibilityTimeout(). Kill -9 a replica mid-command: the Stream entry is reclaimable after the visibility timeout (~60s in sched), but every reclaim is fenced by the dead owner's guard until 5 min elapse — ~5 pointless deliveries and 5× the failover time the queue advertises. Derive attemptLease from the visibility timeout, or document that it dominates failover.

Should fix

6. Unbounded PEL growth from the new re-enqueue rule. :120 re-offers for any non-terminal status. A caller that re-submits on a schedule (which the PR now advertises as the repair mechanism) accumulates one permanent PEL entry per attempt for a command that runs for hours — each redelivered every visibility timeout, each fenced, each costing a guard round trip. Cheapest correct rule: re-enqueue only when status == PENDING. SUBMITTING/PROCESSING prove a delivery already happened, so there is no offer gap to repair.

7. SUBMITTING is the PR's one wire-breaking change and buys nothing functional. Both branches call handler.execute(command) (:330), and the idempotency key is command.id(), which is stable in either status. Redelivery from PENDING already retried execute(). What it costs: a new enum value that old replicas cannot deserialize (Jackson throws on unknown enum; nothing configures READ_UNKNOWN_ENUM_VALUES_USING_DEFAULT_VALUE or @JsonEnumDefaultValue), so during a mixed rollout an old replica retry-storms on any state a new replica wrote — plus one extra Redis write per command. If the value is purely observability ("we attempted at least one submission"), either drop it or add the default-value handling so old readers degrade instead of failing.

8. RedisWorkQueue.release() is now renewLease(). It extends the wait rather than handing the entry back, inverting the method's name and its interface contract. It's also near-redundant: the entry was just delivered (idle ≈ 0) or the heartbeat has been renewing it — for one extra EVAL+XPENDING+XCLAIM per non-terminal invocation. And it makes release mean opposite things per backend (local re-offers immediately), which is precisely why the local cadence test had to lose its assertions (#11). Leaving it a no-op, as before, is both shorter and more honest.

9. Three maps keyed identically. active, activeSince, warned (AbstractWorkQueue.java:137-146) are all keyed by queueId|leaseId and all have exactly the lifetime of active. Put startedAt and a warned flag on the InFlight record; delete two maps and the three-way add/remove bookkeeping.

10. Two lease clocks, two renewers, one shared thread. The Stream PEL heartbeat and the command attempt guard now both need renewal on independent schedules, and they can disagree — the disagreement path is finding #1. On top of that, attemptHeartbeat (CommandServiceImpl.java:78) is a single thread serving up to concurrency() (default 1000) concurrent commands; one slow Redis call there delays every renewal, so ownership loss is a correlated, fleet-wide event rather than an isolated one. It's also never shut down — stop() (:100) closes the queue but leaks the thread per service instance.

Test gaps

11. AsyncWorkQueueLocalTest lost its only real assertion. timestamps.size() == 5 → >= 5, and the gap check (gaps.every { it >= 150 && it <= 1_500 }) was deleted. The test is renamed to "should release an invocation slot before retrying a non-terminal message" but never asserts a slot was released, never asserts concurrency, and never checks that the newly-added 'other' message was processed. Post-rename it verifies only "retries happen at all". Either assert the slot release (offer 2 with concurrency: 1 and require both to be seen) or drop the rename.

12. The two races the PR says it found have no service-level test. CommandAttemptStoreTest covers saveOwned fencing at the store layer, and RENEW_IF_OWNER isn't directly tested at all. Untested behaviours, all of which are findings above: handler succeeding after ownership loss, cancel while a handler is active, submit with created==false && existing==null, hung handler exhausting permits, stale renew after a peer reclaim.

13. sleep 2_500 / sleep 3_000 / sleep 1_500 in the Redis tests — ~7s of wall clock added, and they'll be the first to flake on a loaded CI box.

Nits

• CommandServiceRedisE2ETest.groovy and TestRedisWorkQueueConfig.java have truncated Apache headers (missing the warranty paragraph) — the third new file has the full one.
• CommandStateStoreImpl.saveOwned writes through provider.putIfOwner(key0(...), serialize(...)), bypassing AbstractStateStore.put() and therefore its RequestIdAware secondary-index write. Harmless today (CommandState isn't RequestIdAware) but it silently diverges from save() if that ever changes.
• Compat treatment is inconsistent: WorkQueue.receive() was kept as a default for external implementors, while receiveNew/reclaim (+3 on StateProvider, +6 on CommandStateStore) are abstract with no defaults — breaking any downstream implementor or hand-rolled test double of a published interface.
• putIfOwner is the first two-key Lua script in the codebase; it would CROSSSLOT under Redis Cluster. Moot with JedisPool, worth a comment if cluster is ever on the table.

---

The retry-cadence change (pollInterval → visibility timeout) is documented in the README table and matches what sched's sync model already does, so I'd leave it — just note that one knob now governs both retry cadence and the failover window, so you can't poll faster than you're willing to fail over.

#1, #2 and #4 are the ones I'd block on; #7 and #8 are pure deletions that shrink the diff.