From e5fc1b3702a68624181ad440275543ff52888385 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?F=C3=A9lix=20Malfait?= <felix.malfait@gmail.com>
Date: Fri, 3 Jul 2026 18:19:27 +0200
Subject: [PATCH] feat(server): drain queue workers gracefully on SIGTERM
 (#22514)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

## 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>
---
 .../message-queue/drivers/bullmq.driver.ts    | 37 ++++++++++++++++---
 .../src/queue-worker/queue-worker.ts          |  2 +
 2 files changed, 34 insertions(+), 5 deletions(-)

diff --git a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts
index 258760ba67..8770297de4 100644
--- a/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts
+++ b/packages/twenty-server/src/engine/core-modules/message-queue/drivers/bullmq.driver.ts
@@ -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>(
diff --git a/packages/twenty-server/src/queue-worker/queue-worker.ts b/packages/twenty-server/src/queue-worker/queue-worker.ts
index 68bec8617d..efae3f64c9 100644
--- a/packages/twenty-server/src/queue-worker/queue-worker.ts
+++ b/packages/twenty-server/src/queue-worker/queue-worker.ts
@@ -20,6 +20,8 @@ async function bootstrap() {
 
     // Inject our logger
     app.useLogger(loggerService ?? false);
+
+    app.enableShutdownHooks();
   } catch (err) {
     loggerService?.error(err?.message, err?.name);