From d2aa73af5b6f5298c89c803fc0cc584cdb15b519 Mon Sep 17 00:00:00 2001 From: Brad DerManouelian Date: Tue, 8 Sep 2026 13:53:13 -0500 Subject: [PATCH 1/3] fix(workers): stop each worker bundle from starting the notification worker Job names now live in lib/queueNames.ts. The notification service imported JOB_CREATE_NOTIFICATION from the notification worker module, so esbuild inlined that worker into every bundle that sends a notification, and inside a CommonJS bundle `require.main === module` is true for every inlined module. Every worker process therefore also booted a notification worker, and the budget-alert worker booted lazily inside the LLM-backed workers. The worker build now fails if any bundle inlines a second entry point, and the image smoke test covers the seven workers it was missing. The run-ready check used a fixed job id with removeOnFail: false. BullMQ never re-adds an id it already holds, including one in the failed set, so a single failure blocked every later readiness check for that run. The check now uses a deduplication id with a 5s TTL and leaves failed jobs to the queue's default retention. --- .../api/admin/elasticsearch/reindex/route.ts | 2 +- .../lib/llm/services/llm-manager.service.ts | 6 +-- testplanit/lib/queueNames.ts | 17 +++++++ .../lib/services/notificationService.ts | 2 +- testplanit/lib/services/runReadyCheck.test.ts | 16 +++---- testplanit/lib/services/runReadyCheck.ts | 17 +++++-- testplanit/scheduler.reconcile.test.ts | 17 +------ testplanit/scheduler.ts | 6 +-- testplanit/scripts/build-workers.js | 46 ++++++++++++++++++- testplanit/scripts/smoke-test-workers.js | 7 +++ testplanit/scripts/trigger-forecast-recalc.ts | 2 +- .../trigger-milestone-notifications.ts | 2 +- testplanit/workers/abandonedRunSweep.test.ts | 3 +- testplanit/workers/budgetAlertWorker.ts | 2 - testplanit/workers/forecastWorker.test.ts | 3 +- testplanit/workers/forecastWorker.ts | 25 ++++++---- testplanit/workers/milestoneJobs.test.ts | 3 +- testplanit/workers/notificationWorker.ts | 10 ++-- testplanit/workers/repoCacheWorker.test.ts | 10 ++-- testplanit/workers/repoCacheWorker.ts | 3 +- 20 files changed, 134 insertions(+), 65 deletions(-) diff --git a/testplanit/app/api/admin/elasticsearch/reindex/route.ts b/testplanit/app/api/admin/elasticsearch/reindex/route.ts index a98332dc8..3c668b7ae 100644 --- a/testplanit/app/api/admin/elasticsearch/reindex/route.ts +++ b/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( diff --git a/testplanit/lib/llm/services/llm-manager.service.ts b/testplanit/lib/llm/services/llm-manager.service.ts index 2638bbb77..d3ce803a1 100644 --- a/testplanit/lib/llm/services/llm-manager.service.ts +++ b/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, { diff --git a/testplanit/lib/queueNames.ts b/testplanit/lib/queueNames.ts index 54cdb3274..b4ec292b5 100644 --- a/testplanit/lib/queueNames.ts +++ b/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"; diff --git a/testplanit/lib/services/notificationService.ts b/testplanit/lib/services/notificationService.ts index c624c4478..7da6328fc 100644 --- a/testplanit/lib/services/notificationService.ts +++ b/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 { diff --git a/testplanit/lib/services/runReadyCheck.test.ts b/testplanit/lib/services/runReadyCheck.test.ts index fc241a2f3..7b533a746 100644 --- a/testplanit/lib/services/runReadyCheck.test.ts +++ b/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")); }); }); diff --git a/testplanit/lib/services/runReadyCheck.ts b/testplanit/lib/services/runReadyCheck.ts index 75c04048b..9b54480bb 100644 --- a/testplanit/lib/services/runReadyCheck.ts +++ b/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) { diff --git a/testplanit/scheduler.reconcile.test.ts b/testplanit/scheduler.reconcile.test.ts index ab512fe94..abccd90eb 100644 --- a/testplanit/scheduler.reconcile.test.ts +++ b/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"; diff --git a/testplanit/scheduler.ts b/testplanit/scheduler.ts index 46679dcd2..217775a8d 100644 --- a/testplanit/scheduler.ts +++ b/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) diff --git a/testplanit/scripts/build-workers.js b/testplanit/scripts/build-workers.js index 93c29e2c4..f2586adea 100644 --- a/testplanit/scripts/build-workers.js +++ b/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 diff --git a/testplanit/scripts/smoke-test-workers.js b/testplanit/scripts/smoke-test-workers.js index dad121b7d..383e25dfb 100644 --- a/testplanit/scripts/smoke-test-workers.js +++ b/testplanit/scripts/smoke-test-workers.js @@ -54,6 +54,13 @@ const WORKERS = [ "magicSelectWorker", "stepSequenceScanWorker", "generateFromUrlWorker", + "iterationGenerationWorker", + "scimAccessRecomputeWorker", + "webhookDispatchWorker", + "webhookOutboxWorker", + "webhookRetentionWorker", + "dataChangeLogRetentionWorker", + "datasetLeaseSweepWorker", ]; const entryPoints = [ diff --git a/testplanit/scripts/trigger-forecast-recalc.ts b/testplanit/scripts/trigger-forecast-recalc.ts index 581ab2dac..abe911343 100644 --- a/testplanit/scripts/trigger-forecast-recalc.ts +++ b/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(); diff --git a/testplanit/scripts/trigger-milestone-notifications.ts b/testplanit/scripts/trigger-milestone-notifications.ts index 22c984ac0..42ffe0fc9 100644 --- a/testplanit/scripts/trigger-milestone-notifications.ts +++ b/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(); diff --git a/testplanit/workers/abandonedRunSweep.test.ts b/testplanit/workers/abandonedRunSweep.test.ts index 8acb5f49c..0c9f5d4e1 100644 --- a/testplanit/workers/abandonedRunSweep.test.ts +++ b/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()), FORECAST_QUEUE_NAME: "test-forecast-queue", })); diff --git a/testplanit/workers/budgetAlertWorker.ts b/testplanit/workers/budgetAlertWorker.ts index ce7a268fc..d7c9f85a3 100644 --- a/testplanit/workers/budgetAlertWorker.ts +++ b/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; } diff --git a/testplanit/workers/forecastWorker.test.ts b/testplanit/workers/forecastWorker.test.ts index e020e3c5f..4f58877a1 100644 --- a/testplanit/workers/forecastWorker.test.ts +++ b/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()), FORECAST_QUEUE_NAME: "test-forecast-queue", })); diff --git a/testplanit/workers/forecastWorker.ts b/testplanit/workers/forecastWorker.ts index b657280be..528895c17 100644 --- a/testplanit/workers/forecastWorker.ts +++ b/testplanit/workers/forecastWorker.ts @@ -8,7 +8,15 @@ import { MultiTenantJobData, validateMultiTenantJobData, } from "../lib/multiTenantDb"; -import { FORECAST_QUEUE_NAME } from "../lib/queueNames"; +import { + FORECAST_QUEUE_NAME, + JOB_AUTO_COMPLETE_MILESTONES, + JOB_MILESTONE_DUE_NOTIFICATIONS, + JOB_REVIEW_REMINDERS, + JOB_SWEEP_ABANDONED_RUNS, + JOB_UPDATE_ALL_CASES, + JOB_UPDATE_SINGLE_CASE, +} from "../lib/queueNames"; import { readSystemAbandonedRunIdleMinutes, resolveAbandonedRunTargetStateId, @@ -49,13 +57,14 @@ interface ForecastJobDataBase extends MultiTenantJobData { actorContext?: ActorContextJobData["actorContext"]; } -// Define job names for clarity and export them for the scheduler -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"; +export { + JOB_AUTO_COMPLETE_MILESTONES, + JOB_MILESTONE_DUE_NOTIFICATIONS, + JOB_REVIEW_REMINDERS, + JOB_SWEEP_ABANDONED_RUNS, + JOB_UPDATE_ALL_CASES, + JOB_UPDATE_SINGLE_CASE, +}; /** * Load the name and liveness of a review's subject row. diff --git a/testplanit/workers/milestoneJobs.test.ts b/testplanit/workers/milestoneJobs.test.ts index 278062429..8823eb22d 100644 --- a/testplanit/workers/milestoneJobs.test.ts +++ b/testplanit/workers/milestoneJobs.test.ts @@ -38,7 +38,8 @@ vi.mock("../lib/valkey", () => ({ })); // Mock queue names -vi.mock("../lib/queueNames", () => ({ +vi.mock("../lib/queueNames", async (importOriginal) => ({ + ...(await importOriginal()), FORECAST_QUEUE_NAME: "test-forecast-queue", })); diff --git a/testplanit/workers/notificationWorker.ts b/testplanit/workers/notificationWorker.ts index d4e77dc1b..f7faedad2 100644 --- a/testplanit/workers/notificationWorker.ts +++ b/testplanit/workers/notificationWorker.ts @@ -11,6 +11,11 @@ import { tenantBroadcastChannel, userChannel, } from "../lib/notifications/channels"; +import { + JOB_CREATE_NOTIFICATION, + JOB_PROCESS_USER_NOTIFICATIONS, + JOB_SEND_DAILY_DIGEST, +} from "../lib/queueNames"; import { getEmailQueue, NOTIFICATION_QUEUE_NAME } from "../lib/queues"; import { NotificationService } from "../lib/services/notificationService"; import { resolveRunCompletionRecipients } from "../lib/services/runCompletionRecipients"; @@ -42,11 +47,6 @@ interface SendDailyDigestJobData extends MultiTenantJobData { // No additional fields required } -// Define job names -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"; - const processor = async (job: Job) => { console.log( `Processing notification job ${job.id} of type ${job.name}${job.data.tenantId ? ` for tenant ${job.data.tenantId}` : ""}` diff --git a/testplanit/workers/repoCacheWorker.test.ts b/testplanit/workers/repoCacheWorker.test.ts index f3d938f57..c8c4177f8 100644 --- a/testplanit/workers/repoCacheWorker.test.ts +++ b/testplanit/workers/repoCacheWorker.test.ts @@ -1,7 +1,9 @@ import { Job } from "bullmq"; import { beforeEach, describe, expect, it, vi } from "vitest"; -import { JOB_REFRESH_SINGLE_REPO_CACHE } from "../lib/queueNames"; -import { JOB_REFRESH_EXPIRED_CACHES } from "./repoCacheWorker"; +import { + JOB_REFRESH_EXPIRED_CACHES, + JOB_REFRESH_SINGLE_REPO_CACHE, +} from "../lib/queueNames"; // Create mock db instance const mockDb = { @@ -39,9 +41,9 @@ vi.mock("../lib/services/repoCacheRefreshService", () => ({ })); // Mock queue names -vi.mock("../lib/queueNames", () => ({ +vi.mock("../lib/queueNames", async (importOriginal) => ({ + ...(await importOriginal()), REPO_CACHE_QUEUE_NAME: "test-repo-cache-queue", - JOB_REFRESH_SINGLE_REPO_CACHE: "refresh-single-repo-cache", })); const mockConfigs = [ diff --git a/testplanit/workers/repoCacheWorker.ts b/testplanit/workers/repoCacheWorker.ts index 417945f77..ce1e6a883 100644 --- a/testplanit/workers/repoCacheWorker.ts +++ b/testplanit/workers/repoCacheWorker.ts @@ -7,6 +7,7 @@ import { validateMultiTenantJobData, } from "../lib/multiTenantDb"; import { + JOB_REFRESH_EXPIRED_CACHES, JOB_REFRESH_SINGLE_REPO_CACHE, REPO_CACHE_QUEUE_NAME, } from "../lib/queueNames"; @@ -15,8 +16,6 @@ import { withTenantContext } from "../lib/tenantContext"; import valkeyConnection from "../lib/valkey"; import { BULLMQ_PREFIX } from "../lib/bullPrefix"; -export const JOB_REFRESH_EXPIRED_CACHES = "refresh-expired-repo-caches"; - const processor = async (job: Job) => { console.log( `Processing job ${job.id} of type ${job.name}${job.data.tenantId ? ` for tenant ${job.data.tenantId}` : ""}` From f3340933c1afc27cfeedf92999a17272f493cad6 Mon Sep 17 00:00:00 2001 From: Brad DerManouelian Date: Tue, 8 Sep 2026 15:16:19 -0500 Subject: [PATCH 2/3] fix(docker): keep container boot off the network and unstick fresh-database migrations The entrypoint called zenstack and tsx through npx. npm's first run in a new container asks the registry for a newer version and, without internet egress, waits out that request before the migration step starts. The globally installed binaries are now called directly, and Prisma's telemetry beacon is switched off. The migrate step also sets DEBUG=prisma:engines. On arm64 pods the first deploy against an empty database has hung indefinitely before the schema engine starts, and enabling that debug namespace is the only known way past it. The Helm migrate job gets the same two settings. --- testplanit/docker-entrypoint.sh | 18 +++++++++++++++--- .../helm/testplanit/templates/migrate-job.yaml | 5 ++++- 2 files changed, 19 insertions(+), 4 deletions(-) diff --git a/testplanit/docker-entrypoint.sh b/testplanit/docker-entrypoint.sh index 319545d24..f1009ca6b 100755 --- a/testplanit/docker-entrypoint.sh +++ b/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 "$@" diff --git a/testplanit/helm/testplanit/templates/migrate-job.yaml b/testplanit/helm/testplanit/templates/migrate-job.yaml index 12b6f2a27..4a94f83e5 100644 --- a/testplanit/helm/testplanit/templates/migrate-job.yaml +++ b/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..." From 696145c878f4bd32eb5d9e0e3540dcd82b0f41a8 Mon Sep 17 00:00:00 2001 From: Brad DerManouelian Date: Tue, 8 Sep 2026 15:22:45 -0500 Subject: [PATCH 3/3] fix(docker): install the ZenStack CLI version the lockfile resolves The production image pinned the global @zenstackhq/cli at 3.8.0 while the project's devDependency had moved to 3.9.3. The base stage now records the version its install resolved, and the production stage installs exactly that, so the CLI that runs migrate deploy at boot cannot drift from the one the project develops and tests against. --- testplanit/Dockerfile | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/testplanit/Dockerfile b/testplanit/Dockerfile index cab6591ca..afd613d21 100644 --- a/testplanit/Dockerfile +++ b/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