Skip to content

Commit a410449

Browse files
committed
fix(run-store,webapp): fix a run's snapshot store at birth, not from the live dial
A run's storage residency was re-derived from the live per-org dial on every write and read, but a run's store is fixed when it is born. As the dial drifted, long-lived runs, runs born in the enrollment window, and runs read while an org sat at redis-only were routed to a store that did not hold them, stranding or losing their execution state. Residency is now a per-run decision recorded at birth and remembered per process (a bounded LRU on the shared Redis store, monotonic toward redis-only, learned for free from births, append replies, and a one-off keyspace probe, never a database read). A redis-only-born run suppresses its Postgres snapshot rows and refuses a Postgres fallback for life; every other run, including all runs born before the cutover, writes and reads Postgres for life whatever the dial later does. A halt forces Postgres writes back on so a redis-only run never writes its transition nowhere. Reads and the fallback gate key on the run's regime, not the global dial. Removes the per-read run to org control-plane database read that could deadlock dequeues for split-database runs once any org reached redis-only, along with the live-dial Postgres suppression predicate and the birth-time dial wait. Covered by a transition matrix over birth dial by later dial asserting where each write lands and where each read is served.
1 parent 3230a64 commit a410449

25 files changed

Lines changed: 1152 additions & 864 deletions

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

Lines changed: 5 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,7 @@ import {
2020
} from "~/db.server";
2121
import { env } from "~/env.server";
2222
import { singleton } from "~/utils/singleton";
23-
import { isSnapshotStoreConfigured } from "./snapshotStoreConfigured.server";
2423
import { decorateWithSnapshotStore } from "./snapshotStoreInstance.server";
25-
import { snapshotStoreModeResolver } from "./snapshotStoreMode.server";
2624
import {
2725
resilienceForClient,
2826
type TransactionResilienceConfig,
@@ -56,10 +54,6 @@ type BuildRunStoreDeps = {
5654
singleResilience?: TransactionResilienceConfig;
5755
newResilience?: TransactionResilienceConfig;
5856
legacyResilience?: TransactionResilienceConfig;
59-
/** The redis-only PG-suppression predicate, or undefined for a plain passthrough. Passed in (never
60-
* read from env here) so this builder stays pure; the caller wires it only when the store is
61-
* configured, so an unconfigured deploy writes every snapshot row with no per-write resolution. */
62-
snapshotWrites?: (organizationId?: string) => boolean;
6357
};
6458

6559
/**
@@ -74,24 +68,19 @@ type BuildRunStoreDeps = {
7468
* Split ON: returns a RoutingRunStore that selects between a NEW store (where new runs
7569
* are born) and a LEGACY store (draining) by run-id residency (id shape). There is no cuid
7670
* migration, so a LEGACY-classified id is always LEGACY-resident.
71+
*
72+
* Postgres snapshot-row suppression is NOT wired here. It is a per-run decision the snapshot
73+
* decorator threads onto each snapshot input (`writeSnapshotRow`), fixed at the run's birth, so the
74+
* store never re-derives residency from the live org dial. An unwired/unconfigured deploy leaves the
75+
* field unset and every snapshot row is written, byte-identical to before.
7776
*/
78-
// A run's org is redis-only exactly when the same resolver the snapshot decorator uses says so, so
79-
// the Redis mirror and the Postgres suppression always agree on the effective mode for that run.
80-
const suppressPgAtRedisOnly = (organizationId?: string) =>
81-
snapshotStoreModeResolver.resolve(organizationId) !== "redis-only";
82-
83-
// Wired into the store only when the snapshot store is configured. Unconfigured, this stays undefined
84-
// and PostgresRunStore writes every snapshot row with no per-write mode resolution (the inert state).
85-
const snapshotWritesPredicate = isSnapshotStoreConfigured() ? suppressPgAtRedisOnly : undefined;
86-
8777
export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
8878
if (!deps.splitEnabled) {
8979
return new PostgresRunStore({
9080
prisma: deps.singleWriter,
9181
readOnlyPrisma: deps.singleReplica,
9282
maxWait: deps.singleResilience?.maxWait,
9383
transactionStartRetry: deps.singleResilience?.startRetry,
94-
snapshotWrites: deps.snapshotWrites,
9584
});
9685
}
9786

@@ -107,14 +96,12 @@ export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
10796
schemaVariant: "dedicated",
10897
maxWait: deps.newResilience?.maxWait,
10998
transactionStartRetry: deps.newResilience?.startRetry,
110-
snapshotWrites: deps.snapshotWrites,
11199
});
112100
const legacyStore = new PostgresRunStore({
113101
prisma: deps.legacyWriter,
114102
readOnlyPrisma: deps.legacyReplica,
115103
maxWait: deps.legacyResilience?.maxWait,
116104
transactionStartRetry: deps.legacyResilience?.startRetry,
117-
snapshotWrites: deps.snapshotWrites,
118105
});
119106

