Skip to content

Commit 42d172a

Browse files
committed
feat(run-store,webapp): make redis-only genuinely redis-only (suppress Postgres writes, throw on miss)
1 parent 6c6c939 commit 42d172a

6 files changed

Lines changed: 345 additions & 8 deletions

‎apps/webapp/app/v3/runStore.server.ts‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import {
2121
import { env } from "~/env.server";
2222
import { singleton } from "~/utils/singleton";
2323
import { decorateWithSnapshotStore } from "./snapshotStoreInstance.server";
24+
import { snapshotStoreModeResolver } from "./snapshotStoreMode.server";
2425
import {
2526
resilienceForClient,
2627
type TransactionResilienceConfig,
@@ -68,13 +69,19 @@ type BuildRunStoreDeps = {
6869
* are born) and a LEGACY store (draining) by run-id residency (id shape). There is no cuid
6970
* migration, so a LEGACY-classified id is always LEGACY-resident.
7071
*/
72+
// A run's org is redis-only exactly when the same resolver the snapshot decorator uses says so, so
73+
// the Redis mirror and the Postgres suppression always agree on the effective mode for that run.
74+
const suppressPgAtRedisOnly = (organizationId?: string) =>
75+
snapshotStoreModeResolver.resolve(organizationId) !== "redis-only";
76+
7177
export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
7278
if (!deps.splitEnabled) {
7379
return new PostgresRunStore({
7480
prisma: deps.singleWriter,
7581
readOnlyPrisma: deps.singleReplica,
7682
maxWait: deps.singleResilience?.maxWait,
7783
transactionStartRetry: deps.singleResilience?.startRetry,
84+
snapshotWrites: suppressPgAtRedisOnly,
7885
});
7986
}
8087

@@ -90,12 +97,14 @@ export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
9097
schemaVariant: "dedicated",
9198
maxWait: deps.newResilience?.maxWait,
9299
transactionStartRetry: deps.newResilience?.startRetry,
100+
snapshotWrites: suppressPgAtRedisOnly,
93101
});
94102
const legacyStore = new PostgresRunStore({
95103
prisma: deps.legacyWriter,
96104
readOnlyPrisma: deps.legacyReplica,
97105
maxWait: deps.legacyResilience?.maxWait,
98106
transactionStartRetry: deps.legacyResilience?.startRetry,
107+
snapshotWrites: suppressPgAtRedisOnly,
99108
});
100109

101110
// Gen-2 shards: one dedicated store per descriptor, handed to the N-way router. An aliased shard
@@ -109,6 +118,7 @@ export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
109118
schemaVariant: "dedicated" as const,
110119
maxWait: shard.resilience?.maxWait,
111120
transactionStartRetry: shard.resilience?.startRetry,
121+
snapshotWrites: suppressPgAtRedisOnly,
112122
}),
113123
aliasOf: shard.aliasOf,
114124
}));

‎internal-packages/run-store/src/PostgresRunStore.snapshotWrites.test.ts‎

