Daniel Sutton
e4ae8cbcd4
## Summary On the run-ops database split, a run that waits (`triggerAndWait`, `batchTriggerAndWait`, `wait.forToken`) could hang forever after its wait had already completed. The runner reads a resume from `/snapshots/since` exactly once: if that read returned the resume snapshot without its completed-waitpoints, the runner logged "executing without completed waitpoints", advanced its cursor, and never re-read it, so the awaiting run never continued. ## Root cause The resume snapshot and its completed-waitpoint rows were written as two separate commits. This regressed when the split replaced Prisma's atomic nested `connect` with an FK-free insert (in [#4163](https://github.com/triggerdotdev/trigger.dev/pull/4163)), and `/snapshots/since` is served from a read replica. A fetch landing in the sub-millisecond gap between the two commits, or a multi-reader replica serving the snapshot from a different point in time than its join rows, delivered an empty resume. Because the runner consumes each snapshot once and treats an empty resume as terminal, a single stale read was fatal and produced a permanent, nondeterministic hang. ## Fixes - Commit a snapshot and its completed-waitpoint links in one transaction, restoring the atomicity the split removed. - Repair the completed-waitpoints from the owning primary when a multi-reader replica serves the snapshot without its join rows. This covers single-waitpoint resumes, which carry no `completedWaitpointOrder` and so were missed by the count-based repair. - Read the primary in the checkpoint `WAIT_FOR_BATCH` pre-check, so a batch that already resumed is not re-suspended into a stall. - Fall back to the primary when a waitpoint token misses both read replicas, so a token completed immediately after it was minted no longer returns a spurious 404. - Route batch-item creation by `batchTaskRunId`, consistent with the batch-completion count and the row's foreign key. - Reject control-plane-only relation selects on the dedicated schema with a clear error instead of an opaque Prisma failure, and stop `createDateTimeWaitpoint` bypassing residency routing through a caller transaction. Verified against the deployed split topology: a resume snapshot and its completed-waitpoints are now always delivered together, so the runner can no longer drop a resume.
66 lines
2.8 KiB
TypeScript
66 lines
2.8 KiB
TypeScript
// Unit red-green for the checkpoint WAIT_FOR_BATCH replica-lag fix (createCheckpoint.server.ts).
|
|
// The service decides whether to suspend a run on `batchRun.resumedAt`; reading it from a lagging
|
|
// replica makes a just-resumed batch look unresumed -> it suspends an already-resumed run -> stall.
|
|
// The fix threads the primary (`this._prisma`) into `runStore.findBatchTaskRunByFriendlyId`. Here a
|
|
// spy runStore records which client the service passed and simulates the lag (only the primary read
|
|
// sees the fresh resumedAt): RED = no client -> stale null -> no early return; GREEN = primary -> kept alive.
|
|
|
|
import { describe, expect, it, vi } from "vitest";
|
|
|
|
vi.mock("~/services/logger.server", () => ({
|
|
logger: { debug: vi.fn(), info: vi.fn(), log: vi.fn(), error: vi.fn(), warn: vi.fn() },
|
|
}));
|
|
vi.mock("~/v3/marqs/index.server", () => ({
|
|
marqs: { replaceMessage: vi.fn(), cancelHeartbeat: vi.fn() },
|
|
}));
|
|
|
|
import { CreateCheckpointService } from "~/v3/services/createCheckpoint.server";
|
|
|
|
describe("checkpoint WAIT_FOR_BATCH reads the primary, not a lagging replica", () => {
|
|
it("threads the primary so an already-resumed batch keeps the run alive", async () => {
|
|
// A freezable attempt so control reaches the WAIT_FOR_BATCH arm. This object IS the primary the
|
|
// fix must thread into the batch read.
|
|
const prisma = {
|
|
taskRunAttempt: {
|
|
findFirst: async () => ({
|
|
id: "attempt_1",
|
|
status: "EXECUTING",
|
|
taskRunId: "run_1",
|
|
taskRun: { id: "run_1", status: "EXECUTING", runtimeEnvironmentId: "env_1" },
|
|
backgroundWorker: { id: "bw_1", deployment: { imageReference: "img:1" } },
|
|
}),
|
|
},
|
|
};
|
|
|
|
let seenClient: unknown = "NOT_CALLED";
|
|
const runStore = {
|
|
findBatchTaskRunByFriendlyId: async (
|
|
_friendlyId: string,
|
|
_environmentId: string,
|
|
_args: unknown,
|
|
client?: unknown
|
|
) => {
|
|
seenClient = client;
|
|
// Lagging replica: only a read handed the primary sees the just-committed resumedAt.
|
|
return { resumedAt: client === prisma ? new Date() : null };
|
|
},
|
|
};
|
|
|
|
const service = new CreateCheckpointService(prisma as never, {} as never, runStore as never);
|
|
|
|
let result: unknown;
|
|
try {
|
|
result = await service.call({
|
|
attemptFriendlyId: "attempt_1",
|
|
reason: { type: "WAIT_FOR_BATCH", batchFriendlyId: "batch_1" },
|
|
} as never);
|
|
} catch {
|
|
// Buggy path falls through the pre-check into checkpoint creation (unstubbed) and throws; the
|
|
// recorded client below is what distinguishes RED from GREEN.
|
|
}
|
|
|
|
expect(seenClient).toBe(prisma); // the fix: primary threaded into the batch read
|
|
expect(result).toEqual({ success: false, keepRunAlive: true }); // early-return, run kept alive
|
|
});
|
|
});
|