mirror of
https://github.com/snapotter-hq/SnapOtter.git
synced 2026-08-03 07:46:42 +02:00
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.
This commit is contained in:
@@ -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);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user