Lines changed: 91 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import { PostgresRunStore } from "./PostgresRunStore.js";
88
import {
99
buildCreateRunData,
1010
seedSnapshotEnvironment,
11+
seedSnapshotWaitpoints,
1112
seedSnapshotWorker,
1213
setupSnapshotIdFixture,
1314
} from "./testFixtures/snapshotIdFixture.js";
@@ -309,4 +310,94 @@ describe("PostgresRunStore snapshotWrites flag", () => {
309310
})
310311
).rejects.toThrow(/snapshotWrites is off/);
311312
});
313+
314+
postgresTest(
315+
"a per-org predicate suppresses the snapshot only for the redis-only org",
316+
async ({ prisma }) => {
317+
const ro = await setupSnapshotIdFixture(prisma);
318+
const keep = await setupSnapshotIdFixture(prisma);
319+
const store = new PostgresRunStore({
320+
prisma,
321+
readOnlyPrisma: prisma,
322+
snapshotWrites: (org) => org !== ro.env.organizationId,
323+
});
324+
325+
for (const fx of [ro, keep]) {
326+
await store.completeAttemptSuccess(
327+
fx.run.id,
328+
{
329+
completedAt: new Date(),
330+
outputType: "application/json",
331+
usageDurationMs: 1,
332+
costInCents: 0,
333+
snapshot: {
334+
executionStatus: "FINISHED",
335+
description: "Run completed",
336+
runStatus: "COMPLETED_SUCCESSFULLY",
337+
attemptNumber: 1,
338+
environmentId: fx.env.id,
339+
environmentType: fx.env.type,
340+
projectId: fx.env.projectId,
341+
organizationId: fx.env.organizationId,
342+
},
343+
},
344+
{ select: { id: true } }
345+
);
346+
}
347+
348+
expect(await prisma.taskRunExecutionSnapshot.count({ where: { runId: ro.run.id } })).toBe(0);
349+
expect(await prisma.taskRunExecutionSnapshot.count({ where: { runId: keep.run.id } })).toBe(
350+
1
351+
);
352+
}
353+
);
354+
355+
postgresTest(
356+
"a per-org predicate suppresses the completed-waitpoint join rows for the redis-only org",
357+
async ({ prisma }) => {
358+
const ro = await setupSnapshotIdFixture(prisma);
359+
const keep = await setupSnapshotIdFixture(prisma);
360+
const store = new PostgresRunStore({
361+
prisma,
362+
readOnlyPrisma: prisma,
363+
snapshotWrites: (org) => org !== ro.env.organizationId,
364+
});
365+
366+
for (const fx of [ro, keep]) {
367+
const { workerId, taskId } = await seedSnapshotWorker(prisma, fx.env);
368+
const waitpointIds = await seedSnapshotWaitpoints(prisma, fx.env, 2);
369+
const snapshotId = generateInternalId();
370+
await store.lockRunToWorker(fx.run.id, {
371+
lockedAt: new Date(),
372+
lockedById: taskId,
373+
lockedToVersionId: workerId,
374+
lockedQueueId: undefined,
375+
startedAt: new Date(),
376+
baseCostInCents: 0,
377+
machinePreset: "small-1x",
378+
taskVersion: "1.0.0",
379+
snapshot: {
380+
id: snapshotId,
381+
previousSnapshotId: generateInternalId(),
382+
attemptNumber: 1,
383+
environmentId: fx.env.id,
384+
environmentType: fx.env.type,
385+
projectId: fx.env.projectId,
386+
organizationId: fx.env.organizationId,
387+
completedWaitpointIds: waitpointIds,
388+
completedWaitpointOrder: waitpointIds,
389+
},
390+
});
391+
}
392+
393+
// The redis-only org has no snapshot row, so no join rows link to it either.
394+
expect(await prisma.taskRunExecutionSnapshot.count({ where: { runId: ro.run.id } })).toBe(0);
395+
396+
const keepSnapshot = await prisma.taskRunExecutionSnapshot.findFirstOrThrow({
397+
where: { runId: keep.run.id },
398+
include: { completedWaitpoints: true },
399+
});
400+
expect(keepSnapshot.completedWaitpoints).toHaveLength(2);
401+
}
402+
);
312403
});

‎internal-packages/run-store/src/PostgresRunStore.ts‎

Lines changed: 14 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -123,9 +123,10 @@ export type PostgresRunStoreOptions = {
123123
* When false the store writes no execution-snapshot rows: every nested `executionSnapshots.create`
124124
* is omitted and `createExecutionSnapshot` echoes its input instead of inserting. Only the
125125
* redis-only dial position sets this, once the Redis store is the sole snapshot writer.
126-
* Defaults to true, so the store behaves exactly as it always has.
126+
* A predicate form decides per run from its organisation id, so a per-org redis-only override
127+
* suppresses only that org's snapshots. Defaults to true, so the store behaves exactly as it always has.
127128
*/
128-
snapshotWrites?: boolean;
129+
snapshotWrites?: boolean | ((organizationId?: string) => boolean);
129130
};
130131

