fix(jobs): pre-warm QueueEvents to kill first-sync-wait flake (#285)

The csv-json integration test intermittently timed out at 30000ms on
the first worker-backed job in a fork. Root cause: waitForJob() creates
the BullMQ QueueEvents consumer lazily on first use, and a fresh consumer
reads the Redis events stream from "$" (the tail at the moment its run
loop starts). A trivial tool can publish its completed:<id> event before
the brand-new consumer positions itself, so waitUntilFinished() never
sees the event and blocks for the full sync-wait window. In tests
SYNC_WAIT_MS is floored at 30000ms, exactly the vitest per-test budget,
so the stall surfaces as an opaque timeout instead of a 202 fallback.
This is also a latent production latency bug: the first synchronous tool
request after each boot could hang up to the 8s prod window.

Fix: warmQueueEvents() eagerly constructs and connects every pool's
consumer at spine startup, before any job is enqueued, so each consumer
is positioned at the stream tail up front and never misses a completion.
Awaited in the test spine (deterministic for the first request) and fired
non-blocking at prod boot (a slow Redis must not stall startup).

Adds a regression guard in job-spine.test.ts that drops the cached
consumers, warms explicitly, and asserts a fast job's completion is
captured on the first sync-wait.

Verified: 3 parallel stress runs (276 file-runs across all pools), zero
timeouts; targeted job-spine + csv-json suites green; typecheck clean.

apps/api/src/index.ts

@@ -13,7 +13,7 @@ import { closeDb, db, schema } from "./db/index.js";
 import { runMigrations } from "./db/migrate.js";
 import { startCancelListener, stopCancelListener } from "./jobs/cancel.js";
 import { closeRedis, pingRedis } from "./jobs/connection.js";
-import { closeFlowProducer, closeQueueEvents } from "./jobs/enqueue.js";
+import { closeFlowProducer, closeQueueEvents, warmQueueEvents } from "./jobs/enqueue.js";
 import { closeQueues, perPoolHealth, queueCounts } from "./jobs/queues.js";
 import { enqueueSystemJob, SYSTEM_JOBS, scheduleSystemJobs } from "./jobs/system-jobs.js";
 import { closeWorkers, startWorkers } from "./jobs/worker.js";
@@ -596,6 +596,14 @@ if (await shouldRunStartupCleanup()) {
 // Start BullMQ worker pools (after route registration so the tool registry is full)
 startWorkers();
 
+// Warm the per-pool QueueEvents consumers so the first synchronous tool request
+// after boot does not pay the lazy-connect cost (and cannot miss a fast job's
+// completion event). Non-blocking: a slow/unreachable Redis must not stall boot;
+// the consumers fall back to lazy creation on first use if this has not finished.
+void warmQueueEvents().catch((err) => {
+  app.log.warn({ err }, "QueueEvents warm-up failed; consumers will connect lazily");
+});
+
 // Start
 try {
   await app.listen({ port: env.PORT, host: "0.0.0.0" });

apps/api/src/jobs/enqueue.ts

@@ -12,7 +12,7 @@ import { env } from "../config.js";
 import { db, schema } from "../db/index.js";
 import { createBullMQConnection } from "./connection.js";
 import { getQueue } from "./queues.js";
-import { type Pool, queueName, type ToolJobData, type ToolJobResult } from "./types.js";
+import { POOLS, type Pool, queueName, type ToolJobData, type ToolJobResult } from "./types.js";
 
 // ── QueueEvents (one per pool, lazy) ────────────────────────────
 
@@ -35,6 +35,23 @@ export async function closeQueueEvents(): Promise<void> {
   queueEventsMap.clear();
 }
 
+/**
+ * Eagerly create and connect the QueueEvents consumer for every pool.
+ *
+ * A QueueEvents consumer reads the Redis events stream from "$" (the tail at
+ * the moment its run loop starts). If it is created lazily *inside* the first
+ * waitForJob() call, a fast job can publish its `completed:<id>` event before
+ * the brand-new consumer positions itself at the tail -- the event is then
+ * never delivered and waitUntilFinished() blocks for the full sync-wait window
+ * (SYNC_WAIT_MS). Warming every consumer at spine startup, before any job is
+ * enqueued, positions them at the tail up front so no completion event is ever
+ * missed and the first sync request is as fast as every later one. Idempotent:
+ * getQueueEvents caches one consumer per pool.
+ */
+export async function warmQueueEvents(): Promise<void> {
+  await Promise.all(POOLS.map((pool) => getQueueEvents(pool).waitUntilReady()));
+}
+
 // ── FlowProducer (lazy singleton, used by Task 9) ───────────────
 
 let _flowProducer: FlowProducer | null = null;

tests/integration/platform/job-spine.test.ts

@@ -10,7 +10,12 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import { db, schema } from "../../../apps/api/src/db/index.js";
 import { requestCancel } from "../../../apps/api/src/jobs/cancel.js";
 import { sharedRedis } from "../../../apps/api/src/jobs/connection.js";
-import { enqueueToolJob, waitForJob } from "../../../apps/api/src/jobs/enqueue.js";
+import {
+  closeQueueEvents,
+  enqueueToolJob,
+  waitForJob,
+  warmQueueEvents,
+} from "../../../apps/api/src/jobs/enqueue.js";
 import { bullPrefix, type ToolJobData } from "../../../apps/api/src/jobs/types.js";
 import { putObject } from "../../../apps/api/src/lib/object-storage.js";
 import {
@@ -174,3 +179,43 @@ describe("Job spine", () => {
     expect(parsed.jobId).toBe(jobId);
   });
 });
+
+describe("QueueEvents warm-up (sync-wait flake guard)", () => {
+  it("warmQueueEvents() resolves for all pools and is idempotent", async () => {
+    // First call connects every pool's consumer; the second reuses the cached,
+    // already-ready consumers and must still resolve.
+    await expect(warmQueueEvents()).resolves.toBeUndefined();
+    await expect(warmQueueEvents()).resolves.toBeUndefined();
+  });
+
+  it("a warmed consumer captures a fast job's completion on the first sync-wait", async () => {
+    // Drop the cached consumers to mimic a cold fork, then warm *before*
+    // enqueueing so every consumer is positioned at the events-stream tail up
+    // front. This is the exact invariant that prevents the csv-json 30s flake:
+    // without the warm, a consumer created lazily inside the first waitForJob()
+    // can miss a fast job's `completed` event and block for the whole window.
+    await closeQueueEvents();
+    await warmQueueEvents();
+
+    const jobId = randomUUID();
+    const inputRef = `uploads/${jobId}/warm.png`;
+    await putObject(inputRef, Buffer.from("warm-test"));
+
+    await enqueueToolJob({
+      jobId,
+      toolId: "spine-echo",
+      userId: null,
+      pool: "image",
+      inputRefs: [inputRef],
+      filename: "warm.png",
+      settings: {},
+      kind: "tool",
+    });
+
+    // A warmed consumer observes the completion promptly; a regression (cold or
+    // missed event) would null out only when this window expires.
+    const result = await waitForJob("image", jobId, 10_000);
+    expect(result).not.toBeNull();
+    expect(result!.outputRefs.length).toBeGreaterThan(0);
+  }, 20_000);
+});

tests/integration/test-server.ts

@@ -36,7 +36,7 @@ import {
   stopCancelListener,
 } from "../../apps/api/src/jobs/cancel.js";
 import { pingRedis } from "../../apps/api/src/jobs/connection.js";
-import { closeQueueEvents } from "../../apps/api/src/jobs/enqueue.js";
+import { closeQueueEvents, warmQueueEvents } from "../../apps/api/src/jobs/enqueue.js";
 import { closeWorkers, startWorkers } from "../../apps/api/src/jobs/worker.js";
 import { requirePermission } from "../../apps/api/src/permissions.js";
 import {
@@ -83,6 +83,12 @@ async function ensureSpine(): Promise<void> {
   spineStarted = true;
   await startCancelListener();
   startWorkers();
+  // Position every pool's QueueEvents consumer at the stream tail *before* the
+  // first job is enqueued. Without this, the first sync-wait per fork lazily
+  // creates a consumer that can miss a fast job's completion event and block
+  // for the full SYNC_WAIT_MS window -- the root cause of the csv-json 30s
+  // timeout flake. Awaited here so it is deterministic for the first request.
+  await warmQueueEvents();
 }
 
 // Module-scope afterAll: vitest registers this into any importing file's