fix(queue): stop the NATS consumer manufacturing duplicate deliveries (CLO-4773) - #186
Conversation
CLO-4773. Six ways a message that was delivered once got processed twice. Nothing extended the ack while a handler ran: JetStreamMessage::inProgress() implements +WPI and had no caller, so ackWait was a hard ceiling on job duration. Running past it is not a retry after a failure -- the redelivery is concurrent with the first attempt, so anything with a side effect does it twice at once. Screenshots ran 76.9% of production renders past their old 10s ackWait with four processes on the same durable. The broker now extends, and the Swoole adapter runs the heartbeat alongside the handler. It waits on a channel the handler pushes to rather than sleeping and then checking a flag, so finishing and waking are one operation: a flag leaves a window where the handler is done and the consume loop is reading the socket again, and extending there would write to a connection another coroutine is reading. The Swoole child's SIGTERM handler closed the consumer immediately. At maxCoroutines=1 the loop is parked while a handler runs, so the close landed mid-job: the handler finished, commit() threw on a closed connection, execution fell through to reject() which threw too, and an empty catch ate both -- a job that had just succeeded was NAK'd and re-run. Every rolling restart re-ran one. SIGTERM now flips the flag and lets the loop drain; the per-queue consumers close once consume() returns. Handling, acking and the success hook shared one try, so a transient ack timeout was treated exactly like a handler that threw and the message went back to the broker -- rerunning a job that had already run, or dead-lettering it at the ceiling as though it never had. Only a handler failure means the work did not happen, so only that one rejects now. reject() called a bare nak(), which redelivers at once, so the tier backoff governed only the per-attempt ack timer and never the rescheduling. The attempt's backoff entry is now passed to nak(). The MAX_DELIVERIES advisory subscription had no queue group, so every worker received it and each published its own dead-letter copy -- the dead letter multiplied by worker count, on exactly the messages an operator is trying to read. And a dead-stream publish that failed was discarded: the message is past maxDeliver, so nothing will deliver it again, and without a record it is simply gone. It goes to an optional onError now. Broker\Nats gains reconnect() so Pool::recover() can recover a failed lease in place instead of destroying the whole broker. It returns false when the broker was handed a live Connection rather than a factory, because then there is nothing to rebuild from and claiming success would hand back a broker whose every call fails on a closed socket. Message also carries the stream sequence now, which distinguishes one delivery from another where the pid deliberately does not. Verified against a real server, each behavioural test checked against the unfixed code first. The SIGTERM path is the exception: it needs ext-swoole, which is not available here, so it rests on CI.
|
Superseded by #173, which now carries this commit. Closing as part of collapsing the stack. Why the stack could not go green: with the base set to the branch below, On #173, base The commit is unchanged apart from the merge resolution in Branch |
1 participant
All six findings of CLO-4773 — Epic 3 of the NATS queue migration readiness audit. Six ways a message that was delivered once got processed twice.
Stacked on #185 (Epic 2), which is stacked on #173 (Epic 1). Merge order matters here and not only for the diff: every fix below converts silent loss into at-least-once delivery, which without Epic 2's idempotency trades a loss problem for a duplication problem.
P0 — Extend the ack while a handler runs
JetStreamMessage::inProgress()implements+WPIand had no caller, soackWaitwas a hard ceiling on job duration. Running past it is not a retry after a failure — the redelivery is concurrent with the first attempt, so anything with a side effect does it twice at once. Screenshots ran 76.9% of production renders past their old 10sackWaitwith four processes on the same durable.Broker\Nats::extend()sends the progress signal;extendInterval()reports the cadence (a third ofackWait, so two beats can be lost before the server gives up). The Swoole adapter runs the heartbeat alongside the handler.Design note: the heartbeat waits on a
Channelthe handler pushes to, rather than sleeping and then checking a flag. Finishing and waking become one operation — a flag leaves a window where the handler is done and the consume loop is back to reading the socket, and extending there would write to a connection another coroutine is reading. It also ends the heartbeat the moment the handler returns instead of at the end of an interval it no longer needs.P0 — Drain on SIGTERM instead of closing the consumer
The Swoole child's SIGTERM handler called
consumer->close()immediately. AtmaxCoroutines=1the loop is parked while a handler runs, so the close landed mid-job: the handler finished,commit()threw on a closed connection, execution fell through toreject()which threw too, and an empty catch ate both. A job that had just succeeded was NAK'd and re-run — every rolling restart re-ran one.SIGTERM now flips
stoppedand lets the loop drain (noticed withinRECEIVE_TIMEOUT). The per-queue consumers close onceconsume()has returned, so they stay open exactly as long as a handler might still need to ack.P1 — Separate an ack failure from a handler failure
Handling, acking and the success hook shared one
try, so a transient ack timeout was treated exactly like a handler that threw and the message went back to the broker — rerunning a job that had already run, or dead-lettering it at the ceiling as though it never had.Only a handler failure means the work did not happen, so only that one rejects. An ack failure is reported; the broker redelivers on its own deadline if the ack genuinely never landed, which a handler can guard against — a NAK here is a rerun for certain.
Behaviour change worth review: a throwing
successCallbackno longer rejects either. The message is acked and gone at that point, so there is nothing left to reject and rerunning the job would not fix a shutdown hook. Previously it routed toreject().P1 — Pass the backoff delay to
nak()A bare
nak()redelivers at once, so the tier backoff governed only the per-attempt ack timer and never the rescheduling: a permanently failing job burned its wholemaxDeliverbudget in a tight loop and dead-lettered in seconds instead of spreading over the intended window. The attempt's backoff entry is now passed tonak(), with the last entry repeating — the same way JetStream reads the array for its own timer, so the two agree on every attempt.P1 — Queue group on the MAX_DELIVERIES advisory
The advisory subscription was a plain ephemeral core subscription, so every worker received it and each published its own dead-letter copy — the dead letter multiplied by worker count, on exactly the messages an operator is trying to read.
Separately, a failed dead-stream publish was swallowed. The message is past
maxDeliver, so nothing will deliver it again; without a record it is simply gone. Those now reach an optionalonErrorreporter.P2 —
reconnect()and the stream sequencePool::recover()finds neitherreset()norreconnect(), so a failed lease destroyed and rebuilt the whole broker.reconnect()recovers in place.It returns
falsewhen the broker was handed a liveConnectionrather than a factory — there is nothing to rebuild from, and claiming success would hand back a broker whose every call fails on a closed socket.falseis what lets the pool destroy and recreate it, which is the only way back for that shape. Found by the test: the first version returnedtrueunconditionally and the broker was dead afterwards.Message::getSequence()exposes the stream sequence — the stored copy's position, which distinguishes one delivery from another where thepiddeliberately does not.Verification
NatsBrokerTest/NatsPoolTestbin/monorepo validateEach behavioural test was checked against the unfixed code: disabling
inProgress()makes the long-job test fail on redelivery, and restoring the barenak()makes the backoff test fail on immediate redelivery.Gap to flag: the SIGTERM change is the one fix with no local test run — it needs
ext-swoole, which is not available in this environment, so it rests on CI'sSwooleTest/SwooleConcurrencyTest. Please look at that hunk with that in mind.The queue-group fix is asserted structurally (the subscription carries a group) rather than by racing two workers to a dead letter: that behaviour only diverges after a message exhausts
maxDeliveronackWaitalone, which is several seconds of wall clock and exactly the kind of timing-dependent test that fails on a loaded runner for unrelated reasons.