131132
// A caller sub-select for a relation: `{ select?, include? }` or `true` for a bare `key: true`.
@@ -646,7 +647,7 @@ export class PostgresRunStore implements RunStore {
646647
private readonly prisma: RunOpsCapableClient;
647648
private readonly readOnlyPrisma: RunOpsCapableClient;
648649
private readonly schemaVariant: RunStoreSchemaVariant;
649-
private readonly snapshotWrites: boolean;
650+
private readonly snapshotWrites: boolean | ((organizationId?: string) => boolean);
650651
private readonly maxWait?: number;
651652
private readonly transactionStartRetry?: TransactionStartRetryConfig;
652653

@@ -662,6 +663,13 @@ export class PostgresRunStore implements RunStore {
662663
this.snapshotWrites = options.snapshotWrites ?? true;
663664
}
664665

666+
// Resolves the write flag for one run, from its organisation id when the option is a predicate.
667+
#writesSnapshot(organizationId?: string): boolean {
668+
return typeof this.snapshotWrites === "function"
669+
? this.snapshotWrites(organizationId)
670+
: this.snapshotWrites;
671+
}
672+
665673
/**
666674
* Wraps a nested snapshot create so a single flag removes it everywhere. Prisma treats an absent
667675
* key and `undefined` alike, so spreading an empty object drops the nested write entirely rather
@@ -674,7 +682,7 @@ export class PostgresRunStore implements RunStore {
674682
};
675683
}
676684
| Record<string, never> {
677-
return this.snapshotWrites ? { executionSnapshots: { create } } : {};
685+
return this.#writesSnapshot(create.organizationId) ? { executionSnapshots: { create } } : {};
678686
}
679687

680688
// The writer handle in read-client form, so the routing layer can honor a caller-passed client
@@ -1316,7 +1324,7 @@ export class PostgresRunStore implements RunStore {
13161324

13171325
// The join rows link to the snapshot row above. With snapshot writes off there is no such row,
13181326
// so inserting them would leave dangling links for a snapshot that only the Redis store holds.
1319-
if (this.snapshotWrites) {
1327+
if (this.#writesSnapshot(data.snapshot.organizationId)) {
13201328
if (dedicated) {
13211329
await this.#connectCompletedWaitpoints(
13221330
prisma,
@@ -2036,7 +2044,7 @@ export class PostgresRunStore implements RunStore {
20362044

20372045
// Redis-only: no row is written and the decorator owns the document. Echo the input in the shape
20382046
// the caller expects, so every caller of this method keeps working while Postgres holds nothing.
2039-
if (!this.snapshotWrites) {
2047+
if (!this.#writesSnapshot(input.organizationId)) {
20402048
if (!id) {
20412049
throw new Error(
20422050
"PostgresRunStore.createExecutionSnapshot: snapshotWrites is off, so the caller must supply the snapshot id"

‎internal-packages/run-store/src/taskRunExecutionSnapshotStore.redisOnlyFallback.test.ts‎

Lines changed: 108 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -132,3 +132,111 @@ describe("redis-only fallback gate is org-aware", () => {
132132
expect(h.delegateTouched).not.toContain("findSnapshotCompletedWaitpointIds");
133133
});
134134
});
135+
136+
// The other half of the gate: a Redis MISS (not an error). At redis-only Postgres holds nothing,
137+
// so a miss must THROW rather than delegate into an empty Postgres. Below redis-only a miss still
138+
// falls back, because Postgres legitimately holds the pre-cutover data.
139+
function missHarness(opts: {
140+
globalMode: SnapshotStoreMode;
141+
readModeFor?: (runId: string, environmentId?: string) => SnapshotStoreMode | undefined;
142+
anyOrgRedisOnly?: () => boolean;
143+
anyOrgReadEnabled?: () => boolean;
144+
}) {
145+
// Every Redis read reports a MISS: null for the single/by-id reads, a miss window, an absent
146+
// waitpoint set.
147+
const redis = new Proxy({} as RedisSnapshotStore, {
148+
get: (_t, prop) => {
149+
if (prop === "getSinceCreatedAt") return () => Promise.resolve({ kind: "miss" });
150+
if (prop === "getSnapshotWaitpointIds")
151+
return () => Promise.resolve({ present: false, distinctIds: [], order: [] });
152+
return () => Promise.resolve(null);
153+
},
154+
});
155+
156+
const delegateTouched: string[] = [];
157+
const delegate = new Proxy({} as Record<string, unknown>, {
158+
get: (_t, prop) => {
159+
return (...__: unknown[]) => {
160+
delegateTouched.push(String(prop));
161+
return Promise.resolve({ id: "pg_fallback" });
162+
};
163+
},
164+
}) as unknown as RunStore;
165+
166+
const modeResolver: SnapshotStoreModeResolver = {
167+
resolve: () => opts.globalMode,
168+
...(opts.readModeFor && {
169+
readModeFor: opts.readModeFor as SnapshotStoreModeResolver["readModeFor"],
170+
}),
171+
...(opts.anyOrgRedisOnly && { anyOrgRedisOnly: opts.anyOrgRedisOnly }),
172+
...(opts.anyOrgReadEnabled && { anyOrgReadEnabled: opts.anyOrgReadEnabled }),
173+
};
174+
175+
const decorated = new TaskRunExecutionSnapshotStore(delegate, {
176+
store: redis,
177+
mode: opts.globalMode,
178+
modeResolver,
179+
readPercent: 100,
180+
});
181+
182+
return { decorated, delegateTouched };
183+
}
184+
185+
describe("redis-only throws on a miss instead of serving empty Postgres", () => {
186+
it("findLatestExecutionSnapshot throws at global redis-only, never delegating", async () => {
187+
const h = missHarness({ globalMode: "redis-only" });
188+
189+
await expect(
190+
h.decorated.findLatestExecutionSnapshot("run_x", undefined, "env_x")
191+
).rejects.toThrow(/redis-only/);
192+
expect(h.delegateTouched).not.toContain("findLatestExecutionSnapshot");
193+
});
194+
195+
it("findLatestExecutionSnapshot falls back at redis-read (Postgres holds pre-cutover data)", async () => {
196+
const h = missHarness({ globalMode: "redis-read" });
197+
198+
const result = await h.decorated.findLatestExecutionSnapshot("run_x", undefined, "env_x");
199+
expect(result).toEqual({ id: "pg_fallback" });
200+
expect(h.delegateTouched).toContain("findLatestExecutionSnapshot");
201+
});
202+
203+
it("throws for a run whose org resolves to redis-only, falls back for a redis-read org", async () => {
204+
const h = missHarness({
205+
globalMode: "redis-read",
206+
anyOrgReadEnabled: () => true,
207+
readModeFor: (runId) => (runId === "run_ro" ? "redis-only" : "redis-read"),
208+
});
209+
210+
await expect(
211+
h.decorated.findLatestExecutionSnapshot("run_ro", undefined, "env_a")
212+
).rejects.toThrow(/redis-only/);
213+
expect(h.delegateTouched).not.toContain("findLatestExecutionSnapshot");
214+
215+
const result = await h.decorated.findLatestExecutionSnapshot("run_rr", undefined, "env_b");
216+
expect(result).toEqual({ id: "pg_fallback" });
217+
expect(h.delegateTouched).toContain("findLatestExecutionSnapshot");
218+
});
219+
220+
it("findManyExecutionSnapshots throws on a miss window at redis-only", async () => {
221+
const h = missHarness({ globalMode: "redis-only" });
222+
223+
await expect(
224+
h.decorated.findManyExecutionSnapshots({
225+
where: { runId: "run_x", isValid: true, createdAt: { gt: new Date(0) } },
226+
include: { checkpoint: true },
227+
orderBy: { createdAt: "desc" },
228+
take: 10,
229+
} as never)
230+
).rejects.toThrow(/redis-only/);
231+
expect(h.delegateTouched).not.toContain("findManyExecutionSnapshots");
232+
});
233+
234+
it("findSnapshotCompletedWaitpointIds throws on an absent set at redis-only", async () => {
235+
const h = missHarness({ globalMode: "redis-only" });
236+
237+
await expect(
238+
h.decorated.findSnapshotCompletedWaitpointIds("snap_1", undefined, "run_x")
239+
).rejects.toThrow(/redis-only/);
240+
expect(h.delegateTouched).not.toContain("findSnapshotCompletedWaitpointIds");
241+
});
242+
});

0 commit comments

Comments
 (0)