120107
// Gen-2 shards: one dedicated store per descriptor, handed to the N-way router. An aliased shard
@@ -128,7 +115,6 @@ export function buildRunStore(deps: BuildRunStoreDeps): RunStore {
128115
schemaVariant: "dedicated" as const,
129116
maxWait: shard.resilience?.maxWait,
130117
transactionStartRetry: shard.resilience?.startRetry,
131-
snapshotWrites: deps.snapshotWrites,
132118
}),
133119
aliasOf: shard.aliasOf,
134120
}));
@@ -227,7 +213,6 @@ export const runStoreWithoutSnapshotDecorator: RunStore = singleton("RunStore.un
227213
singleWriter: prisma,
228214
singleReplica: $replica,
229215
singleResilience: resilienceForClient(prisma),
230-
snapshotWrites: snapshotWritesPredicate,
231216
});
232217
}
233218
const { shardHandles, ...storeHandles } = handles;
@@ -246,7 +231,6 @@ export const runStoreWithoutSnapshotDecorator: RunStore = singleton("RunStore.un
246231
singleResilience: resilienceForClient(prisma),
247232
newResilience: resilienceForClient(handles.newWriter),
248233
legacyResilience: resilienceForClient(handles.legacyWriter),
249-
snapshotWrites: snapshotWritesPredicate,
250234
});
251235
});
252236

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

Lines changed: 1 addition & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -1,31 +1,9 @@
11
import { LRUCache } from "lru-cache";
2-
import { prisma } from "~/db.server";
32
import { env } from "~/env.server";
43
import { singleton } from "~/utils/singleton";
54

6-
/**
7-
* How long an authoritative run→org read will wait before giving up. Shares the shape of the
8-
* org-mode source's warm timeout, but here the deadline REJECTS: this feeds the redis-only fallback
9-
* gate, which must fail closed rather than answer with a wrong org.
10-
*/
11-
const AUTHORITATIVE_TIMEOUT_MS = 500;
12-
135
const DEFAULT_CACHE_MAX = 50_000;
146

15-
/**
16-
* What resolution needs from Prisma, narrowed so a test injects a hand-written fake rather than
17-
* mocking the client. A run's organisation is reached through its environment: TaskRun.organizationId
18-
* is nullable on historical rows, while RuntimeEnvironment.organizationId is not.
19-
*/
20-
type RunOrgClient = {
21-
taskRun: {
22-
findFirst(args: {
23-
where: { id: string };
24-
select: { runtimeEnvironment: { select: { organizationId: true } } };
25-
}): Promise<{ runtimeEnvironment: { organizationId: string } } | null>;
26-
};
27-
};
28-
297
export type SnapshotRunOrgSource = {
308
/**
319
* A PURE cache get: the cached org id, or undefined on a miss. Never queries, never blocks, never
@@ -41,39 +19,14 @@ export type SnapshotRunOrgSource = {
4119
* changes), no TTL, no invalidation. Fire-and-forget: never queries, never throws.
4220
*/
4321
prime(runId: string, organizationId: string): void;
44-
/**
45-
* Awaits a bounded primary read, caches, and returns the org id. Throws on timeout, client
46-
* failure, or a run with no organisation, so a caller can fail closed. This is the redis-only
47-
* fallback gate's last leg, reached only when the sync cache is cold AND some org is redis-only.
48-
*/
49-
resolveAuthoritative(runId: string): Promise<string>;
5022
};
5123

