refactor(webapp): move mollifier drainer bootstrap out of legacy worker.server.ts

worker.server.ts is the original graphile-worker / ZodWorker file —
every task in its catalog is annotated "@deprecated, moved to
commonWorker.server.ts" (or similar). Adding new lifecycle wiring
there during phase-2 was a mis-routing.

Move the SIGTERM/SIGINT registration + drainer.start() call into a new
mollifierDrainerWorker.server.ts alongside the redis-worker workers,
and invoke its initMollifierDrainerWorker() from entry.server.tsx
right after Worker.init(). The drainer's own factory still validates
shutdown timeouts before constructing; the bootstrap registers signal
handlers BEFORE calling start(), preserving the create+start contract.

Also adds a header to worker.server.ts marking it legacy and pointing
new lifecycle code at the redis-worker pattern, so the next person
doesn't have to re-derive the routing rule.

Author: d-cs

apps/webapp/app/entry.server.tsx

@@ -6,6 +6,7 @@ import isbot from "isbot";
 import { renderToPipeableStream } from "react-dom/server";
 import { PassThrough } from "stream";
 import * as Worker from "~/services/worker.server";
+import { initMollifierDrainerWorker } from "~/v3/mollifierDrainerWorker.server";
 import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import {
@@ -247,6 +248,8 @@ Worker.init().catch((error) => {
   logError(error);
 });
 
+initMollifierDrainerWorker();
+
 bootstrap().catch((error) => {
   logError(error);
 });

apps/webapp/app/services/worker.server.ts

@@ -1,3 +1,24 @@
+/**
+ * ⚠️ LEGACY — Graphile-worker / ZodWorker setup. Do not touch.
+ *
+ * This file wires the original background-job system the webapp was
+ * built on (`@internal/zod-worker` → graphile-worker → Postgres). It is
+ * now in deprecation mode: every task in `workerCatalog` below is
+ * annotated with `@deprecated, moved to <new home>` and the live jobs
+ * for new features all run on `@trigger.dev/redis-worker` instead.
+ *
+ * Where to put new things:
+ *   - Background jobs / queues → use redis-worker, alongside
+ *     `~/v3/commonWorker.server.ts`, `~/v3/alertsWorker.server.ts`, or
+ *     `~/v3/batchTriggerWorker.server.ts`.
+ *   - Run lifecycle → `@internal/run-engine` via `~/v3/runEngine.server`.
+ *   - Custom polling loops with their own Redis connection → keep them
+ *     in their own lifecycle module (e.g. `~/v3/mollifierDrainerWorker.server.ts`)
+ *     and wire the bootstrap from `entry.server.tsx`. Don't reach into
+ *     `init()` below.
+ *
+ * Edit only when removing legacy paths.
+ */
 import { ZodWorker } from "@internal/zod-worker";
 import { DeliverEmailSchema } from "emails";
 import { z } from "zod";
@@ -26,7 +47,6 @@ import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
 import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
 import { RetryAttemptService } from "~/v3/services/retryAttempt.server";
 import { TimeoutDeploymentService } from "~/v3/services/timeoutDeployment.server";
-import { getMollifierDrainer } from "~/v3/mollifier/mollifierDrainer.server";
 import { GraphileMigrationHelperService } from "./db/graphileMigrationHelper.server";
 import { sendEmail } from "./email.server";
 import { logger } from "./logger.server";
@@ -107,7 +127,6 @@ let workerQueue: ZodWorker<typeof workerCatalog>;
 
 declare global {
   var __worker__: ZodWorker<typeof workerCatalog>;
-  var __mollifierShutdownRegistered__: boolean | undefined;
 }
 
 // this is needed because in development we don't want to restart
@@ -130,55 +149,6 @@ export async function init() {
   if (env.WORKER_ENABLED === "true") {
     await workerQueue.initialize();
   }
-
-  // Only the worker role drains the mollifier buffer. API-only replicas
-  // still produce into the buffer via the trigger hot path, but the
-  // polling loop + Redis consumer connection only belongs on workers —
-  // otherwise every webapp replica races for the same entries.
-  if (env.WORKER_ENABLED !== "true") {
-    return;
-  }
-
-  try {
-    // getMollifierDrainer() runs the singleton factory, which validates the
-    // shutdown-timeout reconciliation against GRACEFUL_SHUTDOWN_TIMEOUT and
-    // throws BEFORE constructing the drainer if it's misconfigured. The
-    // outer catch below logs and aborts drainer registration on either that
-    // validation error or a Redis init failure — no half-started state. The
-    // returned drainer is configured-but-stopped; start() runs below, AFTER
-    // the SIGTERM/SIGINT handlers are registered, so a signal landing during
-    // boot can never find the polling loop running without a graceful-stop
-    // path. Same `__mollifierShutdownRegistered__` guard owns both the
-    // handler registration and the start() call so dev hot-reloads don't
-    // double-register or double-start.
-    const drainer = getMollifierDrainer();
-    if (drainer && !global.__mollifierShutdownRegistered__) {
-      // The drainer owns a polling loop and a Redis client; let it drain
-      // in-flight pops on shutdown rather than tearing the process down
-      // mid-handler. `init()` is called per request from entry.server.tsx,
-      // and `process.once()` only removes its listener after it fires — so
-      // without a process-global guard, dev hot-reloads would stack a fresh
-      // listener pair every request. Mirrors the `__worker__` singleton
-      // pattern above.
-      // Bound shutdown so a hung handler can't block process exit past the
-      // pod's termination grace period. `drainer.stop({ timeoutMs })` logs a
-      // warning and returns if the deadline is hit while a handler is still
-      // in flight.
-      const stopDrainer = () => {
-        drainer
-          .stop({ timeoutMs: env.MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS })
-          .catch((error) => {
-            logger.error("Failed to stop mollifier drainer", { error });
-          });
-      };
-      process.once("SIGTERM", stopDrainer);
-      process.once("SIGINT", stopDrainer);
-      global.__mollifierShutdownRegistered__ = true;
-      drainer.start();
-    }
-  } catch (error) {
-    logger.error("Failed to initialise mollifier drainer", { error });
-  }
 }
 
 function getWorkerQueue() {

apps/webapp/app/v3/mollifierDrainerWorker.server.ts

@@ -0,0 +1,67 @@
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { getMollifierDrainer } from "./mollifier/mollifierDrainer.server";
+
+declare global {
+  // eslint-disable-next-line no-var
+  var __mollifierShutdownRegistered__: boolean | undefined;
+}
+
+/**
+ * Bootstraps the mollifier drainer.
+ *
+ * Two-step lifecycle:
+ *   1. Construct the drainer via the gated singleton in
+ *      `mollifierDrainer.server.ts`. That factory validates the
+ *      shutdown-timeout reconciliation against `GRACEFUL_SHUTDOWN_TIMEOUT`
+ *      and throws BEFORE returning if it's misconfigured; the returned
+ *      drainer is configured-but-stopped.
+ *   2. Register SIGTERM/SIGINT shutdown handlers, then call
+ *      `drainer.start()`. Doing this in the bootstrap (and not in the
+ *      factory) guarantees a signal landing during boot can never find
+ *      the polling loop running without a graceful-stop path.
+ *
+ * The drainer is intentionally NOT wired through `~/services/worker.server`
+ * — that file is the legacy ZodWorker / graphile-worker setup. The
+ * mollifier drainer is a custom polling loop over `MollifierBuffer`, not
+ * a graphile-worker job, so it gets its own lifecycle file alongside the
+ * redis-worker workers (`commonWorker`, `alertsWorker`,
+ * `batchTriggerWorker`).
+ *
+ * Gating order:
+ *   - `WORKER_ENABLED !== "true"`  → early return (API-only replicas
+ *     still produce into the buffer via the trigger hot path; only worker
+ *     replicas drain it, otherwise every replica races for the same
+ *     entries).
+ *   - `MOLLIFIER_ENABLED !== "1"`  → `getMollifierDrainer()` returns null
+ *     and the bootstrap is a no-op.
+ */
+export function initMollifierDrainerWorker(): void {
+  if (env.WORKER_ENABLED !== "true") {
+    return;
+  }
+
+  try {
+    const drainer = getMollifierDrainer();
+    if (drainer && !global.__mollifierShutdownRegistered__) {
+      // `__mollifierShutdownRegistered__` guards against double-register
+      // on dev hot-reloads (this bootstrap is called from
+      // entry.server.tsx, which Remix dev re-evaluates on every change).
+      // Same guard owns both the handler registration and the start()
+      // call so the two never get out of sync.
+      const stopDrainer = () => {
+        drainer
+          .stop({ timeoutMs: env.MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS })
+          .catch((error) => {
+            logger.error("Failed to stop mollifier drainer", { error });
+          });
+      };
+      process.once("SIGTERM", stopDrainer);
+      process.once("SIGINT", stopDrainer);
+      global.__mollifierShutdownRegistered__ = true;
+      drainer.start();
+    }
+  } catch (error) {
+    logger.error("Failed to initialise mollifier drainer", { error });
+  }
+}