Skip to content

Commit f088bea

Browse files
d-csd-cs
committed
test: apply CodeRabbit review fixes (engine cleanup, seed tx, post-fix narratives)
- quit the RunEngine when a test helper's post-construction setup throws, so a failed setup can't leak handles into later tests - wrap the batch-family seed in a transaction with SET LOCAL session_replication_role so the FK-trigger disable stays on the connection that runs the inserts (session-scoped setting + pooled connections) - reword store-seam test comments/titles that still described now-fixed paths as outstanding bugs; assertions unchanged
1 parent df6b995 commit f088bea

6 files changed

Lines changed: 132 additions & 108 deletions

‎internal-packages/run-engine/src/engine/tests/runAttemptSystemReplicaLag.guard.test.ts‎

Lines changed: 56 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -151,24 +151,34 @@ async function triggerAndStartAttempt(
151151
...baseEngineOptions(redisOptions),
152152
});
153153

154-
const taskIdentifier = "test-task";
155-
await setupBackgroundWorker(engine, environment, taskIdentifier, undefined, retryOptions);
156-
157-
const run = await engine.trigger(triggerParams(friendlyId, environment, taskIdentifier), prisma);
158-
159-
await setTimeout(500);
160-
const dequeued = await engine.dequeueFromWorkerQueue({
161-
consumerId: "test_guard",
162-
workerQueue: "main",
163-
});
164-
expect(dequeued.length).toBe(1);
165-
166-
const attempt = await engine.startRunAttempt({
167-
runId: dequeued[0].run.id,
168-
snapshotId: dequeued[0].snapshot.id,
169-
});
170-
171-
return { engine, environment, run, attempt, store, replica };
154+
try {
155+
const taskIdentifier = "test-task";
156+
await setupBackgroundWorker(engine, environment, taskIdentifier, undefined, retryOptions);
157+
158+
const run = await engine.trigger(
159+
triggerParams(friendlyId, environment, taskIdentifier),
160+
prisma
161+
);
162+
163+
await setTimeout(500);
164+
const dequeued = await engine.dequeueFromWorkerQueue({
165+
consumerId: "test_guard",
166+
workerQueue: "main",
167+
});
168+
expect(dequeued.length).toBe(1);
169+
170+
const attempt = await engine.startRunAttempt({
171+
runId: dequeued[0].run.id,
172+
snapshotId: dequeued[0].snapshot.id,
173+
});
174+
175+
return { engine, environment, run, attempt, store, replica };
176+
} catch (e) {
177+
// Setup threw after the engine was constructed: quit it here so a failed helper does not leak the
178+
// engine (the callers only run `await engine.quit()` on the success path).
179+
await engine.quit();
180+
throw e;
181+
}
172182
}
173183

174184
describe("RunAttemptSystem completion-path reads must hit the owning primary, not a lagging replica", () => {
@@ -407,24 +417,34 @@ async function triggerAndStartAttemptGated(
407417
...baseEngineOptions(redisOptions),
408418
});
409419

410-
const taskIdentifier = "test-task";
411-
await setupBackgroundWorker(engine, environment, taskIdentifier, undefined, retryOptions);
412-
413-
const run = await engine.trigger(triggerParams(friendlyId, environment, taskIdentifier), prisma);
414-
415-
await setTimeout(500);
416-
const dequeued = await engine.dequeueFromWorkerQueue({
417-
consumerId: "test_guard",
418-
workerQueue: "main",
419-
});
420-
expect(dequeued.length).toBe(1);
421-
422-
const attempt = await engine.startRunAttempt({
423-
runId: dequeued[0].run.id,
424-
snapshotId: dequeued[0].snapshot.id,
425-
});
426-
427-
return { engine, environment, run, attempt, store, replica };
420+
try {
421+
const taskIdentifier = "test-task";
422+
await setupBackgroundWorker(engine, environment, taskIdentifier, undefined, retryOptions);
423+
424+
const run = await engine.trigger(
425+
triggerParams(friendlyId, environment, taskIdentifier),
426+
prisma
427+
);
428+
429+
await setTimeout(500);
430+
const dequeued = await engine.dequeueFromWorkerQueue({
431+
consumerId: "test_guard",
432+
workerQueue: "main",
433+
});
434+
expect(dequeued.length).toBe(1);
435+
436+
const attempt = await engine.startRunAttempt({
437+
runId: dequeued[0].run.id,
438+
snapshotId: dequeued[0].snapshot.id,
439+
});
440+
441+
return { engine, environment, run, attempt, store, replica };
442+
} catch (e) {
443+
// Setup threw after the engine was constructed: quit it here so a failed helper does not leak the
444+
// engine (the callers only run `await engine.quit()` on the success path).
445+
await engine.quit();
446+
throw e;
447+
}
428448
}
429449

