feat(server): drain queue workers gracefully on SIGTERM (#22514)
## Why Deploys kill workers mid-job: neither the queue worker nor the API server ever calls `enableShutdownHooks()`, so NestJS never listens for SIGTERM and the graceful close in `BullMQDriver.onModuleDestroy` is dead code. On every rollout a worker dies instantly — an in-flight 10-minute AI stream freezes for the watching user, and BullMQ silently re-runs the half-executed job on another worker ~10 minutes later. This is the root cause, not a symptom: the correct drain semantics already exist in the driver (`worker.close()` waits for active jobs and stops picking new ones, per the BullMQ graceful-shutdown docs) — the process just never received the signal. ## What - `queue-worker.ts` + `main.ts`: enable shutdown hooks. SIGTERM now runs `onModuleDestroy` across providers: the BullMQ driver drains active jobs, `RedisClientService` and the AI cancel subscriber quit their Redis connections, TypeORM closes its pools, then the process exits on its own. - BullMQ close order: workers drain before queues close, so a job finishing during the drain can still enqueue follow-ups (e.g. the AI queue flushing the next queued message). - API server: `forceCloseConnections` so long-lived subscription sockets don't hold `close()` open until the pod is force-killed. They were dropped abruptly on every deploy before this PR too — clients already recover. - Drain start/completion logs so pod terminations are debuggable. ## User impact Deploys stop corrupting in-flight background work. Follow-ups build on this: bounded drain-then-abort for AI stream jobs, and eliminating the stalled-job zombie re-run. ## Validation - Local: SIGTERM'd a running worker mid-job — drain log appears, the active job completes, "Message queue shutdown complete" is logged, process exits by itself. (Also verified with `LOGGER_IS_BUFFER_ENABLED=true` that final logs are not swallowed.) - The k8s side (termination grace period ≥ drain budget, exec'ing `node` directly so PID 1 receives SIGTERM) lands separately in twenty-infra. <!-- This is an auto-generated description by cubic. --> <a href="https://cubic.dev/pr/twentyhq/twenty/pull/22514?utm_source=github" target="_blank" rel="noopener noreferrer" data-no-image-dialog="true"><picture><source media="(prefers-color-scheme: dark)" srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img alt="Review in cubic" src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a> <!-- End of auto-generated description by cubic. --> --------- Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com>
This commit is contained in:
+32
-5
@@ -89,13 +89,40 @@ export class BullMQDriver
|
||||
}
|
||||
|
||||
async onModuleDestroy() {
|
||||
const workers = Object.values(this.workerMap);
|
||||
const workers = Object.entries(this.workerMap);
|
||||
const queues = Object.values(this.queueMap);
|
||||
|
||||
await Promise.all([
|
||||
...queues.map((q) => q.close()),
|
||||
...workers.map((w) => w.close()),
|
||||
]);
|
||||
if (workers.length > 0) {
|
||||
this.logger.log(
|
||||
`Draining active jobs on queues: ${workers.map(([queueName]) => queueName).join(', ')}`,
|
||||
);
|
||||
}
|
||||
|
||||
let workerCloseError: unknown;
|
||||
|
||||
try {
|
||||
await Promise.all(workers.map(([, worker]) => worker.close()));
|
||||
} catch (error) {
|
||||
workerCloseError = error;
|
||||
}
|
||||
|
||||
try {
|
||||
await Promise.all(queues.map((queue) => queue.close()));
|
||||
} catch (error) {
|
||||
if (!isDefined(workerCloseError)) {
|
||||
throw error;
|
||||
}
|
||||
|
||||
this.logger.error(
|
||||
`Failed to close queues during shutdown: ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
}
|
||||
|
||||
if (isDefined(workerCloseError)) {
|
||||
throw workerCloseError;
|
||||
}
|
||||
|
||||
this.logger.log('Message queue shutdown complete');
|
||||
}
|
||||
|
||||
work<T>(
|
||||
|
||||
Reference in New Issue
Block a user