Async, non-blocking consumer processing with heartbeat lease (lib-data-stream-redis + lib-cmd-queue-redis)

[pditommaso]

Summary

Reworks lib-data-stream-redis so message handlers run asynchronously and concurrently off the poll thread, while a heartbeated per-message lease guarantees a single live runner across replicas. lib-cmd-queue-redis adopts the shared model and drops its own timeout/lock machinery. Behaviour is opt-in via concurrency() (default 1).

This fixes three coupled defects that were previously impossible to resolve by tuning a single claim-timeout:

• Head-of-line blocking — the dispatcher thread hands each message to a virtual-thread executor and never runs a handler inline, so a slow handler no longer stalls the poll loop.
• Duplicate execution — a worker-scoped heartbeat (XCLAIM … JUSTID every claim-timeout/3) keeps an alive consumer from being reclaimed regardless of how long its handler runs; the lease lapses on crash, preserving real dead-consumer failover.
• Slow status detection — a not-yet-terminal command keeps its lease and is re-invoked in-process after pollInterval (Model B), decoupling re-poll cadence from claim-timeout.

Sched tickets addressed

• seqeralabs/sched#600 — VM provisioning blocked the command-queue dispatch thread (and, via claim-timeout reclaim, was re-executed cross-pod → churn). Fixed by async dispatch (no blocking) + heartbeat lease (no duplicate).
• seqeralabs/sched#303 — Nextflow-observed task start/complete lagged real ECS transitions by up to ~60s because RUNNING commands were only re-polled once idle crossed the 60s claim-timeout. Fixed by in-process re-poll at pollInterval (~1s), independent of claim-timeout.

Supersedes the accept()-scoped heartbeat proposal in libseqera#73 (that lease was only correct while accept() blocked the listener thread; async dispatch requires a worker-scoped lease).

Changes

lib-data-stream-redis 1.5.0 → 1.6.0

• SPI: Lease<M> record + poll / renew / ack / release; new heartbeatInterval() / maxProcessingTime() (the Redis impl wires them from RedisStreamConfig so the renew cadence tracks the configured claim-timeout). consume(streamId, consumer) retained as a backward-compatible default over the triad.
• AbstractMessageStream: dispatcher + re-poll scheduler + heartbeat daemon; handlers run on a shared virtual-thread executor (swap in the Micronaut @Named(BLOCKING) executor via withHandlerExecutor(...)); in-flight bounded by a Semaphore (concurrency(), default 1).
• RedisStreamConfig: getHeartbeatInterval() (default claim-timeout/3) and getMaxProcessingTime() (default 15m).
• LocalMessageStream gets the lease triad (renew no-op, release re-offer) and benefits from async dispatch; no PEL ⇒ no heartbeat needed.
• README: new Architecture section + diagram.

lib-cmd-queue-redis 0.4.0 → 0.5.0

• Removed executeWithTimeout / BLOCKING future.get(timeout) / the per-command lock. execute() and checkStatus() run directly on the shared virtual-thread executor; cross-replica single-runner comes from the per-message lease.
• CommandQueue.concurrency() = 1000 (in-flight ceiling; requires lib-data-stream-redis 1.6.0+).

Verification

43 tests green — unit + Testcontainers Redis integration — including: no reclaim of live work, crash failover, re-poll cadence decoupled from claim-timeout, serial-per-command, backpressure, and backward-compatible consume().

Design spec: docs/superpowers/specs/2026-07-11-async-message-stream-processing.md.

Notes for reviewers / rollout

• Behaviour change for all consumers even at concurrency()=1: ack is deferred to handler completion; re-poll of a not-yet-terminal message is pollInterval (was claim-timeout); dispatch is off-thread. wave should get a regression pass before raising its concurrency().
• A separate status-poll interval (distinct from pollInterval) is intentionally deferred until backend API rate limits require it.
• Known follow-up (not a blocker; lib-cmd-queue-redis is insulated): a direct AbstractMessageStream consumer whose handler throws every invocation re-polls every pollInterval with no backoff while holding its permit.

🤖 Generated with Claude Code