fix(deploy): stop worker bundles starting extra workers and unstick migrate deploy on fresh databases

testplanit/Dockerfile

@@ -56,6 +56,9 @@ ENV NODE_OPTIONS="--max-old-space-size=8192"
 # `next build` cannot resolve `~/zenstack/models`. (Install runs with
 # --ignore-scripts, so the postinstall generate never fires here.)
 RUN pnpm zenstack generate --schema schema.zmodel -o zenstack
+# The production stage installs the same CLI globally for `migrate deploy` at
+# boot; recording the resolved version here keeps the two from drifting.
+RUN node -p "require('./node_modules/@zenstackhq/cli/package.json').version" > /app/zenstack-cli-version
 RUN apk del .gyp
 
 COPY testplanit/wait-for-postgres.sh /usr/local/bin/wait-for-postgres.sh
@@ -214,10 +217,13 @@ RUN chmod +x ./testplanit/docker-entrypoint.sh
 
 WORKDIR /app/testplanit
 RUN npm install -g tsx
-# ZenStack CLI for the db-init `zenstack db push` (the prod deploy tree is
-# --prod, so the devDep CLI isn't present; install it globally). It pulls
-# Prisma's schema engine internally to perform the push.
-RUN npm install -g @zenstackhq/cli@3.8.0
+# ZenStack CLI for `zenstack migrate deploy` in docker-entrypoint.sh (the prod
+# deploy tree is --prod, so the devDep CLI isn't present; install it globally).
+# The version is the one the workspace lockfile resolved, recorded by the base
+# stage, so the boot-time CLI always matches the project's devDependency. It
+# pulls Prisma's schema engine internally to apply the migrations.
+COPY --from=build /app/zenstack-cli-version /tmp/zenstack-cli-version
+RUN npm install -g "@zenstackhq/cli@$(cat /tmp/zenstack-cli-version)" && rm /tmp/zenstack-cli-version
 
 USER nextjs
 EXPOSE 3000

testplanit/app/api/admin/elasticsearch/reindex/route.ts

@@ -10,7 +10,7 @@ import {
 } from "~/lib/auditContextWrappers";
 import { getServerAuthSession } from "~/server/auth";
 import { getElasticsearchClient } from "~/services/elasticsearchService";
