Merge pull request 'fix(worker): graceful BullMQ worker shutdown (stop Missing-lock on deploy)' (#7) from fix/worker-graceful-shutdown into main

apps/worker/src/index.ts

@@ -1,3 +1,4 @@
+import type { Worker } from "bullmq";
 import { startEventBus } from "./consumers/event-bus";
 import { startScheduledJobs } from "./schedulers/nightly";
 import { startInsightPipeline } from "./schedulers/pipeline";
@@ -6,6 +7,8 @@ import { upsertSeedData } from "./lib/seed-runtime";
 import { redis } from "./redis";
 import { prisma } from "./db";
 
+const workers: Worker[] = [];
+
 async function main() {
   console.log("[worker] starting…");
   await redis.ping();
@@ -15,16 +18,27 @@ async function main() {
 
   await upsertSeedData().catch((e) => console.warn("[seed] failed:", e.message));
 
-  await startScheduledJobs();
-  await startInsightPipeline();
-  await startContentPipeline();
+  workers.push(await startScheduledJobs());
+  workers.push(await startInsightPipeline());
+  workers.push(await startContentPipeline());
   await startEventBus();
 
   console.log("[worker] up.");
 }
 
+let shuttingDown = false;
 const shutdown = async (sig: string) => {
-  console.log(`[worker] ${sig} — shutting down`);
+  if (shuttingDown) return;
+  shuttingDown = true;
+  console.log(`[worker] ${sig} — closing ${workers.length} workers gracefully`);
+  // Close workers FIRST: this stops them taking new jobs and releases held job
+  // locks, so the next container doesn't inherit half-finished jobs as "stalled"
+  // and spam "Missing lock … moveToFinished". Bounded so an in-flight job can't
+  // block the shutdown past the orchestrator's stop grace period.
+  await Promise.race([
+    Promise.allSettled(workers.map((w) => w.close())),
+    new Promise((resolve) => setTimeout(resolve, 15_000)),
+  ]);
   await prisma.$disconnect().catch(() => {});
   await redis.quit().catch(() => {});
   process.exit(0);

apps/worker/src/schedulers/content.ts

@@ -49,11 +49,12 @@ export async function startContentPipeline() {
     { name: "content-generate", data: {}, opts: { removeOnComplete: 50, removeOnFail: 25 } },
   );
 
-  new Worker(QUEUE, runJob, {
+  const worker = new Worker(QUEUE, runJob, {
     connection: redis,
     concurrency: 1,
     lockDuration: 5 * 60_000,
     stalledInterval: 60_000,
   });
   console.log("[content] armed: content-topics@*/8h, content-generate@*/10min");
+  return worker;
 }

apps/worker/src/schedulers/nightly.ts

@@ -46,6 +46,15 @@ export async function startScheduledJobs() {
     { name: "panel-backup", data: {}, opts: { removeOnComplete: 30, removeOnFail: 30 } },
   );
 
-  new Worker(QUEUE, runJob, { connection: redis, concurrency: 1 });
+  const worker = new Worker(QUEUE, runJob, {
+    connection: redis,
+    concurrency: 1,
+    // panel-backup (pg_dump) can run longer than the 30s default lock; a too-short
+    // lock makes the job look "stalled", gets re-run, and the original then fails
+    // its moveToFinished with "Missing lock". Match the other queues' 5min lock.
+    lockDuration: 5 * 60_000,
+    stalledInterval: 60_000,
+  });
   console.log("[scheduler] armed: nightly-refresh@03:00, audit-archive@03:30, panel-backup@04:00");
+  return worker;
 }

apps/worker/src/schedulers/pipeline.ts

@@ -202,7 +202,7 @@ export async function startInsightPipeline() {
     { name: "sentry-archive", data: {}, opts: { removeOnComplete: 30, removeOnFail: 15 } },
   );
 
-  new Worker(QUEUE, runJob, {
+  const worker = new Worker(QUEUE, runJob, {
     connection: redis,
     concurrency: 2,
     lockDuration: 5 * 60_000,
@@ -211,4 +211,5 @@ export async function startInsightPipeline() {
   console.log(
     "[pipeline] armed: posthog-ingest@*/5min, tag-sessions@*/2min, compress-sessions@*/3min, analyze@*/4min, validation@05:00, github-sync@*/10min, retention@04:15, daily-brief@05:00, vin-anomaly-detect@*/5min, posthog-event-archive@*/15min, archive-recordings@*/6h, posthog-identity-archive@03:00, sentry-archive@hourly",
   );
+  return worker;
 }