52-
export function createSnapshotRunOrgSource(clients?: {
53-
primary: RunOrgClient;
54-
}): SnapshotRunOrgSource {
55-
const primaryClient = (clients?.primary ?? prisma) as RunOrgClient;
24+
export function createSnapshotRunOrgSource(): SnapshotRunOrgSource {
5625
// No ttl: run→org is immutable, so a cached mapping never goes stale.
5726
const cache = new LRUCache<string, string>({
5827
max: env.RUN_ENGINE_SNAPSHOT_STORE_RUN_ORG_CACHE_MAX ?? DEFAULT_CACHE_MAX,
5928
});
6029

61-
async function read(runId: string): Promise<string> {
62-
return primaryClient.taskRun
63-
.findFirst({
64-
where: { id: runId },
65-
select: { runtimeEnvironment: { select: { organizationId: true } } },
66-
})
67-
.then((row) => {
68-
const organizationId = row?.runtimeEnvironment?.organizationId;
69-
if (!organizationId) {
70-
throw new Error(`snapshotRunOrg: no organization for run ${runId}`);
71-
}
72-
cache.set(runId, organizationId);
73-
return organizationId;
74-
});
75-
}
76-
7730
return {
7831
resolve(runId) {
7932
return cache.get(runId);
@@ -83,29 +36,6 @@ export function createSnapshotRunOrgSource(clients?: {
8336
// active run hot in the LRU, which is exactly the run whose mapping the read path will want.
8437
cache.set(runId, organizationId);
8538
},
86-
async resolveAuthoritative(runId) {
87-
const cached = cache.get(runId);
88-
if (cached !== undefined) {
89-
return cached;
90-
}
91-
92-
let timer: NodeJS.Timeout | undefined;
93-
const deadline = new Promise<never>((_, reject) => {
94-
timer = setTimeout(
95-
() =>
96-
reject(
97-
new Error(`snapshotRunOrg: run→org read exceeded ${AUTHORITATIVE_TIMEOUT_MS}ms`)
98-
),
99-
AUTHORITATIVE_TIMEOUT_MS
100-
);
101-
});
102-
103-
try {
104-
return await Promise.race([read(runId), deadline]);
105-
} finally {
106-
if (timer) clearTimeout(timer);
107-
}
108-
},
10939
};
11040
}
11141

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

Lines changed: 2 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -90,23 +90,19 @@ export function isSnapshotStoreCohortMember(organizationId: string): boolean {
9090
return snapshotStoreIsCohortMember(loadedOrgDials(), organizationId);
9191
}
9292

93-
/** What resolution needs from the org source. Read-only and synchronous; warm is a birth-only hook. */
93+
/** What resolution needs from the org source. Read-only and synchronous. */
9494
type ResolverOrgSource = {
9595
/** The org's dial, or NO_OVERRIDE when absent so `resolveMode` falls back to the global dial. */
9696
get(organizationId: string): CachedOrgMode | undefined;
97-
/** No-op with the map source: the dial arrives via the poll, so there is nothing to warm off-path. */
97+
/** No-op with the map source: the dial arrives via the poll, so there is nothing to refresh. */
9898
refresh(organizationId: string): void;
99-
/** No-op with the map source: births read the map synchronously, so there is nothing to await. */
100-
warm?(organizationId: string): Promise<void>;
10199
};
102100

103101
/** Resolves a run to its organisation. Cache-only and synchronous, undefined on a miss. */
104102
type ResolverRunOrgSource = {
105103
resolve(runId: string): string | undefined;
106104
/** Records an immutable run→org mapping learned off a mirrored write or a Redis read hit. */
107105
prime?(runId: string, organizationId: string): void;
108-
/** Bounded authoritative read, throws on failure/timeout, for the redis-only fallback gate. */
109-
resolveAuthoritative?(runId: string): Promise<string>;
110106
};
111107

112108
/** The census read accessors the resolver delegates to. Both synchronous and no-query. */
@@ -169,11 +165,6 @@ export function buildSnapshotStoreModeResolver(deps: {
169165
// probing rather than suppressing a resident run.
170166
orgDefinitelyNeverEnabled: (organizationId: string): boolean =>
171167
deps.orgDefinitelyNeverEnabled?.(organizationId) ?? false,
172-
// Awaited at birth sites only. With the map source this resolves immediately: the dial is read
173-
// synchronously from the polled map, so there is no per-org read to await.
174-
warm: async (organizationId: string): Promise<void> => {
175-
await deps.orgMode.warm?.(organizationId);
176-
},
177168
resolve: (organizationId?: string): DialMode => resolveMode(organizationId),
178169
// The org-scoped read position. Resolve run→org synchronously; on a miss return undefined so
179170
// the decorator falls back to the global mode, which is safe during soak.
@@ -184,16 +175,6 @@ export function buildSnapshotStoreModeResolver(deps: {
184175
}
185176
return resolveMode(organizationId);
186177
},
187-
// Authoritative counterpart, used only when the sync read is unresolved and some org is
188-
// redis-only. Resolves run→org from the primary (bounded, throws on failure), then answers with
189-
// the org's mode straight from the map. A throw propagates so the decorator fails closed.
190-
readModeForAuthoritative: async (runId: string): Promise<DialMode | undefined> => {
191-
if (!deps.runOrg?.resolveAuthoritative) {
192-
return undefined;
193-
}
194-
const organizationId = await deps.runOrg.resolveAuthoritative(runId);
195-
return resolveMode(organizationId);
196-
},
197178
anyOrgReadEnabled: (): boolean => deps.census?.anyOrgReadEnabled() ?? false,
198179
anyOrgRedisOnly: (): boolean => deps.census?.anyOrgRedisOnly() ?? false,
199180
// The decorator hands back a run→org mapping it learned on a mirrored write or a Redis read hit.
@@ -240,12 +221,10 @@ export const snapshotStoreModeResolver: SnapshotStoreModeResolver = buildSnapsho
240221
// global dial. A cold registry yields an empty map, which reads as NO_OVERRIDE for every org.
241222
get: (organizationId) => orgDials()[organizationId] ?? NO_OVERRIDE,
242223
refresh: () => {},
243-
warm: async () => {},
244224
},
245225
runOrg: {
246226
resolve: (runId) => snapshotRunOrgSource().resolve(runId),
247227
prime: (runId, organizationId) => snapshotRunOrgSource().prime(runId, organizationId),
248-
resolveAuthoritative: (runId) => snapshotRunOrgSource().resolveAuthoritative(runId),
249228
},
250229
census: {
251230
anyOrgReadEnabled: () => snapshotStoreAnyOrgReadEnabled(loadedOrgDials()),
Lines changed: 8 additions & 83 deletions
Original file line numberDiff line numberDiff line change
@@ -1,112 +1,37 @@
11
import { describe, expect, it } from "vitest";
22
import { createSnapshotRunOrgSource } from "~/v3/snapshotRunOrg.server";
33

4-
type Row = { runtimeEnvironment: { organizationId: string } } | null;
5-
6-
/** A hand-written fake standing in for the Prisma client, so no mocking is needed. */
7-
function fakeClient(opts: {
8-
mapping?: Record<string, string>;
9-
reject?: boolean;
10-
delayMs?: number;
11-
}) {
12-
let calls = 0;
13-
return {
14-
get calls() {
15-
return calls;
16-
},
17-
taskRun: {
18-
async findFirst(args: { where: { id: string } }): Promise<Row> {
19-
calls++;
20-
if (opts.delayMs) {
21-
await new Promise((resolve) => setTimeout(resolve, opts.delayMs));
22-
}
23-
if (opts.reject) {
24-
throw new Error("db unreachable");
25-
}
26-
const organizationId = opts.mapping?.[args.where.id];
27-
return organizationId ? { runtimeEnvironment: { organizationId } } : null;
28-
},
29-
},
30-
};
31-
}
32-
334
const tick = () => new Promise((resolve) => setTimeout(resolve, 10));
345

356
describe("snapshot run→org source", () => {
36-
it("resolve is a pure cache get: a cold miss is undefined and never queries", async () => {
37-
const primary = fakeClient({ mapping: { run_a: "org_a" } });
38-
const source = createSnapshotRunOrgSource({ primary });
7+
it("resolve is a pure cache get: a cold miss is undefined and never blocks", async () => {
8+
const source = createSnapshotRunOrgSource();
399

4010
expect(source.resolve("run_a")).toBeUndefined();
4111

4212
await tick();
4313

44-
// No off-path populate exists anymore, so a miss stays a miss and the DB is never touched.
14+
// There is no off-path populate and no DB read, so a miss stays a miss for life.
4515
expect(source.resolve("run_a")).toBeUndefined();
46-
expect(primary.calls).toBe(0);
4716
});
4817

49-
it("prime makes a later resolve a pure hit, with no query", async () => {
50-
const primary = fakeClient({ mapping: { run_a: "org_a" } });
51-
const source = createSnapshotRunOrgSource({ primary });
18+
it("prime makes a later resolve a pure hit", async () => {
19+
const source = createSnapshotRunOrgSource();
5220

5321
source.prime("run_a", "org_a");
5422

5523
expect(source.resolve("run_a")).toBe("org_a");
5624
await tick();
57-
expect(primary.calls).toBe(0);
25+
expect(source.resolve("run_a")).toBe("org_a");
5826
});
5927

60-
it("prime is idempotent and never queries, however many times it is called", async () => {
61-
const primary = fakeClient({ mapping: { run_a: "org_a" } });
62-
const source = createSnapshotRunOrgSource({ primary });
28+
it("prime is idempotent, however many times it is called", () => {
29+
const source = createSnapshotRunOrgSource();
6330

6431
source.prime("run_a", "org_a");
6532
source.prime("run_a", "org_a");
6633
source.prime("run_a", "org_a");
6734

6835
expect(source.resolve("run_a")).toBe("org_a");
69-
expect(primary.calls).toBe(0);
70-
});
71-
72-
it("resolveAuthoritative returns the org id on success and caches it", async () => {
73-
const primary = fakeClient({ mapping: { run_a: "org_a" } });
74-
const source = createSnapshotRunOrgSource({ primary });
75-
76-
await expect(source.resolveAuthoritative("run_a")).resolves.toBe("org_a");
77-
expect(source.resolve("run_a")).toBe("org_a");
78-
});
79-
80-
it("resolveAuthoritative serves a primed mapping without querying", async () => {
81-
const primary = fakeClient({ mapping: { run_a: "org_a" } });
82-
const source = createSnapshotRunOrgSource({ primary });
83-
84-
source.prime("run_a", "org_a");
85-
86-
await expect(source.resolveAuthoritative("run_a")).resolves.toBe("org_a");
87-
expect(primary.calls).toBe(0);
88-
});
89-
90-
it("resolveAuthoritative throws when the run has no organization", async () => {
91-
const primary = fakeClient({ mapping: {} });
92-
const source = createSnapshotRunOrgSource({ primary });
93-
94-
await expect(source.resolveAuthoritative("run_missing")).rejects.toThrow();
95-
});
96-
97-
it("resolveAuthoritative throws when the client rejects", async () => {
98-
const primary = fakeClient({ reject: true });
99-
const source = createSnapshotRunOrgSource({ primary });
100-
101-
await expect(source.resolveAuthoritative("run_a")).rejects.toThrow();
102-
});
103-
104-
it("resolveAuthoritative throws when the read exceeds the deadline", async () => {
105-
// Just over the 500 ms deadline: enough to trip it, without leaving a long timer running past
106-
// the rejection and holding the vitest worker open for the difference.
107-
const primary = fakeClient({ mapping: { run_a: "org_a" }, delayMs: 600 });
108-
const source = createSnapshotRunOrgSource({ primary });
109-
110-
await expect(source.resolveAuthoritative("run_a")).rejects.toThrow(/deadline|exceed/i);
11136
});
11237
});

0 commit comments

Comments
 (0)