Drain in-flight work before releasing the queue

[pditommaso]

Retargeted to master now that #100 and #102 have landed. No VERSION or changelog changes — see below.

The diff against master is the drain and nothing else: 5 files, +516/−14.

File
CommandService.java	drain(Duration) + activeCommands()
CommandServiceImpl.java	drain implementation, in-flight counter
AbstractMessageStream.java	awaitQuiescent(Duration), cooperative close()
CommandServiceDrainTest.groovy	new
AbstractMessageStreamDrainTest.groovy	new

Why

Neither module can express "stop taking work, but let what is already running finish". A consumer still executing when the application shuts down is cut short, and then fails against collaborators that have already been destroyed.

In sched this stranded tasks (seqeralabs/sched#888): a rollout destroyed the Hikari pool while a command handler was mid-execution. The handler failed on its next query, failed again trying to record that failure, and the command was acked as terminally failed — 6 tasks across 2 runs left in a state clients read as SUBMITTED, still being polled 27h later.

The consumer had no way to say "wait for me", and the caller had no way to ask.

lib-data-stream-redis

• awaitQuiescent(Duration) — sets a closing flag the dispatcher observes at the head of its loop, then waits for it to exit. The cycle already in progress always runs to completion, which is what gives a consumer time to finish its work and its writes.

• close() now drains cooperatively first, interrupting only a dispatcher that overran the new closeTimeout() (10s, overridable).

  This is the behavioural change worth a second look. Interrupt used to be the primary mechanism, and interrupting a thread parked in a Redis read can hand a RESP-desynced connection back to the pool — precisely what [release] Add redis.pool.testOnBorrow option to lib-jedis-pool — v1.1.0 #95 mitigates via redis.pool.testOnBorrow, which defaults to false. This removes the cause rather than relying on a disabled-by-default mitigation.

• Poll-interval and error-backoff sleeps are sliced (50ms) so neither holds up a drain.

• An interrupt is now recognised in the dispatcher's catch arm and exits the loop, instead of being logged at ERROR as a stream error and retried after a backoff — previously every forced stop produced that noise.

lib-cmd-queue-redis

• drain(Duration) — quiesce the dispatcher → wait for in-flight handlers → then release the queue. That order matters: steps 1–2 run with every collaborator still usable. Returns false if the timeout elapsed with work outstanding, leaving the caller to decide.
• activeCommands() — count of executions in progress.
• The counter is taken around the task submitted to the executor, not around Future.get. executeWithTimeout abandons the future on overrun while the handler keeps running, and that abandoned execution is exactly what a drain must wait for. It also cannot be observed from the shared, container-managed TaskExecutors.BLOCKING pool, so the bookkeeping has to be explicit.
• stop() is unchanged and remains the immediate-release path.

Framework-agnostic by necessity, not taste: this module targets Micronaut 4.8.18, and GracefulShutdownCapable landed in 4.9.0. The binding lives in sched (on 4.10), which hooks drain() to ShutdownEvent — published by DefaultBeanContext.stop() before any bean is destroyed.

How this relates to #102, now that it has landed

They cover different windows, and neither subsumes the other:

• An execution that throws while the dispatcher is still waiting on it (duration < execute-timeout) reaches Retry a command whose handler throws instead of terminal-failing it (cmd-queue 0.4.1) #102's catch and is retried.
• An execution abandoned on timeout keeps running on the executor, and executeWithTimeout never retrieves its result — so an exception it throws is swallowed by the un-awaited Future and never reaches that catch. It can only be waited for, which is what this PR adds.

Where execute-timeout is short relative to real handler duration, the abandoned case is the common one. In sched, execute-timeout is 1s against measured 5–7s batches.

Also worth correcting something I wrote earlier in this PR: #102 does not fix the sched#888 path. TaskLaunchBatchHandler catches Exception itself and returns CommandResult.failure(...), so its exception never reaches #102's catch. #102 protects a different and larger surface — chiefly TaskSubmitHandler.execute, which has no try block at all.

No VERSION or changelog changes — deliberate

Master merged #102 and #103 together in d457e5d without bumping cmd-queue past 0.4.0 and without a changelog entry. So the 0.4.x line currently has three unreleased functional changes (retry-on-throw, error tracking, and this drain) and no version to describe them.

Rather than have this PR pick a number and write an entry covering two changes it did not make, versioning is left as one decision in one place. Whoever cuts the release sets VERSION and writes a single entry for all three. Nothing here can be consumed downstream until that happens — and .github/workflows/build.yml:66 publishes only when the master head commit message contains [release].

Note seqeralabs/sched#890 currently pins 0.4.1 / 1.5.1 in anticipation; those pins need to match whatever numbers are actually cut.

Verification

:lib-cmd-queue-redis:test + :lib-data-stream-redis:test with --rerun-tasks, against master's tree: 0 failures. Full repo suite also green (1058 tests).

Spec	Tests
AbstractMessageStreamDrainTest	5 — cooperative quiescence lets an in-progress consumer finish uninterrupted; no further messages claimed after the drain begins; false reported when the consumer outlives the timeout; close() no longer interrupts; no-op with no consumer registered
CommandServiceDrainTest	3 — waits for a handler abandoned by the execute-timeout and never interrupts it; reports false with the count still visible on overrun; no-op when never started

The drain spec sets execute-timeout: 200ms against a 1.5s handler, reproducing the production shape where an execution outlives the timeout and continues after execute() has already returned to the dispatcher.

Honest limits

• Bounded, therefore not a guarantee. The caller's timeout is a hard cap; on expiry the queue is released and teardown proceeds with handlers still running. Per the window split above, an abandoned execution's exception is swallowed, so on overrun that work is lost with nothing recorded. Handler-side idempotency remains necessary (seqeralabs/sched#889).
• Only covers graceful termination. SIGKILL, OOM-kill, node loss and spot reclamation produce no drain at all.
• Recovery latency after an incomplete drain is bounded by the stream claim timeout (60s in sched), since an unacked PEL entry waits for XAUTOCLAIM.
• Capables drain in parallel, not in order on the consumer side, so the HTTP server does not necessarily stop accepting before the queue quiesces. A late request can enqueue a command this replica will not process; benign, since a peer picks it up, but not sequenced.
• A single stuck handler can burn the whole budget, starving the rest of the drain. No per-command deadline.

🤖 Generated with Claude Code