Add async worker-pool processing to lib-cmd-queue-redis

[pditommaso]

Summary

Extends lib-cmd-queue-redis to process queued commands on a bounded worker pool instead of serially on the single poll thread, so multiple commands run concurrently and the poll thread never blocks on a handler — while guaranteeing a command's handler is not run concurrently on more than one replica (exactly-once-ish: single concurrent runner, at-least-once on crash).

Design spec: docs/superpowers/specs/2026-07-01-async-command-processing-design.md.

How it works

• Non-blocking dispatch — processCommand (poll thread) hands each command to a bounded ExecutorService and returns immediately; runCommand runs the handler to completion on a worker.
• Single concurrent runner across replicas — a per-command distributed lock (lib-lock tryAcquire + watchdog). The lock is auto-renewed while the holder is alive and expires by TTL on crash (no hand-rolled heartbeat). Same-replica dedup via an in-flight Set registered before dispatch. In non-Redis mode an in-JVM lock is selected automatically.
• Ack-on-reclaim — the worker never acks; a finished command's stream entry is removed on the next reclaim once the store shows a terminal state, so the shared lib-data-stream-redis interface is untouched. Result is visible in the store immediately; only the (harmless) stream entry lingers up to one claim timeout.
• Backpressure — bounded pool queue; a rejected command is left queued and retried on a later delivery.

Backward compatibility

• The execute() → running() + checkStatus() external-job pattern is preserved (handlers overriding checkStatus are polled; in-process handlers re-run on crash recovery).
• executeTimeout() retained and @Deprecated (no longer consulted); new CommandConfig methods are defaults.
• No change to MessageStream/AbstractMessageStream/RedisMessageStream.

New deps: lib-lock, lib-lock-redis, lib-activator, redis.clients:jedis.

Testing — 36 tests, all green

• 15 existing (no regression, incl. the slow/checkStatus scenario) · 17 new unit · 4 Testcontainers integration.
• Integration (real Redis, 1s claim timeout) proves: single-runner across two replicas (loser reclaims the in-flight message and is refused the lock — asserted via a counting lock delegate), crash-recovery re-run, ack-on-reclaim, and non-blocking intake.

Review

Went through two adversarial review passes (spec + code). All spec-review blockers were fixed and verified; the code review found no blockers.

Not included (release steps)

README.md / changelog.txt / VERSION bump — left for a separate release commit.

🤖 Generated with Claude Code

[jordeu]

Reviewed the async worker-pool changes. The core design is sound and well-tested — these are edge / failure-path hardening notes, each independently re-verified by an adversarial check (one was confirmed with a deterministic reproduction). None are happy-path correctness bugs.

• 2 worth fixing — both silently defeat the recovery guarantees the design advertises: the inFlight leak on a tryAcquire exception, and the reflection-based checkStatus misclassification that can strand crash-recovered commands.
• 2 low-cost defensive fixes — lock leak on forced shutdownNow(), and the reserved-qualifier bean collision.
• 1 trivial nicety — null error on a no-message exception.

Severity/likelihood and a suggested fix are in each inline comment.

[jordeu]

[med–high] inFlight id leaks permanently if tryAcquire throws.

inFlight.add(id) runs at L299 before this unguarded tryAcquire; only pool.execute below is wrapped in try/catch. RedisLockManager.tryAcquire does jedisPool.getResource() + jedis.set(...), which throw an (unchecked) JedisException on a connection blip or pool exhaustion — more reachable now that the worker pool + lock watchdogs hold many connections. On a throw here, id is never removed from inFlight: the three remove sites (L306, L320, L376) are all unreachable on this path. Every later redelivery then hits !inFlight.add(id) → return false forever, so the command is unprocessable on this replica until JVM restart (and the set grows unbounded).

Fix: move inFlight.add(id) to after a successful acquire, or wrap the post-add body so inFlight.remove(id) runs on any exception.

[jordeu]

[high when triggered] checkStatus-override detection misclassifies handlers, stranding crash-recovered commands.

getDeclaringClass() != CommandHandler.class is true for any class interposed between the leaf handler and the interface — e.g. an in-process handler extending an abstract base that declares checkStatus (no Micronaut needed), or a Micronaut $Intercepted proxy when type-level advice such as @Transactional/@Timed covers checkStatus. Such a handler is then treated as external-job: after a crash leaves it RUNNING, reclaim dispatches CHECK_STATUS, the default checkStatus returns running(), and the command stays RUNNING forever (redelivered indefinitely, never terminal). A deterministic repro confirmed both the abstract-base and proxy cases.

Fix: prefer an explicit signal (marker interface / registration flag / distinct subtype) over reflecting on class shape.

[jordeu]

[low–med] Forced shutdownNow() leaks the locks of queued-but-unstarted tasks.

Locks are acquired on the poll thread (L304) and released only in the worker's finally (L377). Up to commandPoolQueueSize (default 100) tasks can sit in the pool queue already holding a lock; shutdownNow() discards them (its returned task list is ignored here), so those locks/watchdogs never release. The common case self-heals via TTL (~1 min), but since stop() is a public method, calling it without a full context teardown leaves the watchdog (on a separate scheduler) renewing the orphaned lock indefinitely — and local mode's ConcurrentHashMap has no TTL at all.

Fix: capture the Runnables returned by shutdownNow() (or track held locks) and release them after the pool terminates.

[jordeu]

[high if triggered, low likelihood] Reserved string qualifier can cause a NonUniqueBeanException at startup.

@EachBean(LockConfig) in RedisLockManagerFactory/LocalLockManagerFactory propagates each seqera.lock.<name> qualifier onto its LockManager. A host that configures seqera.lock.cmd-queue-internal-lock therefore creates a second @Named("cmd-queue-internal-lock") LockManager, making CommandServiceImpl's injection ambiguous → a hard context-startup failure. There is no programmatic guard (no @Primary), and CommandLockWiringTest can't catch it (it never sets that config).

Fix: a dedicated custom @Qualifier-meta-annotated annotation (e.g. @CommandLock) on the factory methods and the injection point makes the collision structurally impossible and is fully self-contained in this module.

[jordeu]

[minor] A handler exception with a null message is persisted as error = null.

The removed executeWithTimeout wrapped throws as "Command execution failed"; this now stores e.getMessage(), which is null for a no-message exception (e.g. new IllegalStateException()). Bounded — JVM-generated NPEs carry messages on Java 17/21, and the full stack trace is still logged at L368 — so it's diagnostic-only, no crash.

Fix (trivial): e.getMessage() != null ? e.getMessage() : e.toString().