fix(worker): make shutdown promptly cancel inflight work - #2169
Conversation
Loops sleep via setTimeout that don't observe shouldStop, and fetch() calls have no timeout. SIGTERM during a long sleep (auto-topup 2m, data-retention 5m) or hung upstream poll exceeds the 15s shutdown budget. Centralize shutdown state in shutdown.ts: a stop AbortController plus a stop-signal Promise the new interruptibleSleep races against. Replace per-loop setTimeout sleeps with interruptibleSleep, thread the abort signal into video-job and webhook fetches with a 30s timeout, and short-circuit per-item inner loops on stop. https://claude.ai/code/session_01BY1g5cpuGDLPAAPcYsdooY
WalkthroughAdds a centralized cooperative shutdown API and makes the worker and its services shutdown-aware: interruptible sleeps, abortable fetches tied to the stop signal, and early-loop exits so long-running loops and external requests stop promptly when shutdown is requested. ChangesCoordinated Shutdown + Worker Integration
Sequence Diagram(s)sequenceDiagram
autonumber
participant Worker
participant Shutdown
participant UpstreamAPI
participant Webhook
Worker->>Shutdown: resetShutdown()
Worker->>Shutdown: isStopRequested() (loop guard)
Worker->>Shutdown: interruptibleSleep(ms)
alt no stop requested
Worker->>UpstreamAPI: fetchWithSignals (uses AbortSignal)
UpstreamAPI-->>Worker: response
Worker->>Webhook: deliverWebhook (fetchWithSignals)
Webhook-->>Worker: delivery response
else stop requested
Shutdown->>Worker: abort signal (stopSignal)
Worker->>Shutdown: isShutdownAbort? -> true
Worker-->>Worker: break loops / skip bookkeeping
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.github.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 69d5ec02ab
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| const signal = | ||
| typeof AbortSignal.any === "function" | ||
| ? AbortSignal.any([stopSignal, timeoutSignal]) | ||
| : stopSignal; |
There was a problem hiding this comment.
By wiring the global stop signal into every upstream fetch, a SIGTERM now makes in-flight polls reject with AbortError, but processPendingVideoJobs() still handles those in its generic error path (nextErrorCount, backoff, and eventual forced failure). During normal restarts, this can artificially accumulate poll errors and eventually mark healthy jobs as failed even though the upstream request was canceled only for shutdown; the abort reason should be detected and short-circuited when stop is requested.
Useful? React with 👍 / 👎.
| const response = await fetchWithSignals( | ||
| delivery.targetUrl, | ||
| { | ||
| method: "POST", | ||
| headers, |
There was a problem hiding this comment.
Using the stop signal for webhook POSTs means a shutdown now aborts in-flight deliveries, but deliverWebhook() treats that abort like a transport failure and records retry/failed attempts. On deployments or scale-down events, this can consume retry budget and incorrectly transition callbacks toward failure despite no real endpoint error; aborts caused by worker stop should bypass attempt escalation.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Pull request overview
This PR improves the worker’s ability to shut down within the 15s termination budget by centralizing shutdown state and making both long sleeps and in-flight network work promptly cancellable.
Changes:
- Introduces a shared shutdown module (
shutdown.ts) that provides a stop state, anAbortSignal, and aninterruptibleSleep()helper. - Replaces per-loop
setTimeoutsleeps withinterruptibleSleep()and adds stop checks in inner batch loops to exit promptly. - Threads shutdown cancellation + a 30s timeout into upstream polling and webhook delivery
fetch()calls.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
| apps/worker/src/worker.ts | Swaps loop sleeps to interruptibleSleep(), checks stop state in batch loops, and uses centralized shutdown controls. |
| apps/worker/src/shutdown.ts | Adds global shutdown primitives: stop flag, stop promise, abort controller, and interruptible sleep helper. |
| apps/worker/src/services/video-jobs.ts | Adds stop-aware + timed fetch() wrapper and stop short-circuiting inside job/webhook processing loops. |
| apps/worker/src/services/follow-up-emails.ts | Makes email pacing sleep interruptible and stops per-org processing when shutdown is requested. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| const stopSignal = getStopSignal(); | ||
| const timeoutSignal = AbortSignal.timeout(timeoutMs); | ||
| const signal = | ||
| typeof AbortSignal.any === "function" | ||
| ? AbortSignal.any([stopSignal, timeoutSignal]) | ||
| : stopSignal; |
| const response = await fetchWithSignals( | ||
| delivery.targetUrl, | ||
| { | ||
| method: "POST", | ||
| headers, | ||
| body: payload, | ||
| redirect: "manual", | ||
| }, | ||
| WEBHOOK_DELIVERY_TIMEOUT_MS, | ||
| ); |
| export async function interruptibleSleep(ms: number): Promise<void> { | ||
| if (shouldStop || ms <= 0) { | ||
| return; | ||
| } | ||
| let timer: NodeJS.Timeout | undefined; |
![coderabbitai[bot]](https://avatars.github.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
apps/worker/src/services/video-jobs.ts (1)
2010-2087:⚠️ Potential issue | 🟡 Minor | ⚡ Quick winConsider: shutdown-induced AbortError is currently recorded as a poll failure.
When
requestStop()aborts an in-flightfetchUpstreamStatusmid-poll, this catch path incrementsllmgateway_poll_error_countand persists it. Across multiple unlucky restarts hitting the same job, this can drive a healthy job towardVIDEO_JOB_MAX_POLL_ERROR_COUNTand mark it failed. Detecting AbortError whenisStopRequested()is true and skipping the error-bookkeeping path would avoid that.🛡️ Proposed early-skip for shutdown-induced aborts
} catch (error) { const message = error instanceof Error ? error.message : String(error); + if ( + isStopRequested() && + error instanceof Error && + (error.name === "AbortError" || error.name === "TimeoutError") + ) { + logger.info("Skipping poll error bookkeeping for shutdown-aborted job", { + videoJobId: job.id, + upstreamId: job.upstreamId, + }); + break; + } logger.error( "Error polling video job", error instanceof Error ? error : new Error(message),🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/services/video-jobs.ts` around lines 2010 - 2087, The catch block treating all exceptions as poll failures should skip bookkeeping when the error is a shutdown-induced abort: detect an AbortError (e.g. error.name === "AbortError" or instanceof DOMException) AND isStopRequested() is true, and in that case do not increment llmgateway_poll_error_count or update the job record—just log/skip and continue; change the catch in the polling loop (the block around fetchUpstreamStatus / finalizeVideoJob) to early-return/continue on shutdown aborts before running the existing error-count logic that uses getVideoJobPollErrorCount, VIDEO_JOB_MAX_POLL_ERROR_COUNT, and upstreamStatusResponse updates.
🧹 Nitpick comments (1)
apps/worker/src/worker.ts (1)
1197-1223: 💤 Low valueRetry/stop interleave looks correct, but minor flow nit.
The retry loop correctly stops on
isStopRequested()both before sleeping and after waking. One small observation: when the loop breaks due to shutdown,recordLogInsertFailure()at L1226 still fires, treating the abort as a real Postgres failure for the circuit breaker. That's fine in the immediate-shutdown path (the loop is exiting anyway), but ifstartWorker()is later invoked again in the same process (e.g. tests), the breaker's backoff persists across restarts becauseresetShutdown()doesn't resetlogInsertCircuit. Worth noting if you support in-process restart; otherwise safe to ignore.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/worker.ts` around lines 1197 - 1223, The loop currently treats an abort due to shutdown as a real failure by calling recordLogInsertFailure() after breaking, causing logInsertCircuit backoff to persist across in-process restarts; fix by either (A) guarding the failure recording so recordLogInsertFailure() is only called when !isStopRequested() (i.e., check isStopRequested() before calling recordLogInsertFailure() in the code path after the retry loop), or (B) ensure resetShutdown() also resets the circuit by invoking logInsertCircuit.reset() (or equivalent) so startWorker() restarts with a clean circuit; update the code around the retry loop and resetShutdown()/startWorker() to implement one of these approaches.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@apps/worker/src/services/video-jobs.ts`:
- Around line 46-58: The fetchWithSignals function currently falls back to just
getStopSignal() when AbortSignal.any isn't present, which drops the timeout;
remove that fallback and always combine stopSignal and timeoutSignal with
AbortSignal.any (using AbortSignal.any([stopSignal, timeoutSignal])) so the
fetch call always gets a composite signal that enforces both user stop and
timeout; update references in fetchWithSignals to use AbortSignal.any directly
and keep creation of timeoutSignal via AbortSignal.timeout(timeoutMs) and
stopSignal via getStopSignal().
---
Outside diff comments:
In `@apps/worker/src/services/video-jobs.ts`:
- Around line 2010-2087: The catch block treating all exceptions as poll
failures should skip bookkeeping when the error is a shutdown-induced abort:
detect an AbortError (e.g. error.name === "AbortError" or instanceof
DOMException) AND isStopRequested() is true, and in that case do not increment
llmgateway_poll_error_count or update the job record—just log/skip and continue;
change the catch in the polling loop (the block around fetchUpstreamStatus /
finalizeVideoJob) to early-return/continue on shutdown aborts before running the
existing error-count logic that uses getVideoJobPollErrorCount,
VIDEO_JOB_MAX_POLL_ERROR_COUNT, and upstreamStatusResponse updates.
---
Nitpick comments:
In `@apps/worker/src/worker.ts`:
- Around line 1197-1223: The loop currently treats an abort due to shutdown as a
real failure by calling recordLogInsertFailure() after breaking, causing
logInsertCircuit backoff to persist across in-process restarts; fix by either
(A) guarding the failure recording so recordLogInsertFailure() is only called
when !isStopRequested() (i.e., check isStopRequested() before calling
recordLogInsertFailure() in the code path after the retry loop), or (B) ensure
resetShutdown() also resets the circuit by invoking logInsertCircuit.reset() (or
equivalent) so startWorker() restarts with a clean circuit; update the code
around the retry loop and resetShutdown()/startWorker() to implement one of
these approaches.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 216d69b0-bc8f-4705-b2a9-35dd73135729
📒 Files selected for processing (4)
apps/worker/src/services/follow-up-emails.tsapps/worker/src/services/video-jobs.tsapps/worker/src/shutdown.tsapps/worker/src/worker.ts
Address review: aborts caused by shutdown were treated as upstream failures, inflating poll error counts and webhook retry budget. isShutdownAbort short-circuits both paths so shutdown leaves jobs and deliveries untouched for the next worker run. Drop AbortSignal.any fallback (Node 24 target). Add unit tests for shutdown primitives. https://claude.ai/code/session_01BY1g5cpuGDLPAAPcYsdooY
There was a problem hiding this comment.
🧹 Nitpick comments (1)
apps/worker/src/services/video-jobs.ts (1)
2132-2134: ⚡ Quick winAdd early function-level stop guards to avoid extra DB work during shutdown.
You already break inner loops, but both flows can still do avoidable queries when stop is requested. A top-level
isStopRequested()fast-return in each processor keeps shutdown tighter.💡 Minimal diff
export async function processPendingVideoJobs(): Promise<void> { + if (isStopRequested()) { + return; + } const now = new Date(); const jobsToPoll = await claimDueVideoJobsForPolling(now); @@ - const terminalJobsToFinalize = await db + if (isStopRequested()) { + return; + } + const terminalJobsToFinalize = await db .select() .from(tables.videoJob) @@ export async function processPendingWebhookDeliveries(): Promise<void> { + if (isStopRequested()) { + return; + } const dueDeliveries = await db .select() .from(tables.webhookDeliveryLog)Also applies to: 2351-2353
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@apps/worker/src/services/video-jobs.ts` around lines 2132 - 2134, Add a top-level fast-return guard using isStopRequested() at the start of each processor function that contains the inner loop (the code that currently checks isStopRequested() and breaks) so the function returns immediately and avoids any further DB queries during shutdown; specifically add this early check in the processor that contains the shown snippet (the loop where isStopRequested() currently triggers a break) and mirror the same guard in the other flow around the lines noted (the block referenced at 2351-2353) so both code paths exit quickly without performing extra work.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@apps/worker/src/services/video-jobs.ts`:
- Around line 2132-2134: Add a top-level fast-return guard using
isStopRequested() at the start of each processor function that contains the
inner loop (the code that currently checks isStopRequested() and breaks) so the
function returns immediately and avoids any further DB queries during shutdown;
specifically add this early check in the processor that contains the shown
snippet (the loop where isStopRequested() currently triggers a break) and mirror
the same guard in the other flow around the lines noted (the block referenced at
2351-2353) so both code paths exit quickly without performing extra work.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: cbad06a6-e4e3-4e81-959f-ce2a38e9edf5
📒 Files selected for processing (2)
apps/worker/src/services/video-jobs.tsapps/worker/src/shutdown.spec.ts
Loops sleep via setTimeout that don't observe shouldStop, and fetch()
calls have no timeout. SIGTERM during a long sleep (auto-topup 2m,
data-retention 5m) or hung upstream poll exceeds the 15s shutdown
budget. Centralize shutdown state in shutdown.ts: a stop AbortController
plus a stop-signal Promise the new interruptibleSleep races against.
Replace per-loop setTimeout sleeps with interruptibleSleep, thread the
abort signal into video-job and webhook fetches with a 30s timeout, and
short-circuit per-item inner loops on stop.
https://claude.ai/code/session_01BY1g5cpuGDLPAAPcYsdooY
Summary by CodeRabbit