Skip to content

Commit ec4bf3f

Browse files
committed
feat(webapp,run-store): activate sound per-org transition probe suppression
1 parent da6f3ab commit ec4bf3f

6 files changed

Lines changed: 146 additions & 182 deletions

‎apps/webapp/app/routes/admin.api.v2.orgs.$organizationId.feature-flags.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.
99
import { globalFlagsRegistry } from "~/v3/globalFlagsRegistry.server";
1010
import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
1111
import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server";
12+
import { snapshotStoreOrgCensus } from "~/v3/snapshotStoreOrgCensus.server";
1213
import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
1314
import { flags as getGlobalFlags } from "~/v3/featureFlags.server";
1415
import {
@@ -220,6 +221,10 @@ export async function action({ request, params }: ActionFunctionArgs) {
220221
// Org feature flags are embedded in every env of the org; drop all its cached env rows.
221222
controlPlaneResolver.invalidateOrganization(organizationId);
222223
invalidateSnapshotStoreOrgMode(organizationId);
224+
// Refresh the census in THIS process at once, so a just-enabled org stops reading as
225+
// definitely-never-enabled here immediately. Other pods lag at most the reload interval. This only
226+
// shrinks the enabling-edge transient at the transition skip; it does not make it airtight.
227+
void snapshotStoreOrgCensus.refresh();
223228

224229
return json({ success: true });
225230
}

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

Lines changed: 29 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -37,27 +37,12 @@ export function cachedOrgModeFor(raw: unknown): CachedOrgMode {
3737
return parsed.success ? parsed.data : NO_OVERRIDE;
3838
}
3939

40-
/**
41-
* The per-org residency latch, cached alongside the dial from the same row. `undefined` means the
42-
* key was absent or unreadable, kept distinct from an explicit `false` so the resolver's fail-safe
43-
* treats absence as "still probe" rather than "never enabled".
44-
*/
45-
export function cachedOrgEverEnabledFor(raw: unknown): boolean | undefined {
46-
const parsed = FeatureFlagCatalog[FEATURE_FLAG.snapshotStoreOrgEverEnabled].safeParse(raw);
47-
return parsed.success ? parsed.data : undefined;
48-
}
49-
50-
/** One organisation's cached blob values, both read from the same row. */
51-
type CachedOrg = { mode: CachedOrgMode; everEnabled: boolean | undefined };
40+
/** One organisation's cached dial value. */
41+
type CachedOrg = { mode: CachedOrgMode };
5242

5343
type OrgModeSource = {
5444
/** Cache-only: returns undefined on a true miss rather than querying. */
5545
get(organizationId: string): CachedOrgMode | undefined;
56-
/**
57-
* Cache-only read of the per-org residency latch. undefined on a miss OR a cached-but-absent key
58-
* (both mean "still probe"); a definite boolean only when the key was explicitly set.
59-
*/
60-
orgEverEnabled(organizationId: string): boolean | undefined;
6146
/** Fire-and-forget, de-duplicated per organisation, never throws. */
6247
refresh(organizationId: string): void;
6348
/** Re-reads from the primary after a write, so replica lag cannot re-cache the old value. */
@@ -86,10 +71,17 @@ type ResolverCensus = {
8671

8772
export function buildSnapshotStoreModeResolver(deps: {
8873
globalMode: () => DialMode | undefined;
89-
/** The one-way residency latch. Absent is treated as latched, so behaviour is unchanged. */
90-
everEnabled?: () => boolean | undefined;
91-
/** The per-organisation residency latch. Absent is treated as latched, so behaviour is unchanged. */
92-
everEnabledForOrg?: (organizationId: string) => boolean | undefined;
74+
/**
75+
* The one-way global-mode latch, cold-aware: true when the global dial has ever been non-off, and
76+
* also true when the source is cold, so a cold read never permits a transition skip. Only a loaded
77+
* source with the latch unset returns false. Absent is treated as true (never skip).
78+
*/
79+
globalModeEverEnabled?: () => boolean;
80+
/**
81+
* Whether an org is DEFINITELY never-enabled, per the census. Absent or cold means false, so an
82+
* unknown answer keeps probing rather than suppressing a resident run.
83+
*/
84+
orgDefinitelyNeverEnabled?: (organizationId: string) => boolean;
9385
orgMode: ResolverOrgSource;
9486
/** Run→org resolution for the read path. Absent means readModeFor always falls back to global. */
9587
runOrg?: ResolverRunOrgSource;
@@ -124,13 +116,13 @@ export function buildSnapshotStoreModeResolver(deps: {
124116
};
125117

126118
return {
127-
// False ONLY when the flag is explicitly false. An unreadable or cold registry must never
128-
// report "never enabled", because that would suppress transitions for runs that are resident.
129-
everEnabled: (): boolean => deps.everEnabled?.() !== false,
130-
// Same fail-safe as the global latch: false ONLY when this org's flag is explicitly false, so an
131-
// absent source or unknown org keeps probing rather than suppressing a resident run.
132-
everEnabledForOrg: (organizationId: string): boolean =>
133-
deps.everEnabledForOrg?.(organizationId) !== false,
119+
// Cold-aware read delegated to deps; absent means true, so a transition never skips on a missing
120+
// signal. The deps reader answers true while its source is cold.
121+
globalModeEverEnabled: (): boolean => deps.globalModeEverEnabled?.() ?? true,
122+
// False ONLY on a definite census negative; absent or cold means false, so an unknown answer
123+
// keeps probing rather than suppressing a resident run.
124+
orgDefinitelyNeverEnabled: (organizationId: string): boolean =>
125+
deps.orgDefinitelyNeverEnabled?.(organizationId) ?? false,
134126
// Awaited at birth sites only. Absent org id means nothing to look up, so it is a no-op.
135127
warm: async (organizationId: string): Promise<void> => {
136128
await deps.orgMode.warm?.(organizationId);
@@ -220,7 +212,6 @@ export function createOrgModeSource(clients?: {
220212

221213
return {
222214
get: (organizationId) => cache.get(organizationId)?.mode,
223-
orgEverEnabled: (organizationId) => cache.get(organizationId)?.everEnabled,
224215
invalidate: (organizationId) => {
225216
// Drop first, so a resolve between now and the re-read falls back rather than serving a
226217
// value the write just replaced.
@@ -304,7 +295,6 @@ export function createOrgModeSource(clients?: {
304295
}
305296
cache.set(organizationId, {
306297
mode: cachedOrgModeFor(flags?.[FEATURE_FLAG.snapshotStoreOrgMode]),
307-
everEnabled: cachedOrgEverEnabledFor(flags?.[FEATURE_FLAG.snapshotStoreOrgEverEnabled]),
308298
});
309299
return "loaded" as const;
310300
})
@@ -325,7 +315,15 @@ function orgModeSource(): OrgModeSource {
325315

326316
export const snapshotStoreModeResolver: SnapshotStoreModeResolver = buildSnapshotStoreModeResolver({
327317
globalMode: () => globalFlagsRegistry.current()?.[FEATURE_FLAG.snapshotStoreMode],
328-
everEnabled: () => globalFlagsRegistry.current()?.[FEATURE_FLAG.snapshotStoreEverEnabled],
318+
globalModeEverEnabled: () => {
319+
// Conservative cold default: a cold registry must never permit a transition skip, so answer
320+
// true. Only a loaded registry with the latch unset answers false (skip permitted).
321+
const cur = globalFlagsRegistry.current();
322+
if (cur === undefined) return true;
323+
return cur[FEATURE_FLAG.snapshotStoreGlobalModeEverEnabled] === true;
324+
},
325+
orgDefinitelyNeverEnabled: (organizationId) =>
326+
snapshotStoreOrgCensus.orgDefinitelyNeverEnabled(organizationId),
329327
orgMode: {
330328
get: (organizationId) => orgModeSource().get(organizationId),
331329
refresh: (organizationId) => orgModeSource().refresh(organizationId),

‎apps/webapp/test/snapshotStoreOrgEverEnabled.test.ts‎

Lines changed: 0 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@ import {
55
stampSnapshotStoreOrgEverEnabled,
66
withoutOrgForbiddenSnapshotKeys,
77
} from "~/v3/featureFlags";
8-
import { createOrgModeSource } from "~/v3/snapshotStoreMode.server";
98

109
const MODE = FEATURE_FLAG.snapshotStoreOrgMode;
1110
const LATCH = FEATURE_FLAG.snapshotStoreOrgEverEnabled;
@@ -62,48 +61,3 @@ describe("clearedOrgFlagsPreservingLatch (clear-all keeps the one-way latch)", (
6261
expect(clearedOrgFlagsPreservingLatch({ [LATCH]: false })).toBeNull();
6362
});
6463
});
65-
66-
describe("orgEverEnabled reader", () => {
67-
function clientReturning(flags: unknown) {
68-
return {
69-
organization: { findFirst: () => Promise.resolve({ featureFlags: flags }) },
70-
} as never;
71-
}
72-
73-
it("returns undefined for an unknown organisation (cache miss)", () => {
74-
const source = createOrgModeSource({
75-
primary: clientReturning({}),
76-
replica: clientReturning({}),
77-
});
78-
expect(source.orgEverEnabled("org_unknown")).toBeUndefined();
79-
});
80-
81-
it("returns the cached true latch", async () => {
82-
const source = createOrgModeSource({
83-
primary: clientReturning({}),
84-
replica: clientReturning({ [LATCH]: true }),
85-
});
86-
await source.warm("org_1");
87-
expect(source.orgEverEnabled("org_1")).toBe(true);
88-
});
89-
90-
it("returns the cached false latch, distinct from absent", async () => {
91-
const source = createOrgModeSource({
92-
primary: clientReturning({}),
93-
replica: clientReturning({ [LATCH]: false }),
94-
});
95-
await source.warm("org_1");
96-
expect(source.orgEverEnabled("org_1")).toBe(false);
97-
});
98-
99-
it("returns undefined when the cached flags omit the latch (not false)", async () => {
100-
const source = createOrgModeSource({
101-
primary: clientReturning({}),
102-
replica: clientReturning({ [MODE]: "dual-write" }),
103-
});
104-
await source.warm("org_1");
105-
// The dial is cached, but the latch key is absent, so the reader must not report false.
106-
expect(source.get("org_1")).toBe("dual-write");
107-
expect(source.orgEverEnabled("org_1")).toBeUndefined();
108-
});
109-
});

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

Lines changed: 66 additions & 64 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,11 @@
22
// only record of that, so at `off` AFTER a ramp every transition still has to ask or a resident
33
// run's head freezes while Postgres moves on.
44
//
5-
// Before any ramp, nothing CAN be resident: only a birth creates a keyspace and every birth was
6-
// refused. So the question has one possible answer and asking it is pure cost. Measured: 2 per cent
7-
// with a healthy endpoint, four times the run duration with a slow one, for every run, and it did
8-
// not decay.
9-
//
10-
// The latch is one way. Unset means this deployment has never enabled the store, so transitions skip
11-
// it entirely. Once set it is never cleared, and `off` returns to meaning "drain".
5+
// A transition may therefore be skipped only when nothing of this org's could be resident. A run is
6+
// resident only if its org's dial was non-off at its birth, i.e. the global dial had ever gone
7+
// non-off OR the org itself was ever enabled. So the skip is sound only when the global dial has
8+
// NEVER been non-off (globalModeEverEnabled() === false) AND this org is DEFINITELY never-enabled
9+
// (orgDefinitelyNeverEnabled(org) === true). Any uncertainty falls to "probe".
1210
import { describe, expect, it } from "vitest";
1311
import { TaskRunExecutionSnapshotStore } from "./taskRunExecutionSnapshotStore.js";
1412
import type { SnapshotStoreModeResolver } from "./taskRunExecutionSnapshotStore.js";
@@ -24,8 +22,8 @@ const SCOPE = {
2422

2523
function harness(opts: {
2624
mode: "off" | "dual-write";
27-
everEnabled?: boolean;
28-
everEnabledForOrg?: (organizationId: string) => boolean;
25+
globalModeEverEnabled?: boolean;
26+
orgDefinitelyNeverEnabled?: (organizationId: string) => boolean;
2927
}) {
3028
const touched: string[] = [];
3129
const redis = new Proxy({} as RedisSnapshotStore, {
@@ -42,8 +40,12 @@ function harness(opts: {
4240

4341
const modeResolver: SnapshotStoreModeResolver = {
4442
resolve: () => opts.mode,
45-
...(opts.everEnabled !== undefined && { everEnabled: () => opts.everEnabled! }),
46-
...(opts.everEnabledForOrg && { everEnabledForOrg: opts.everEnabledForOrg }),
43+
...(opts.globalModeEverEnabled !== undefined && {
44+
globalModeEverEnabled: () => opts.globalModeEverEnabled!,
45+
}),
46+
...(opts.orgDefinitelyNeverEnabled && {
47+
orgDefinitelyNeverEnabled: opts.orgDefinitelyNeverEnabled,
48+
}),
4749
};
4850

4951
const decorated = new TaskRunExecutionSnapshotStore(delegate, {
@@ -73,68 +75,44 @@ const completion = {
7375
},
7476
};
7577

76-
describe("the residency latch", () => {
77-
it("a transition touches the store NOT AT ALL while the latch is unset", async () => {
78-
const h = harness({ mode: "off", everEnabled: false });
78+
describe("the sound transition skip", () => {
79+
it("touches the store NOT AT ALL for a definitely-never org while the global dial has never moved", async () => {
80+
// The assertion the whole change exists for: no keyspace can exist for this org, so no probe.
81+
const h = harness({
82+
mode: "dual-write",
83+
globalModeEverEnabled: false,
84+
orgDefinitelyNeverEnabled: () => true,
85+
});
7986

80-
await h.decorated.completeAttemptSuccess("run_1", completion, { select: { id: true } });
87+
await h.decorated.completeAttemptSuccess("run_1", completionForOrg("org_a"), {
88+
select: { id: true },
89+
});
8190

82-
// The assertion the whole change exists for: no probe, so Redis is off the run path entirely.
8391
expect(h.touched).toEqual([]);
8492
});
8593

86-
it("a transition still asks once the latch is set, even at off", async () => {
87-
// Non-negotiable. A run resident from an earlier ramp must keep mirroring, or its head freezes
88-
// and the remedy becomes worse than the fault.
89-
const h = harness({ mode: "off", everEnabled: true });
90-
91-
await h.decorated.completeAttemptSuccess("run_1", completion, { select: { id: true } });
92-
93-
expect(h.touched).toContain("append");
94-
});
95-
96-
it("asks when the resolver offers no latch, so an unlatched deployment is unchanged", async () => {
97-
const h = harness({ mode: "off" });
98-
99-
await h.decorated.completeAttemptSuccess("run_1", completion, { select: { id: true } });
100-
101-
expect(h.touched).toContain("append");
102-
});
103-
104-
it("never suppresses a transition once the dial itself is past off", async () => {
105-
// Belt and braces: the guard makes this state unreachable, but if it ever occurred, suppressing
106-
// transitions while births are mirroring is the one combination that strands a head.
107-
const h = harness({ mode: "dual-write", everEnabled: false });
108-
109-
await h.decorated.completeAttemptSuccess("run_1", completion, { select: { id: true } });
110-
111-
expect(h.touched).toContain("append");
112-
});
113-
});
114-
115-
describe("the per-organisation latch", () => {
116-
it("keeps redis off a never-enabled org's transition path", async () => {
117-
// Deployment-wide latch unset AND this org's latch unset: nothing can be resident for it, so the
118-
// probe is pure cost even though the dial has moved past off for the ramping orgs.
94+
it("still probes an org that has ever been enabled", async () => {
95+
// orgDefinitelyNeverEnabled is false for an org that may hold resident runs, so it keeps asking.
11996
const h = harness({
12097
mode: "dual-write",
121-
everEnabled: false,
122-
everEnabledForOrg: () => false,
98+
globalModeEverEnabled: false,
99+
orgDefinitelyNeverEnabled: (org) => org !== "org_a",
123100
});
124101

125102
await h.decorated.completeAttemptSuccess("run_1", completionForOrg("org_a"), {
126103
select: { id: true },
127104
});
128105

129-
expect(h.touched).toEqual([]);
106+
expect(h.touched).toContain("append");
130107
});
131108

132-
it("still probes an org whose latch is set", async () => {
133-
// An org that has ever been enabled may hold resident runs, so its transitions must keep asking.
109+
it("never skips once the global dial has ever been non-off, even for a definitely-never org", async () => {
110+
// Non-negotiable. Once the global dial has moved, resident runs exist, and suppressing their
111+
// transitions freezes a head while Postgres advances. The global latch beats the per-org one.
134112
const h = harness({
135-
mode: "dual-write",
136-
everEnabled: false,
137-
everEnabledForOrg: (org) => org === "org_a",
113+
mode: "off",
114+
globalModeEverEnabled: true,
115+
orgDefinitelyNeverEnabled: () => true,
138116
});
139117

140118
await h.decorated.completeAttemptSuccess("run_1", completionForOrg("org_a"), {
@@ -144,18 +122,42 @@ describe("the per-organisation latch", () => {
144122
expect(h.touched).toContain("append");
145123
});
146124

147-
it("falls back to the global-only latch when the org is undefined", () => {
148-
// No org id means the per-org latch cannot be consulted, so an unknown org errs toward asking:
149-
// the deployment-wide check alone decides, exactly as before this latch existed.
125+
it("never skips when the organisation is undefined", () => {
126+
// No org id means orgDefinitelyNeverEnabled cannot be consulted, so an unknown org errs toward
127+
// asking. The per-org term is only reached once an org id is supplied.
150128
const h = harness({
151129
mode: "dual-write",
152-
everEnabled: false,
153-
everEnabledForOrg: () => false,
130+
globalModeEverEnabled: false,
131+
orgDefinitelyNeverEnabled: () => true,
154132
});
155133

156-
// Past off with the global latch unset, so the global-only check probes; the per-org term is
157-
// only reached once an org id is supplied.
158134
expect(h.decorated.writesRedisForTransitionTest()).toBe(true);
159135
expect(h.decorated.writesRedisForTransitionTest("org_a")).toBe(false);
160136
});
137+
138+
it("asks when the resolver offers no latch signals, so an unwired deployment is unchanged", async () => {
139+
const h = harness({ mode: "off" });
140+
141+
await h.decorated.completeAttemptSuccess("run_1", completion, { select: { id: true } });
142+
143+
expect(h.touched).toContain("append");
144+
});
145+
146+
it("asks when only one of the two signals permits a skip", () => {
147+
// Both halves are required. A cold census (orgDefinitelyNeverEnabled false) with an unmoved
148+
// global dial must still probe, and vice versa.
149+
const globalOnly = harness({
150+
mode: "off",
151+
globalModeEverEnabled: false,
152+
orgDefinitelyNeverEnabled: () => false,
153+
});
154+
expect(globalOnly.decorated.writesRedisForTransitionTest("org_a")).toBe(true);
155+
156+
const orgOnly = harness({
157+
mode: "off",
158+
globalModeEverEnabled: true,
159+
orgDefinitelyNeverEnabled: () => true,
160+
});
161+
expect(orgOnly.decorated.writesRedisForTransitionTest("org_a")).toBe(true);
162+
});
161163
});

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

Lines changed: 16 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,4 +49,20 @@ describe("store residency is decided at birth", () => {
4949
expect(s.writesRedisForTransitionTest()).toBe(true);
5050
}
5151
});
52+
53+
it("keeps an org-scoped transition on once the global dial has ever moved, whatever the census says", () => {
54+
// The sound skip needs BOTH the global dial unmoved AND the org definitely-never-enabled. With
55+
// the global latch set, a resident run exists, so an org-scoped transition must still mirror even
56+
// if the census believes the org was never enabled.
57+
const s = new TaskRunExecutionSnapshotStore({} as unknown as RunStore, {
58+
store: {} as never,
59+
mode: "off",
60+
modeResolver: {
61+
resolve: () => "off",
62+
globalModeEverEnabled: () => true,
63+
orgDefinitelyNeverEnabled: () => true,
64+
},
65+
});
66+
expect(s.writesRedisForTransitionTest("org_a")).toBe(true);
67+
});
5268
});

0 commit comments

Comments
 (0)