Context
Follow-up hardening from the investigation of seqeralabs/sched#712 (a rolling redeploy stranded tasks). The primary fix — a thrown handler must not be treated as a terminal command failure — is addressed in lib-cmd-queue-redis PR #87 (CommandServiceImpl catch now returns false instead of persisting FAILED + acking). This ticket tracks the secondary shutdown-hygiene improvements in lib-data-stream-redis that came out of the same analysis.
These are hardening, not a correctness fix. The stranding was caused by the command layer writing a terminal FAILED to Redis on a transient throw — which happens before the ack, so no amount of close()-side cleverness can substitute for PR #87. With #87 in place the existing close() already behaves correctly (throw → return false → accepted=false → not acked → lease released in close() → redelivered → peer re-drives). The items below reduce deploy-time churn and narrow races; none of them cover steady-state transient DB errors (which #87 does).
Trigger recap
During pod shutdown Micronaut can close the Postgres (HikariCP) datasource before the command queue finishes draining. There is no ordering edge between them (the command-service bean does not depend on the datasource). So a re-poll's handler runs against an already-closed pool and throws. Current close() (AbstractMessageStream.java:600-635):
closed=true, interrupt dispatcher, join(1s)scheduler.shutdownNow() (cancel pending re-polls)- does not stop the handler executor — comment assumes "any active handler finishes on its own (short-lived) and acks"
- release in-flight leases (
stream.release) for redelivery
- stop heartbeat last
The flawed premise is step 3's "finishes cleanly and acks" — during shutdown the handler is not clean, it throws (or the datasource is gone).
Proposed improvements (ranked by value)
1. Destroy-ordering: stop + drain command/stream processing before the datasource closes (highest leverage)
Removes the trigger entirely — if handlers never observe a closed pool, no transient throw happens at shutdown. Today CommandHandlerRegistry uses a plain @PreDestroy; there's no guarantee it runs before HikariCP. Explore a high-priority ApplicationEventListener<ServerShutdownEvent> (or an explicit destroy-order edge) so command processing is quiesced first.
- Caveats: Micronaut destroy-ordering guarantees are weak/fiddly; shutdown-only; bounded by a drain timeout. Part of this lands in the downstream app (sched), not purely in the lib.
2. Drain in-flight handlers before releasing leases (step 4)
close() currently releases leases while handlers may still be running on the shared pool, so a peer can immediately reclaim and run the same command concurrently — a double-execution window on every rolling deploy. Await in-flight handlers (bounded) before releasing.
3. Honor closed on the ack path (belt-and-suspenders)
run()/acknowledge() ack unconditionally regardless of closed. Make them refuse to ack during shutdown and release for redelivery instead.
Acceptance / scope
- Decide which of 1–3 are worth doing (2 + 1 look highest value; 3 is optional).
- Any drain must be bounded by a configurable timeout and must not lengthen shutdown unacceptably.
- Preserve current redelivery semantics (leases released/lapsed → reclaimed by peers).
References
Context
Follow-up hardening from the investigation of seqeralabs/sched#712 (a rolling redeploy stranded tasks). The primary fix — a thrown handler must not be treated as a terminal command failure — is addressed in lib-cmd-queue-redis PR #87 (
CommandServiceImplcatch now returnsfalseinstead of persistingFAILED+ acking). This ticket tracks the secondary shutdown-hygiene improvements inlib-data-stream-redisthat came out of the same analysis.These are hardening, not a correctness fix. The stranding was caused by the command layer writing a terminal
FAILEDto Redis on a transient throw — which happens before the ack, so no amount of close()-side cleverness can substitute for PR #87. With #87 in place the existingclose()already behaves correctly (throw →return false→accepted=false→ not acked → lease released inclose()→ redelivered → peer re-drives). The items below reduce deploy-time churn and narrow races; none of them cover steady-state transient DB errors (which #87 does).Trigger recap
During pod shutdown Micronaut can close the Postgres (HikariCP) datasource before the command queue finishes draining. There is no ordering edge between them (the command-service bean does not depend on the datasource). So a re-poll's handler runs against an already-closed pool and throws. Current
close()(AbstractMessageStream.java:600-635):closed=true, interrupt dispatcher,join(1s)scheduler.shutdownNow()(cancel pending re-polls)stream.release) for redeliveryThe flawed premise is step 3's "finishes cleanly and acks" — during shutdown the handler is not clean, it throws (or the datasource is gone).
Proposed improvements (ranked by value)
1. Destroy-ordering: stop + drain command/stream processing before the datasource closes (highest leverage)
Removes the trigger entirely — if handlers never observe a closed pool, no transient throw happens at shutdown. Today
CommandHandlerRegistryuses a plain@PreDestroy; there's no guarantee it runs before HikariCP. Explore a high-priorityApplicationEventListener<ServerShutdownEvent>(or an explicit destroy-order edge) so command processing is quiesced first.2. Drain in-flight handlers before releasing leases (step 4)
close()currently releases leases while handlers may still be running on the shared pool, so a peer can immediately reclaim and run the same command concurrently — a double-execution window on every rolling deploy. Await in-flight handlers (bounded) before releasing.3. Honor
closedon the ack path (belt-and-suspenders)run()/acknowledge()ack unconditionally regardless ofclosed. Make them refuse to ack during shutdown and release for redelivery instead.true); a genuine terminal completion acking during shutdown is actually correct, so the upside is narrow.Acceptance / scope
References
CommandServiceImplretry-on-throw)lib-data-stream-redis/src/main/java/io/seqera/data/stream/AbstractMessageStream.java(close()~600-635,run()/acknowledge()~483-530),schedCommandHandlerRegistry(@PreDestroy)