[Fix] bullmq drains and aborts resumed Fast turns before shutting down

.changeset/bullmq-fast-turn-drain.md

@@ -0,0 +1,5 @@
+---
+'@roomote/web': patch
+---
+
+The bullmq service now drains and aborts the Fast turns it is executing before it shuts down, the same way the API does for the turns it admits. Interrupted turns are resumed by the parent-event queue inside the bullmq process, so a deploy that restarted both services could kill a resumed turn a second time with its durable claim and conversation lock still held, leaving the row to wait out both leases (up to 15 minutes) before it ran again. Shutdown now closes admissions, stops fetching queue wakeups, gives in-flight turns the drain window to finish, and aborts the stragglers so each hands its row back to the queue immediately. The window is `R_BULLMQ_SHUTDOWN_DRAIN_MS`, falling back to `R_API_SHUTDOWN_DRAIN_MS` and then 20 seconds; the shutdown sequence itself is shared between the two services.

.env.production.example

@@ -15,9 +15,12 @@ PREVIEW_PROXY_SUBDOMAIN_SUFFIX=preview
 # R_APP_ENV=production
 
 # How long an API shutdown lets in-flight Fast turns finish before aborting
-# the remainder. Keep it under the platform's SIGTERM-to-SIGKILL grace
-# window; 0 aborts active turns immediately.
+# the remainder so they resume on the next process. Keep it under the
+# platform's SIGTERM-to-SIGKILL grace window; 0 aborts active turns
+# immediately. The bullmq service, which runs the turns the queue resumes,
+# uses the same window unless R_BULLMQ_SHUTDOWN_DRAIN_MS overrides it.
 # R_API_SHUTDOWN_DRAIN_MS=20000
+# R_BULLMQ_SHUTDOWN_DRAIN_MS=20000
 # Set to true to keep inference retry backoff inside the process that owns
 # the Fast turn instead of scheduling it durably (kill switch for durable
 # retry scheduling).

apps/api/src/graceful-shutdown.ts

@@ -1,33 +1,19 @@
 import type { ServerType } from '@hono/node-server';
 import {
-  abortActiveFastAgentTurns,
-  beginFastAgentTurnDrain,
+  drainAndAbortFastAgentTurns,
   FastAgentProcessShutdownError,
-  waitForActiveFastAgentTurnsToSettle,
+  resolveFastAgentShutdownDrainMs,
+  type FastAgentShutdownDrainDeps,
 } from '@roomote/cloud-agents/server';
 
-// Most Fast turns finish within seconds, so letting them settle turns a
-// deploy-time interruption into a completed answer. The default leaves room
-// for the straggler abort, closeout delivery, and Sentry flush inside a
-// typical 30s SIGTERM-to-SIGKILL grace window. R_API_SHUTDOWN_DRAIN_MS
-// overrides it; 0 restores the previous abort-immediately behavior.
-const DEFAULT_API_SHUTDOWN_DRAIN_MS = 20_000;
-
+/** `R_API_SHUTDOWN_DRAIN_MS` overrides the shared default; 0 aborts at once. */
 export function resolveApiShutdownDrainMs(
   env: NodeJS.ProcessEnv = process.env,
 ): number {
-  const raw = env.R_API_SHUTDOWN_DRAIN_MS?.trim();
-  if (!raw) return DEFAULT_API_SHUTDOWN_DRAIN_MS;
-  const parsed = Number(raw);
-  return Number.isFinite(parsed) && parsed >= 0
-    ? parsed
-    : DEFAULT_API_SHUTDOWN_DRAIN_MS;
+  return resolveFastAgentShutdownDrainMs(['R_API_SHUTDOWN_DRAIN_MS'], env);
 }
 
-type ApiShutdownOptions = {
-  abortTurns?: typeof abortActiveFastAgentTurns;
-  beginDrain?: typeof beginFastAgentTurnDrain;
-  waitForTurns?: typeof waitForActiveFastAgentTurnsToSettle;
+type ApiShutdownOptions = FastAgentShutdownDrainDeps & {
   drainMs?: number;
   exitProcess?: (code?: number) => never;
   flushSentry?: () => Promise<unknown>;
@@ -39,9 +25,9 @@ export async function gracefullyShutdownApi(
   server: ServerType,
   signal: NodeJS.Signals,
   {
-    abortTurns = abortActiveFastAgentTurns,
-    beginDrain = beginFastAgentTurnDrain,
-    waitForTurns = waitForActiveFastAgentTurnsToSettle,
+    abortTurns,
+    beginDrain,
+    waitForTurns,
     drainMs = resolveApiShutdownDrainMs(),
     exitProcess = process.exit,
     flushSentry = async () => undefined,
@@ -53,19 +39,22 @@ export async function gracefullyShutdownApi(
   // Refuse new turn admissions and stop accepting connections first, then
   // give in-flight turns a bounded window to finish on their own. Only the
   // stragglers still active at the deadline are aborted.
-  beginDrain(reason);
-  const closePromise = new Promise<Error | null>((resolve) => {
-    server.close((error) => resolve(error ?? null));
-  });
-  const remaining = await waitForTurns(drainMs);
-  if (remaining > 0) {
-    logWarn(
-      `[api] Aborting ${remaining} Fast turn(s) still active after the ${drainMs}ms shutdown drain.`,
-    );
-  }
-  const abortPromise = abortTurns(reason);
+  let closePromise: Promise<Error | null> = Promise.resolve(null);
+  await drainAndAbortFastAgentTurns(
+    {
+      reason,
+      drainMs,
+      service: 'api',
+      logWarn,
+      onDrainStarted: () => {
+        closePromise = new Promise<Error | null>((resolve) => {
+          server.close((error) => resolve(error ?? null));
+        });
+      },
+    },
+    { abortTurns, beginDrain, waitForTurns },
+  );
   const closeError = await closePromise;
-  await abortPromise;
   if (closeError) {
     logError('[api] Graceful shutdown failed', closeError);
   }