430450
describe("RunEngine engine-other reads must hit the owning primary, not a lagging replica", () => {

‎internal-packages/run-store/src/runOpsStore.routesRealtimeStreamReadView.replicaLag.test.ts‎

Lines changed: 19 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -28,18 +28,18 @@
2828
// reads by ≫ replica lag, and the only fields read off the replica are immutable (id/friendlyId)
2929
// or fail safe (a stale completedAt=null only ALLOWS an append, and the append/PUT completedAt gate
3030
// that actually blocks re-reads the PRIMARY: $target.$streamId.ts:69, append.ts:69 use `prisma`).
31-
// 1 real bug — realtime.v1.runs.$runId.ts:18 (subscribeToRun's findResource). This is the one case in
32-
// a genuine read-your-writes window: `runs.trigger()` commits the run to the control-plane PRIMARY,
33-
// then user code immediately `runs.subscribeToRun(handle.id)` reads the REPLICA. Under lag → null →
34-
// apiBuilder returns 404 (the route sets no `shouldRetryNotFound`, so `x-should-retry: "false"`).
35-
// The subscription does NOT recover: the Electric client retries ONLY 429
36-
// (@electric-sql/client fetch.ts `HTTP_RETRY_STATUS_CODES = [429]`; a 4xx throws), and
37-
// RunSubscription closes terminally on that FetchError (core apiClient/stream.ts #handleError sets
38-
// #isStreamClosed=true and unsubscribes). Net: a spurious permanent 404 of a live, just-triggered
39-
// run. There is no primary fallback here (contrast the mollifier mutation resolver, which adds a
40-
// writer re-probe at resolveRunForMutation.server.ts:65 precisely to survive this). Recommended
41-
// fix: in findResource, re-read the OWNING PRIMARY when the replica returns null (pass the writer
42-
// client to runStore.findRun → findRunOnPrimary), or gate subscription auth on a primary read.
31+
// 1 read-your-writes window — realtime.v1.runs.$runId.ts:18 (subscribeToRun's findResource). This is
32+
// the one case with a genuine read-your-writes window: `runs.trigger()` commits the run to the
33+
// control-plane PRIMARY, then user code immediately `runs.subscribeToRun(handle.id)` reads the
34+
// REPLICA. The replica miss is only the FIRST leg: a bare replica null would make apiBuilder return
35+
// 404 (the route sets no `shouldRetryNotFound`, so `x-should-retry: "false"`) and the subscription
36+
// would not recover (the Electric client retries ONLY 429 — @electric-sql/client fetch.ts
37+
// `HTTP_RETRY_STATUS_CODES = [429]`, a 4xx throws — and RunSubscription closes terminally on that
38+
// FetchError: core apiClient/stream.ts #handleError sets #isStreamClosed=true and unsubscribes). So
39+
// findResource re-reads the OWNING PRIMARY when the replica returns null (the same writer re-probe
40+
// the mollifier mutation resolver uses at resolveRunForMutation.server.ts:65), which resolves the
41+
// just-triggered run. This test pins both legs: the replica returns null under lag, and the
42+
// owning-primary re-read recovers the run.
4343
//
4444
// Real split topology via heteroRunOpsPostgresTest — never mocked, built the same way as
4545
// runOpsStore.engineReadViews.replicaLag.test.ts (heteroRunOpsPostgresTest + shared laggingReplica).
@@ -160,12 +160,12 @@ async function setupLaggingLegacy(
160160
describe("run-ops split — realtime-STREAM route read views vs. a lagging replica", () => {
161161
// realtime.v1.runs.$runId.ts:18 findRun — subscribeToRun's findResource.
162162
// where {friendlyId, runtimeEnvironmentId}, include {batch:{select:{friendlyId}}}, BRANDED $replica.
163-
// The canonical trigger→subscribe read-your-writes window: null under lag → apiBuilder 404 → terminal
164-
// (Electric retries only 429; RunSubscription closes on FetchError). No primary fallback → the live
165-
// run's subscription dies. This test proves (a) the store returns null under lag [the real bug], and
166-
// (b) the fix — a primary re-read (writer client → owning primary) — recovers the run.
163+
// The canonical trigger→subscribe read-your-writes window: the replica miss is the first leg (a bare
164+
// null would make apiBuilder 404 → terminal, since Electric retries only 429 and RunSubscription
165+
// closes on FetchError), so findResource re-reads the owning PRIMARY on a null. This test proves both
166+
// legs: (a) the store returns null under lag, and (b) the owning-primary re-read recovers the run.
167167
heteroRunOpsPostgresTest(
168-
"runs.$runId findRun(branded $replica) is NULL under lag — subscribeToRun gets a spurious 404 of a live run; a primary fallback recovers it",
168+
"runs.$runId findRun(branded $replica) is NULL under lag — the replica miss is the first leg; the owning-primary re-read recovers the live run",
169169
async ({ prisma14, prisma17 }) => {
170170
const { router, legacyReplica, seed, friendlyId } = await setupLaggingLegacy(
171171
prisma14,
@@ -181,8 +181,8 @@ describe("run-ops split — realtime-STREAM route read views vs. a lagging repli
181181
expect(stale).toBeNull(); // findResource null → apiBuilder returns 404 (x-should-retry:"false")
182182
expect(legacyReplica.wasHit("taskRun")).toBe(true);
183183

184-
// The run IS live on the primary — the bug is that findResource never consults it. The recommended
185-
// fix (primary fallback, like resolveRunForMutation.server.ts:65) recovers the just-triggered run.
184+
// The run IS live on the primary — findResource re-reads it there (the owning-primary fallback,
185+
// like resolveRunForMutation.server.ts:65) and recovers the just-triggered run.
186186
const recovered = (await router.findRun(where, args, prisma14 as never)) as {
187187
friendlyId: string;
188188
} | null;

‎internal-packages/run-store/src/runOpsStore.serviceBatchFamilyReadView.replicaLag.test.ts‎

Lines changed: 17 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -116,18 +116,23 @@ async function seedItems(
116116
prisma: PrismaClient,
117117
opts: { batchTaskRunId: string; status: string; count: number; idPrefix: string }
118118
) {
119-
await prisma.$executeRawUnsafe(`SET session_replication_role = replica`);
120-
for (let i = 0; i < opts.count; i++) {
121-
await prisma.$executeRawUnsafe(
122-
`INSERT INTO "BatchTaskRunItem" (id, status, "batchTaskRunId", "taskRunId", "createdAt", "updatedAt")
123-
VALUES ($1, $2::"BatchTaskRunItemStatus", $3, $4, NOW(), NOW())`,
124-
`${opts.idPrefix}_${i}`,
125-
opts.status,
126-
opts.batchTaskRunId,
127-
`run_item_${opts.idPrefix}_${i}`
128-
);
129-
}
130-
await prisma.$executeRawUnsafe(`SET session_replication_role = DEFAULT`);
119+
// `session_replication_role` is session-scoped, so `SET` and the inserts must share one connection —
120+
// separate pooled Prisma calls can land on different connections, leaving the inserts with FK triggers
121+
// still enabled. Run them in a single transaction with `SET LOCAL`, which applies for the duration of
122+
// this transaction (same connection) and auto-resets on commit.
123+
await prisma.$transaction(async (tx) => {
124+
await tx.$executeRawUnsafe(`SET LOCAL session_replication_role = replica`);
125+
for (let i = 0; i < opts.count; i++) {
126+
await tx.$executeRawUnsafe(
127+
`INSERT INTO "BatchTaskRunItem" (id, status, "batchTaskRunId", "taskRunId", "createdAt", "updatedAt")
128+
VALUES ($1, $2::"BatchTaskRunItemStatus", $3, $4, NOW(), NOW())`,
129+
`${opts.idPrefix}_${i}`,
130+
opts.status,
131+
opts.batchTaskRunId,
132+
`run_item_${opts.idPrefix}_${i}`
133+
);
134+
}
135+
});
131136
}
132137

133138
describe("batch-service family batch reads (no client) route to the owning PRIMARY under replica lag", () => {

‎internal-packages/run-store/src/runOpsStore.ttlExpireOrphanReplicaLag.test.ts‎

Lines changed: 21 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,22 @@
1-
// Repro for the run-ops split read-after-write hole on the TTL BATCH EXPIRY GUARD.
1+
// Guard for the run-ops split read-after-write hazard on the TTL BATCH EXPIRY path.
22
//
33
// TtlSystem.expireRunsBatch (internal-packages/run-engine/src/engine/systems/ttlSystem.ts:161) reads
44
// the runs it is about to expire with:
5-
// runStore.findRuns({ where: { id: { in: runIds } }, select: {...} }) // NO client
6-
// A client-less findRuns is NOT a read-your-writes signal, so RoutingRunStore routes each leg to the
7-
// owning store's REPLICA (#ownPrimary(store, undefined) === undefined -> readOnlyPrisma). Any run whose
8-
// row is not yet visible on that replica comes back absent, is bucketed `not_found`
9-
// (ttlSystem.ts:196-200), skipped, and NEVER expired. Because the TTL Lua script has ALREADY removed
10-
// the run from the Redis queue before this guard runs, a single missed read is not self-healing: the
11-
// run is permanently ORPHANED (never expired, never emits `runExpired`, never re-queued). Contrast the
12-
// single-run path expireRun (ttlSystem.ts:36) which passes `prisma` to findRun -> the OWNING store's
13-
// primary, so it never misses.
5+
// runStore.findRuns({ where: { id: { in: runIds } }, select: {...} }, this.$.prisma)
6+
// It threads the owning primary (like the single-run path expireRun at ttlSystem.ts:36 does), so it
7+
// reads the run it just wrote. The hazard the primary read avoids: a client-less findRuns is NOT a
8+
// read-your-writes signal, so RoutingRunStore would route each leg to the owning store's REPLICA
9+
// (#ownPrimary(store, undefined) === undefined -> readOnlyPrisma). A run whose row is not yet visible
10+
// on that replica would come back absent, be bucketed `not_found` (ttlSystem.ts:196-200), skipped, and
11+
// NEVER expired. Because the TTL Lua script has ALREADY removed the run from the Redis queue before
12+
// this guard runs, a single missed read would not self-heal: the run would be permanently ORPHANED
13+
// (never expired, never emits `runExpired`, never re-queued).
1414
//
15-
// This is the SUSPECT/weakest of the read-after-write class: the run row is usually written well before
16-
// TTL fires, so a lagging replica has normally caught up. The read-your-writes window only exists for
17-
// a very-short-TTL run swept moments after creation. This test forces the lag with the shared
18-
// lagging-replica primitive to determine whether the mechanism orphans the run: if the guard misses
19-
// the run on the replica and buckets it `not_found`, the run is orphaned; if it is still found, the
20-
// mechanism is not at fault.
15+
// This is the weakest of the read-after-write class: the run row is usually written well before TTL
16+
// fires, so a lagging replica has normally caught up. The read-your-writes window only exists for a
17+
// very-short-TTL run swept moments after creation. This test forces the lag with the shared
18+
// lagging-replica primitive and pins both paths: the primary-threaded read (the production path)
19+
// expires the run, while the client-less control path misses it on the replica and would orphan it.
2120
//
2221
// Deterministic via heteroRunOpsPostgresTest (real split topology, NEVER mocked): a LEGACY-resident
2322
// (cuid) run is created on the control-plane writer; the control-plane replica lags (its `taskRun`
@@ -164,12 +163,12 @@ async function guardExpireBatch(
164163
return { expired, skipped };
165164
}
166165

167-
describe("run-ops split — TTL batch expiry guard reads a lagging replica and orphans the run", () => {
166+
describe("run-ops split — TTL batch expiry guard threads the owning primary; the client-less path would orphan the run", () => {
168167
// A LEGACY-resident (cuid) run sits PENDING on the control-plane WRITER; the control-plane REPLICA
169168
// lags (its `taskRun` reads come back empty, as just after a very-short-TTL run is created). The
170169
// TTL Lua script has already dequeued the run, so this guard is its only chance to be expired.
171170
heteroRunOpsPostgresTest(
172-
"LEGACY cuid: client-less batch guard misses the run on the lagging replica -> orphaned as not_found",
171+
"LEGACY cuid: primary-threaded guard expires the run; the client-less control path misses it on the lagging replica -> orphaned as not_found",
173172
async ({ prisma14, prisma17 }) => {
174173
// Control-plane replica lags on `taskRun`: the just-created run row is not visible yet.
175174
const legacyReplica = laggingReplica(prisma14, [{ model: "taskRun", mode: "missing" }]);
@@ -210,16 +209,15 @@ describe("run-ops split — TTL batch expiry guard reads a lagging replica and o
210209
expect(viaPrimary.expired).toEqual([runId]);
211210
expect(viaPrimary.skipped).toEqual([]);
212211

213-
// The actual guard as ttlSystem.expireRunsBatch calls it: client-less findRuns → owning store's
214-
// REPLICA (the buggy batch path).
212+
// The client-less control path (pre-fix shape): findRuns with NO client → owning store's REPLICA.
215213
const guard = await guardExpireBatch(router, [runId]);
216214

217215
// Proves the misrouted read is what happened: the (stale) legacy replica was consulted.
218216
expect(legacyReplica.wasHit("taskRun")).toBe(true);
219217

220-
// The HAZARD the fix avoids: the old client-less findRuns misses the run on the lagging replica,
221-
// buckets it `not_found`, and the run is ORPHANED — never expired. Pinning that here documents
222-
// why ttlSystem.ts:161 must thread this.$.prisma (as the viaPrimary path above now does).
218+
// The HAZARD the primary read avoids: a client-less findRuns misses the run on the lagging
219+
// replica, buckets it `not_found`, and the run would be ORPHANED — never expired. Pinning it here
220+
// documents why ttlSystem.ts:161 threads this.$.prisma (the viaPrimary path above).
223221
expect(guard.expired).toEqual([]);
224222
expect(guard.skipped).toEqual([{ runId, reason: "not_found" }]);
225223
}

0 commit comments

Comments
 (0)