Conversation
|
@chrisgeo Can you add yourself to the CLA? https://github.com/riverqueue/rivercla Also, why don't you go ahead and rebase. We merged another fix quite recently that affected the changelog. |
989ee34 to
10c23cf
Compare
|
@brandur Rebased onto current master — the changelog conflict from #1359 is resolved. CLA signed (riverqueue/rivercla#33). |
brandur
left a comment
brandur
left a comment
There was a problem hiding this comment.
Thanks @chrisgeo — sorry about the time it took me to get back to this one.
I ran my LLM against the PR and it flagged a few issues we might want to fix before bringing it in. Do you mind taking a look at these?
[P1] Reconciliation can block cancellation delivery. producer.go:857
JobGetByIDMany runs synchronously on the producer’s main goroutine without a timeout. While it waits, that goroutine cannot
process cancellations or job completions. If workers hold the pool’s connections while waiting for cancellation, this
creates a deadlock. Run the query asynchronously with a bounded context, then apply its results on the producer goroutine.[P1] Local cancellation can hang indefinitely on an unstarted client. client.go:2937
A job’s running state doesn’t establish that this client owns it. Producers exist before Start(), so an unstarted PollOnly
client canceling jobs running elsewhere fills its undrained control channel. I reproduced the 101st cancellation blocking
past its context deadline. Local delivery needs to account for producer lifecycle and avoid an unconditional blocking send.[P2] Reconnection cancels jobs that were deliberately retried. producer.go:876
JobRetry preserves cancel_attempted_at. After canceling a job, retrying it, and starting its new attempt, reconciliation
interprets the old marker as a fresh cancellation and interrupts the worker. Clear the marker when retrying or otherwise
scope cancellation to the current attempt.
| // Notifies an internal producer of a job cancellation so a job running in | ||
| // this same process observes it immediately, for clients that have no | ||
| // notifier to deliver it via listen/notify. | ||
| // | ||
| // This deliberately does NOT share notifyProducerWithoutListenerQueueControlEvent | ||
| // above despite the near-identical dispatch: the two helpers guard on | ||
| // different conditions for a reason, and unifying them would either reopen | ||
| // this gap or introduce new bugs in the other one. | ||
| // | ||
| // - This helper guards on c.notifier == nil (this client has no notifier, | ||
| // for any reason, including an explicit Config.PollOnly on an otherwise | ||
| // listener-capable driver). Queue control events guard on | ||
| // driver.SupportsListener() instead, because pause/resume/metadata | ||
| // changes already have an independent, documented convergence path in | ||
| // poll-only mode (producer.pollForSettingChanges, on QueuePollInterval) — | ||
| // using this same client.notifier == nil guard there would double-deliver | ||
| // metadata changes not deduplicated the way pause/resume are, and would | ||
| // panic on QueueUpdate's nil control event when metadata is empty. | ||
| // - Job cancellation has no such fallback: nothing else ever re-checks a | ||
| // running job's cancel_attempted_at in poll-only mode, so a lost signal | ||
| // here is only caught by JobRescuer's much longer stuck-job sweep. | ||
| // | ||
| // The dispatch below (via TriggerQueueControlEvent) sends unconditionally and | ||
| // blocks if the target producer's control-event channel is full — the same | ||
| // characteristic notifyProducerWithoutListenerQueueControlEvent above already | ||
| // has for genuinely listener-incapable drivers. The caller (JobCancel) is | ||
| // responsible for only invoking this for jobs it knows are actually running, | ||
| // both because there's nothing to dispatch for a job with no in-process | ||
| // executor and to avoid a burst of cancellations against non-running jobs | ||
| // (e.g. many freshly-inserted jobs cancelled before the client is started) | ||
| // filling that channel and blocking indefinitely. | ||
| // | ||
| // Should only ever be invoked *outside* a transaction. If invoked within a | ||
| // transaction, the producer wouldn't yet be able to access the state that | ||
| // triggered the notification because it's not committed yet. |
There was a problem hiding this comment.
In the old days when everything was human written, I'd be okay with this sort of detailed comment. However, with the advent of LLMs, it's become far too easy to filibuster with super long comment blocks everywhere because the cost to generate them has gone to zero. When the whole codebase is full of them, no human will bother reading them anymore.
Do you want to see if you can your LLM to compact this down a little more? A lot of the comments elsewhere in the PR could be similarly boiled down more.
| // Wait until the notifier has actually observed the injected error and is | ||
| // blocked trying to reconnect. | ||
| require.Eventually(t, reconnecting.Load, 5*time.Second, 5*time.Millisecond, | ||
| "notifier never reached its reconnect attempt after the injected connection loss") |
There was a problem hiding this comment.
We have a concept of "test signals" used throughout here that this kind of thing would be a good fit for so you don't have to introducing sleeping into tests.
OOC, what LLM are you using? Normally using Claude and Codex, I find both have been smart enough to find and use the test signal convention automatically.
2 participants
Fixes the mechanism described in #1358:
Client.JobCancel()delivers its cancellation signal via PostgresLISTEN/NOTIFY, andNOTIFYis fire-and-forget — if the notifier is disconnected/reconnecting (its exponential backoff loop) at the moment theNOTIFYcommits, the signal is lost for good. The running job keeps executing unaware untilJobRescuer's stuck-job sweep eventually catches it (defaults to a 1h window).As discussed in #1358 (comment from @brandur): each client now polls its own currently-running jobs by primary key for
cancel_attempted_atevery time its notifier (re)establishes healthy listening — both the very first connect and every subsequent reconnect. This is cheap (a plainid = any(...)lookup via the existingJobGetByIDMany, no new query/index), rare (only fires on reconnect), and closes exactly the lost window: a signal lost during a reconnect is caught the moment the reconnect completes.Changes (first commit):
internal/notifier: newNotifier.RegisterListenerReadyFunc— lets a caller register a callback invoked every time the notifier establishes healthy listening on all its subscribed topics (including the initial connect; callers that only care about genuine reconnects can no-op on an empty precondition, which is what the producer does below).producer.go: registers aListenerReadyFunc(only when a notifier is configured) that signals a newreconnectCh, consumed on the producer's single owning goroutine. The handler is a no-op when there are no active jobs (true at startup, so the initial-connect firing is free); otherwise it looks up its currently-active job IDs viaJobGetByIDManyand cancels any withcancel_attempted_atset. A query failure retries with the existing exponential backoff utility rather than silently dropping the reconciliation; a malformed metadata row is logged and skipped without aborting the rest of the batch.A second gap from the same root observation (second commit, "Deliver cancellation locally when the client has no notifier"): a client configured without a notifier at all — including an explicit
Config.PollOnlyon an otherwise listener-capable driver (pgx, SQLite) — had no delivery path whatsoever for cancelling a job running in its own process; the local same-process dispatch that already exists for this case only ever activated for drivers that can't supportLISTEN/NOTIFYat all, missing the more common explicit-PollOnlycase entirely. This commit adds a narrowly-scoped fix via its own dispatch helper rather than changing the existing shared one (which is also used by queue pause/resume/update and has different, already-correct semantics for those). Happy to split this into its own PR if preferred.Testing: Added
TestProducer_JobCancelSurvivesNotifierReconnect, a deterministic repro against a real Postgres notifier: forces one simulated connection loss after a job starts, holds the notifier at the start of its reconnect attempt, cancels the job while disconnected (so theNOTIFYis lost), then releases the reconnect and asserts the job observes the cancellation. Fails onmaster(job runs to completion unaware); passes with this change. Also added direct coverage forRegisterListenerReadyFuncininternal/notifier, and for the second commit: a same-process explicit-PollOnlycancellation test and a regression test guarding against blocking when cancelling a burst of non-running jobs.No steady-state query cost is added to either fix — the reconciliation only runs on reconnect and is a no-op when nothing is running; the local dispatch only fires for jobs actually running in-process.