-import { ReindexJobData } from "~/workers/elasticsearchReindexWorker";
+import type { ReindexJobData } from "~/workers/elasticsearchReindexWorker";
 
 // Helper to check admin authentication (session or API token)
 async function checkAdminAuth(

testplanit/docker-entrypoint.sh

@@ -5,17 +5,29 @@ set -e
 # falls back to DATABASE_URL when DIRECT_DATABASE_URL is unset (no pooler).
 INIT_DATABASE_URL="${DIRECT_DATABASE_URL:-$DATABASE_URL}"
 
+# Boot must not depend on internet egress. zenstack and tsx are installed
+# globally in the image and are called directly: going through npx makes the
+# first run in every new container ask the npm registry for a newer version
+# and wait out that request when the pod cannot reach it. CHECKPOINT_DISABLE
+# turns off Prisma's telemetry beacon for the same reason.
+#
+# DEBUG=prisma:engines is a workaround, not diagnostics. On arm64 pods the
+# first deploy against an empty database has been seen to stop before the
+# schema engine starts, with no error and no timeout, and enabling this debug
+# namespace is the only known way to get it moving. It prints one or two
+# extra lines at boot.
 echo "Running database migrations..."
 # migrate deploy applies pending migrations only; it never drops data (unlike
 # `db push --accept-data-loss`). Existing databases first built with db push must
 # have the baseline marked applied once — see testplanit/migrations/README.md.
-DATABASE_URL="$INIT_DATABASE_URL" npx zenstack migrate deploy --schema schema.zmodel --no-version-check
+DATABASE_URL="$INIT_DATABASE_URL" CHECKPOINT_DISABLE=1 DEBUG=prisma:engines \
+  zenstack migrate deploy --schema schema.zmodel --no-version-check
 
 echo "Applying audit triggers..."
-DATABASE_URL="$INIT_DATABASE_URL" npx tsx scripts/apply-triggers.ts
+DATABASE_URL="$INIT_DATABASE_URL" tsx scripts/apply-triggers.ts
 
 echo "Setting up PostgreSQL extensions..."
-DATABASE_URL="$INIT_DATABASE_URL" npx tsx db/setup-extensions.ts
+DATABASE_URL="$INIT_DATABASE_URL" tsx db/setup-extensions.ts
 
 echo "Starting application..."
 exec "$@"

testplanit/helm/testplanit/templates/migrate-job.yaml

@@ -45,7 +45,10 @@ spec:
               # Schema sync + DDL run on the direct (non-pooled) connection.
               export DATABASE_URL="${DIRECT_DATABASE_URL:-$DATABASE_URL}"
               echo "==> Applying migrations..."
-              zenstack migrate deploy --schema schema.zmodel --no-version-check
+              # Same guards as docker-entrypoint.sh: no telemetry egress, and
+              # the arm64 fresh-database workaround.
+              CHECKPOINT_DISABLE=1 DEBUG=prisma:engines \
+                zenstack migrate deploy --schema schema.zmodel --no-version-check
               echo "==> Applying audit triggers..."
               tsx scripts/apply-triggers.ts
               echo "==> Setting up PostgreSQL extensions..."

testplanit/lib/llm/services/llm-manager.service.ts

@@ -649,8 +649,7 @@ export class LlmManager {
     if (config.monthlyBudget && Number(config.monthlyBudget) > 0) {
       try {
         const { getBudgetAlertQueue } = await import("~/lib/queues");
-        const { BUDGET_ALERT_JOB_CHECK } =
-          await import("~/workers/budgetAlertWorker");
+        const { BUDGET_ALERT_JOB_CHECK } = await import("~/lib/queueNames");
         const { getCurrentTenantId } = await import("~/lib/multiTenantDb");
         getBudgetAlertQueue()
           ?.add(BUDGET_ALERT_JOB_CHECK, {
@@ -710,8 +709,7 @@ export class LlmManager {
     if (config.monthlyBudget && Number(config.monthlyBudget) > 0) {
       try {
         const { getBudgetAlertQueue } = await import("~/lib/queues");
-        const { BUDGET_ALERT_JOB_CHECK } =
-          await import("~/workers/budgetAlertWorker");
+        const { BUDGET_ALERT_JOB_CHECK } = await import("~/lib/queueNames");
         const { getCurrentTenantId } = await import("~/lib/multiTenantDb");
         getBudgetAlertQueue()
           ?.add(BUDGET_ALERT_JOB_CHECK, {

testplanit/lib/queueNames.ts

@@ -22,3 +22,20 @@ export const GENERATE_FROM_URL_QUEUE_NAME = "generate-from-url";
 export const ITERATION_GENERATION_QUEUE_NAME = "iteration-generation";
 export const WEBHOOK_DISPATCH_QUEUE_NAME = "webhook-dispatch";
 export const SCIM_ACCESS_RECOMPUTE_QUEUE_NAME = "scim-access-recompute";
+
+// Job names shared between enqueue sites and the workers that process them.
+// They live here rather than in the worker modules so that app code, services
+// and the scheduler never import a worker entry file: esbuild inlines whatever
+// a bundle imports, and an inlined worker's `require.main === module` start
+// guard is true inside the bundle that swallowed it.
+export const JOB_CREATE_NOTIFICATION = "create-notification";
+export const JOB_PROCESS_USER_NOTIFICATIONS = "process-user-notifications";
+export const JOB_SEND_DAILY_DIGEST = "send-daily-digest";
+export const BUDGET_ALERT_JOB_CHECK = "check-budget";
+export const JOB_REFRESH_EXPIRED_CACHES = "refresh-expired-repo-caches";
+export const JOB_UPDATE_SINGLE_CASE = "update-single-case-forecast";
+export const JOB_UPDATE_ALL_CASES = "update-all-cases-forecast";
+export const JOB_AUTO_COMPLETE_MILESTONES = "auto-complete-milestones";
+export const JOB_MILESTONE_DUE_NOTIFICATIONS = "milestone-due-notifications";
+export const JOB_REVIEW_REMINDERS = "review-reminders";
+export const JOB_SWEEP_ABANDONED_RUNS = "sweep-abandoned-runs";

testplanit/lib/services/notificationService.ts

@@ -1,6 +1,6 @@
 import { ApplicationArea, NotificationType } from "~/zenstack/models";
-import { JOB_CREATE_NOTIFICATION } from "../../workers/notificationWorker";
 import { getCurrentTenantId } from "../multiTenantDb";
+import { JOB_CREATE_NOTIFICATION } from "../queueNames";
 import { getNotificationQueue } from "../queues";
 
 interface CreateNotificationParams {

testplanit/lib/services/runReadyCheck.test.ts

@@ -13,7 +13,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 
 vi.mock("~/lib/queues", () => ({ getNotificationQueue: () => null }));
 
-const { claimRunReadyTransition, evaluateRunReadiness, runReadyCheckJobId } =
+const { claimRunReadyTransition, evaluateRunReadiness, runReadyDedupId } =
   await import("./runReadyCheck");
 
 interface Counts {
@@ -56,15 +56,13 @@ const READY_RUN = {
 
 beforeEach(() => vi.clearAllMocks());
 
-describe("runReadyCheckJobId", () => {
-  // The job id is the debounce: a bulk submission touching hundreds of cases
-  // must collapse onto one evaluation per run.
+describe("runReadyDedupId", () => {
+  // The deduplication id is the debounce: a bulk submission touching hundreds
+  // of cases must collapse onto one evaluation per run.
   it("is stable per run and tenant", () => {
-    expect(runReadyCheckJobId(42, "acme")).toBe("runready:acme:42");
-    expect(runReadyCheckJobId(42, undefined)).toBe("runready:default:42");
-    expect(runReadyCheckJobId(42, "acme")).not.toBe(
-      runReadyCheckJobId(43, "acme")
-    );
+    expect(runReadyDedupId(42, "acme")).toBe("runready:acme:42");
+    expect(runReadyDedupId(42, undefined)).toBe("runready:default:42");
+    expect(runReadyDedupId(42, "acme")).not.toBe(runReadyDedupId(43, "acme"));
   });
 });
 

testplanit/lib/services/runReadyCheck.ts

@@ -30,7 +30,7 @@ export const JOB_CHECK_RUN_READY = "check-run-ready";
  * Evaluation is deferred by this much so the worker reads committed state —
  * the plugin hook that enqueues it runs inside the writer's transaction.
  * Doubles as a debounce: a bulk submission touching hundreds of cases
- * collapses onto one job id and evaluates once.
+ * collapses onto one job and evaluates once.
  */
 const READY_CHECK_DELAY_MS = 5000;
 
@@ -39,7 +39,14 @@ export interface RunReadyCheckJobData {
   tenantId?: string;
 }
 
-export function runReadyCheckJobId(
+/**
+ * Deduplication id for a run's pending check. While a job carrying this id is
+ * queued, further enqueues for the same run are dropped; the id expires with
+ * the delay, so a check that fails never blocks the next one. The job id is
+ * left to BullMQ on purpose: a fixed job id would also collide with a failed
+ * job kept for inspection, and BullMQ never re-adds an id it already holds.
+ */
+export function runReadyDedupId(
   runId: number,
   tenantId: string | undefined
 ): string {
@@ -63,10 +70,12 @@ export async function enqueueRunReadyCheck(
       JOB_CHECK_RUN_READY,
       { runId, tenantId } satisfies RunReadyCheckJobData,
       {
-        jobId: runReadyCheckJobId(runId, tenantId),
+        deduplication: {
+          id: runReadyDedupId(runId, tenantId),
+          ttl: READY_CHECK_DELAY_MS,
+        },
         delay: READY_CHECK_DELAY_MS,
         removeOnComplete: true,
-        removeOnFail: false,
       }
     );
   } catch (error) {

testplanit/scheduler.reconcile.test.ts

@@ -2,7 +2,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
 
 // Unit tests for reconcileStaleSchedulers (scheduler.ts). The function is
 // exercised directly with fake queues; scheduler.ts's import graph is mocked
-// so the test never touches Valkey, Prisma, or the worker modules.
+// so the test never touches Valkey or Prisma.
 
 vi.mock("./lib/queues", () => ({
   FORECAST_QUEUE_NAME: "forecast-updates",
@@ -20,21 +20,6 @@ vi.mock("./lib/multiTenantDb", () => ({
   isMultiTenantMode: () => true,
 }));
 
-vi.mock("./workers/forecastWorker", () => ({
-  JOB_UPDATE_ALL_CASES: "update-all-cases-forecast",
-  JOB_AUTO_COMPLETE_MILESTONES: "auto-complete-milestones",
-  JOB_MILESTONE_DUE_NOTIFICATIONS: "milestone-due-notifications",
-  JOB_REVIEW_REMINDERS: "review-reminders",
-}));
-
-vi.mock("./workers/notificationWorker", () => ({
-  JOB_SEND_DAILY_DIGEST: "send-daily-digest",
-}));
-
-vi.mock("./workers/repoCacheWorker", () => ({
-  JOB_REFRESH_EXPIRED_CACHES: "refresh-expired-caches",
-}));
-
 import { reconcileStaleSchedulers } from "./scheduler";
 
 const JOB = "update-all-cases-forecast";

testplanit/scheduler.ts

@@ -12,12 +12,12 @@ import {
 import {
   JOB_AUTO_COMPLETE_MILESTONES,
   JOB_MILESTONE_DUE_NOTIFICATIONS,
+  JOB_REFRESH_EXPIRED_CACHES,
   JOB_REVIEW_REMINDERS,
+  JOB_SEND_DAILY_DIGEST,
   JOB_SWEEP_ABANDONED_RUNS,
   JOB_UPDATE_ALL_CASES,
-} from "./workers/forecastWorker";
-import { JOB_SEND_DAILY_DIGEST } from "./workers/notificationWorker";
-import { JOB_REFRESH_EXPIRED_CACHES } from "./workers/repoCacheWorker";
+} from "./lib/queueNames";
 
 // Define the cron schedule (e.g., every day at 3:00 AM server time)
 // Uses standard cron syntax: min hour day(month) month day(week)

testplanit/scripts/build-workers.js

@@ -44,11 +44,52 @@ const entryPoints = [
   "scheduler.ts",
 ];
 
+/**
+ * Fail the build if any bundle inlined another entry point's source file.
+ *
+ * Every worker starts itself behind `require.main === module`. Inside a CJS
+ * bundle that test is true for every inlined module, not just the entry, so
+ * a bundle that swallows a second worker boots both in one process: the
+ * extra worker steals that queue's jobs and its SIGTERM handler races the
+ * real one to `process.exit`. Shared job names belong in lib/queueNames.ts.
+ */
+function assertOneEntryPerBundle(metafile) {
+  const entries = new Set(entryPoints.map((p) => path.normalize(p)));
+  const problems = [];
+
+  for (const [outFile, output] of Object.entries(metafile.outputs)) {
+    if (!output.entryPoint) continue;
+    const own = path.normalize(output.entryPoint);
+
+    for (const input of Object.keys(output.inputs)) {
+      const normalized = path.normalize(input);
+      if (normalized === own || !entries.has(normalized)) continue;
+
+      const importers = Object.entries(metafile.inputs)
+        .filter(([, meta]) =>
+          meta.imports.some((imp) => path.normalize(imp.path) === normalized)
+        )
+        .map(([file]) => file);
+      problems.push(
+        `${outFile} inlines ${input} (imported by ${importers.join(", ") || "unknown"})`
+      );
+    }
+  }
+
+  if (problems.length > 0) {
+    console.error("✗ Each worker bundle must contain exactly one entry point:");
+    for (const problem of problems) {
+      console.error(`  - ${problem}`);
+    }
+    process.exit(1);
+  }
+}
+
 async function build() {
   try {
     console.log("Building workers...");
 
-    await esbuild.build({
+    const result = await esbuild.build({
       entryPoints,
       bundle: true, // Bundle to resolve all imports
       platform: "node",
@@ -60,8 +101,11 @@ async function build() {
       tsconfig: path.join(rootDir, "tsconfig.workers.json"),
       packages: "external", // Don't bundle node_modules, treat them as external
       logLevel: "info",
+      metafile: true,
     });
 
+    assertOneEntryPerBundle(result.metafile);
+
     console.log("✓ Workers built successfully");
 
     // Copy email templates to dist directory

testplanit/scripts/smoke-test-workers.js

@@ -54,6 +54,13 @@ const WORKERS = [
   "magicSelectWorker",
   "stepSequenceScanWorker",
   "generateFromUrlWorker",
+  "iterationGenerationWorker",
+  "scimAccessRecomputeWorker",
+  "webhookDispatchWorker",
+  "webhookOutboxWorker",
+  "webhookRetentionWorker",
+  "dataChangeLogRetentionWorker",
+  "datasetLeaseSweepWorker",
 ];
 
 const entryPoints = [

testplanit/scripts/trigger-forecast-recalc.ts

@@ -1,7 +1,7 @@
 import { enqueueWithAuditContext } from "../lib/auditContextEnqueue";
 import { getAllTenantIds, isMultiTenantMode } from "../lib/multiTenantDb";
+import { JOB_UPDATE_ALL_CASES } from "../lib/queueNames";
 import { getForecastQueue } from "../lib/queues";
-import { JOB_UPDATE_ALL_CASES } from "../workers/forecastWorker";
 
 async function triggerForecastRecalculation() {
   const forecastQueue = getForecastQueue();

testplanit/scripts/trigger-milestone-notifications.ts

@@ -1,7 +1,7 @@
 import { enqueueWithAuditContext } from "../lib/auditContextEnqueue";
 import { getAllTenantIds, isMultiTenantMode } from "../lib/multiTenantDb";
+import { JOB_MILESTONE_DUE_NOTIFICATIONS } from "../lib/queueNames";
 import { getForecastQueue } from "../lib/queues";
-import { JOB_MILESTONE_DUE_NOTIFICATIONS } from "../workers/forecastWorker";
 
 async function triggerMilestoneNotifications() {
   const forecastQueue = getForecastQueue();

testplanit/workers/abandonedRunSweep.test.ts

@@ -40,7 +40,8 @@ vi.mock("../lib/valkey", () => ({
   default: null,
 }));
 
-vi.mock("../lib/queueNames", () => ({
+vi.mock("../lib/queueNames", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../lib/queueNames")>()),
   FORECAST_QUEUE_NAME: "test-forecast-queue",
 }));
 

testplanit/workers/budgetAlertWorker.ts

@@ -12,8 +12,6 @@ import { withTenantContext } from "../lib/tenantContext";
 import valkeyConnection from "../lib/valkey";
 import { BULLMQ_PREFIX } from "../lib/bullPrefix";
 
-export const BUDGET_ALERT_JOB_CHECK = "check-budget";
-
 interface BudgetCheckJobData extends MultiTenantJobData {
   llmIntegrationId: number;
 }

testplanit/workers/forecastWorker.test.ts

@@ -78,7 +78,8 @@ vi.mock("../lib/valkey", () => ({
 }));
 
 // Mock queue names
-vi.mock("../lib/queueNames", () => ({
+vi.mock("../lib/queueNames", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../lib/queueNames")>()),
   FORECAST_QUEUE_NAME: "test-forecast-queue",
 }));