Skip to content

Commit 071d4cc

Browse files
authored
Merge branch 'main' into changeset-release/main
2 parents e3f1f31 + 2be2d39 commit 071d4cc

32 files changed

Lines changed: 3682 additions & 129 deletions

‎apps/webapp/app/env.server.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1385,6 +1385,8 @@ const EnvironmentSchema = z
13851385

13861386
// Infrastructure endpoint only. Modes and organization overrides are polled feature flags.
13871387
RUN_ENGINE_SNAPSHOT_STORE_REDIS_URL: z.string().optional(),
1388+
// How long a fetched org snapshotStoreMode override is reused per process. 0 disables caching.
1389+
RUN_ENGINE_SNAPSHOT_STORE_DIAL_CACHE_MS: z.coerce.number().int().min(0).default(5000),
13881390

13891391
RUN_ENGINE_RUN_LOCK_REDIS_HOST: z
13901392
.string()

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -223,7 +223,9 @@ const snapshotStoreRuntime = singleton("SnapshotStoreRuntime", () => {
223223
});
224224
return createSnapshotStoreRuntime({
225225
connection,
226-
rollout: createSnapshotRolloutResolver(() => globalFlagsRegistry.current(), prisma),
226+
rollout: createSnapshotRolloutResolver(() => globalFlagsRegistry.current(), prisma, {
227+
dialCacheMs: env.RUN_ENGINE_SNAPSHOT_STORE_DIAL_CACHE_MS,
228+
}),
227229
getRunStore: () => runStore,
228230
metrics: metrics.decorator,
229231
});

‎apps/webapp/app/v3/utils/snapshotStoreConnection.server.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,10 @@
11
import { createRedisClusterClient, type Cluster } from "@internal/redis";
22
import { RedisSnapshotStore, type SnapshotStoreMetrics } from "@internal/run-store";
33

4-
const COMMAND_TIMEOUT_MS = 500;
4+
const COMMAND_TIMEOUT_MS = 1_000;
55
const CONNECT_TIMEOUT_MS = 500;
66
const POLICY_CHECK_INTERVAL_MS = 30_000;
7-
const COMPLETED_TTL_MS = 14 * 24 * 60 * 60 * 1_000;
7+
const COMPLETED_TTL_MS = 4 * 24 * 60 * 60 * 1_000;
88

99
/** Infrastructure only. Rollout modes belong to the existing polled flags, not this URL. */
1010
export function snapshotClusterOptions(endpoint: string) {

‎apps/webapp/app/v3/utils/snapshotStoreMetrics.server.test.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ test("snapshot metrics ignore raw organization IDs; the endpoint creates no eage
2424
const plainOptions = snapshotClusterOptions("redis://127.0.0.1:1");
2525
expect(plainOptions.clusterOptions).toEqual({ lazyConnect: true, scaleReads: "master" });
2626
expect(plainOptions.redisOptions.tls).toBeUndefined();
27-
expect(options.redisOptions.commandTimeout).toBe(500);
27+
expect(options.redisOptions.commandTimeout).toBe(1000);
2828
const connection = createSnapshotConnection("redis://127.0.0.1:1");
2929
connection.getStore(); // Registers local Lua commands only.
3030
expect(connection.getClient().status).toBe("wait");

‎apps/webapp/app/v3/utils/snapshotStoreRollout.server.test.ts‎

Lines changed: 81 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,87 @@ import {
2020
type SnapshotRolloutFlags,
2121
} from "./snapshotStoreRollout.server";
2222

23+
function countOrganizationReads<T extends { $extends: (...args: any[]) => any }>(prisma: T) {
24+
const counter = { reads: 0 };
25+
const observed = prisma.$extends({
26+
query: {
27+
organization: {
28+
async findFirst({ args, query }: { args: unknown; query: (args: unknown) => unknown }) {
29+
counter.reads++;
30+
return query(args);
31+
},
32+
},
33+
},
34+
});
35+
return { counter, observed };
36+
}
37+
38+
postgresTest(
39+
"snapshotStoreMode org lookups are cached per organization for the TTL",
40+
async ({ prisma }) => {
41+
const org = await prisma.organization.create({
42+
data: {
43+
title: "Snapshot dial cache",
44+
slug: "snapshot-dial-cache",
45+
featureFlags: { snapshotStoreMode: "redis-only" },
46+
},
47+
});
48+
const { counter, observed } = countOrganizationReads(prisma);
49+
50+
// Two hits within the TTL make one DB call, even after the org row changes.
51+
const cached = createSnapshotRolloutResolver(() => undefined, observed, { dialCacheMs: 200 });
52+
expect(await cached.resolveDial(org.id)).toBe("redis-only");
53+
await prisma.organization.update({
54+
where: { id: org.id },
55+
data: { featureFlags: { snapshotStoreMode: "dual-write" } },
56+
});
57+
expect(await cached.resolveDial(org.id)).toBe("redis-only");
58+
expect(counter.reads).toBe(1);
59+
60+
// After the TTL the lookup runs again and sees the new value.
61+
await new Promise((resolve) => setTimeout(resolve, 250));
62+
expect(await cached.resolveDial(org.id)).toBe("dual-write");
63+
expect(counter.reads).toBe(2);
64+
65+
// An absent override is cached too.
66+
await prisma.organization.update({
67+
where: { id: org.id },
68+
data: { featureFlags: Prisma.DbNull },
69+
});
70+
const absent = createSnapshotRolloutResolver(
71+
() => ({ snapshotStoreMode: "redis-read" }),
72+
observed,
73+
{ dialCacheMs: 200 }
74+
);
75+
counter.reads = 0;
76+
expect(await absent.resolveDial(org.id)).toBe("redis-read");
77+
expect(await absent.resolveDial(org.id)).toBe("redis-read");
78+
expect(counter.reads).toBe(1);
79+
80+
// Supplied organization flags bypass the cache and are never stored in it.
81+
const supplied = createSnapshotRolloutResolver(() => undefined, observed, { dialCacheMs: 200 });
82+
counter.reads = 0;
83+
expect(await supplied.resolveDial(org.id, { snapshotStoreMode: "redis-only" })).toBe(
84+
"redis-only"
85+
);
86+
expect(counter.reads).toBe(0);
87+
expect(await supplied.resolveDial(org.id)).toBe("off");
88+
expect(counter.reads).toBe(1);
89+
expect(await supplied.resolveDial(org.id, { snapshotStoreMode: "dual-write" })).toBe(
90+
"dual-write"
91+
);
92+
expect(await supplied.resolveDial(org.id)).toBe("off");
93+
expect(counter.reads).toBe(1);
94+
95+
// TTL 0 disables caching: every read makes a DB call.
96+
const uncached = createSnapshotRolloutResolver(() => undefined, observed, { dialCacheMs: 0 });
97+
counter.reads = 0;
98+
await uncached.resolveDial(org.id);
99+
await uncached.resolveDial(org.id);
100+
expect(counter.reads).toBe(2);
101+
}
102+
);
103+
23104
postgresTest(
24105
"snapshotStoreMode uses loaded org flags or one lookup, then the global default",
25106
async ({ prisma }) => {

‎apps/webapp/app/v3/utils/snapshotStoreRollout.server.ts‎

Lines changed: 39 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -4,26 +4,55 @@ import { FEATURE_FLAG, FeatureFlagCatalog } from "../featureFlags";
44

55
export type SnapshotRolloutFlags = { snapshotStoreMode?: SnapshotStoreDial };
66

7-
/** Use the normal org flag, falling back to the already-polled global default. No org cache. */
7+
const DIAL_CACHE_MAX_ENTRIES = 10_000;
8+
9+
/**
10+
* Use the normal org flag, falling back to the already-polled global default.
11+
* Fetched org overrides are cached in memory per organization for `dialCacheMs` (absent or 0
12+
* disables), so a flag flip takes up to that long to apply.
13+
*/
814
export function createSnapshotRolloutResolver(
915
readFlags: () => SnapshotRolloutFlags | undefined,
10-
prisma: Pick<PrismaClient, "organization">
16+
prisma: Pick<PrismaClient, "organization">,
17+
options?: { dialCacheMs?: number }
1118
) {
19+
const ttlMs = options?.dialCacheMs ?? 0;
20+
// Insertion-ordered, so the first key is the oldest entry for FIFO eviction.
21+
const overridesCache = new Map<string, { overrides: unknown; fetchedAt: number }>();
22+
23+
async function fetchOverrides(organizationId: string): Promise<unknown> {
24+
if (ttlMs > 0) {
25+
const cached = overridesCache.get(organizationId);
26+
if (cached) {
27+
if (Date.now() - cached.fetchedAt < ttlMs) return cached.overrides;
28+
overridesCache.delete(organizationId);
29+
}
30+
}
31+
const overrides = (
32+
await prisma.organization.findFirst({
33+
where: { id: organizationId },
34+
select: { featureFlags: true },
35+
})
36+
)?.featureFlags;
37+
if (ttlMs > 0) {
38+
overridesCache.delete(organizationId);
39+
if (overridesCache.size >= DIAL_CACHE_MAX_ENTRIES) {
40+
const oldest = overridesCache.keys().next().value;
41+
if (oldest !== undefined) overridesCache.delete(oldest);
42+
}
43+
overridesCache.set(organizationId, { overrides, fetchedAt: Date.now() });
44+
}
45+
return overrides;
46+
}
47+
1248
return {
1349
async resolveDial(
1450
organizationId: string,
1551
organizationFlags?: unknown
1652
): Promise<SnapshotStoreDial> {
1753
// undefined means not loaded; null/{} means loaded with no override. Do not refetch those.
1854
const overrides =
19-
organizationFlags === undefined
20-
? (
21-
await prisma.organization.findFirst({
22-
where: { id: organizationId },
23-
select: { featureFlags: true },
24-
})
25-
)?.featureFlags
26-
: organizationFlags;
55+
organizationFlags === undefined ? await fetchOverrides(organizationId) : organizationFlags;
2756
const value =
2857
typeof overrides === "object" && overrides !== null && "snapshotStoreMode" in overrides
2958
? overrides.snapshotStoreMode

‎internal-packages/run-engine/src/engine/index.ts‎

Lines changed: 82 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -78,13 +78,14 @@ import {
7878
ExecutionSnapshotSystem,
7979
getExecutionSnapshotsSince,
8080
getLatestExecutionSnapshot,
81+
getLatestExecutionSnapshotOrFinishedRun,
8182
} from "./systems/executionSnapshotSystem.js";
8283
import {
8384
PARKED_ON_EXTERNAL_DEPLOYMENT_STATUS_REASON,
8485
PendingVersionSystem,
8586
} from "./systems/pendingVersionSystem.js";
8687
import { RaceSimulationSystem } from "./systems/raceSimulationSystem.js";
87-
import { RunAttemptSystem } from "./systems/runAttemptSystem.js";
88+
import { type CancelRunResult, RunAttemptSystem } from "./systems/runAttemptSystem.js";
8889
import { NoopPendingVersionRunIdLookup } from "./services/pendingVersionLookup.js";
8990
import type { SystemResources } from "./systems/systems.js";
9091
import {
@@ -1753,7 +1754,7 @@ export class RunEngine {
17531754
finalizeRun?: boolean;
17541755
bulkActionId?: string;
17551756
tx?: PrismaClientOrTransaction;
1756-
}): Promise<ExecutionResult & { alreadyFinished: boolean }> {
1757+
}): Promise<CancelRunResult> {
17571758
return this.runAttemptSystem.cancelRun({
17581759
runId,
17591760
organizationId,
@@ -2718,13 +2719,35 @@ export class RunEngine {
27182719
}
27192720

27202721
async #repairRun(runId: string, dryRun: boolean, organizationId: string) {
2721-
const snapshot = await getLatestExecutionSnapshot(
2722+
const { snapshot, finishedRun } = await getLatestExecutionSnapshotOrFinishedRun(
27222723
this.prisma,
27232724
runId,
27242725
this.runStore,
2725-
undefined,
2726-
organizationId
2726+
organizationId,
2727+
{ id: true }
27272728
);
2729+
if (finishedRun) {
2730+
// The run finished and its snapshot state expired, so any concurrency slot it holds has leaked.
2731+
this.logger.warn("RunEngine.repairRun: snapshot state expired, run already finished", {
2732+
runId,
2733+
});
2734+
2735+
if (!dryRun) {
2736+
await this.worker.enqueueOnce({
2737+
id: `repair-in-progress-run:${runId}`,
2738+
job: "repairSnapshot",
2739+
payload: { runId, snapshotId: null, executionStatus: "FINISHED" },
2740+
availableAt: new Date(Date.now() + this.repairSnapshotTimeoutMs),
2741+
});
2742+
}
2743+
2744+
return {
2745+
action: "repairSnapshot",
2746+
runId,
2747+
snapshotStatus: "FINISHED",
2748+
snapshotId: null,
2749+
};
2750+
}
27282751

27292752
if (
27302753
snapshot.executionStatus === "QUEUED" ||
@@ -3173,13 +3196,57 @@ export class RunEngine {
31733196
executionStatus,
31743197
}: {
31753198
runId: string;
3176-
snapshotId: string;
3199+
/** null when scheduled for a finished run whose snapshot state had already expired. */
3200+
snapshotId: string | null;
31773201
executionStatus: string;
31783202
}) {
31793203
return await this.runLock.lock("handleRepairSnapshot", [runId], async () => {
31803204
// No snapshot mirror in off-only mode, so the repair heals only the queue below. The M1+
31813205
// rebuild reintroduces a mirror-repair step here.
3182-
const latestSnapshot = await getLatestExecutionSnapshot(this.prisma, runId, this.runStore);
3206+
const { snapshot: latestSnapshot, finishedRun } =
3207+
await getLatestExecutionSnapshotOrFinishedRun(
3208+
this.prisma,
3209+
runId,
3210+
this.runStore,
3211+
undefined,
3212+
{
3213+
id: true,
3214+
queue: true,
3215+
concurrencyKey: true,
3216+
runtimeEnvironmentId: true,
3217+
}
3218+
);
3219+
if (finishedRun) {
3220+
// The run is finished whichever snapshot the job was queued from (a live one that has since
3221+
// finished and expired, or none when queued after expiry), so its slot has leaked either way.
3222+
this.logger.warn("RunEngine.handleRepairSnapshot: snapshot state expired, run finished", {
3223+
runId,
3224+
snapshotId,
3225+
repairExecutionStatus: executionStatus,
3226+
});
3227+
3228+
const env = await this.controlPlaneResolver.resolveEnv(finishedRun.runtimeEnvironmentId);
3229+
if (!env) {
3230+
this.logger.error("RunEngine.handleRepairSnapshot FINISHED environment not found", {
3231+
runId,
3232+
});
3233+
return;
3234+
}
3235+
3236+
await this.#clearRunFromConcurrencySets({
3237+
runId,
3238+
orgId: env.organizationId,
3239+
queue: finishedRun.queue,
3240+
concurrencyKey: finishedRun.concurrencyKey ?? undefined,
3241+
env: {
3242+
id: env.id,
3243+
type: env.type,
3244+
project: { id: env.projectId },
3245+
organization: { id: env.organizationId },
3246+
},
3247+
});
3248+
return;
3249+
}
31833250

31843251
if (latestSnapshot.id !== snapshotId) {
31853252
this.logger.log(
@@ -3264,8 +3331,7 @@ export class RunEngine {
32643331
return;
32653332
}
32663333

3267-
// We need to clear this run from the current concurrency sets
3268-
await this.runQueue.clearMessageFromConcurrencySets({
3334+
await this.#clearRunFromConcurrencySets({
32693335
runId,
32703336
orgId: latestSnapshot.organizationId,
32713337
queue: taskRun.queue,
@@ -3291,6 +3357,13 @@ export class RunEngine {
32913357
});
32923358
}
32933359

3360+
async #clearRunFromConcurrencySets(
3361+
params: Parameters<RunQueue["clearMessageFromConcurrencySets"]>[0]
3362+
) {
3363+
// We need to clear this run from the current concurrency sets
3364+
await this.runQueue.clearMessageFromConcurrencySets(params);
3365+
}
3366+
32943367
/**
32953368
* Applies `defaultMaxTtl` to a run's TTL:
32963369
* - No max configured → pass through as-is.

0 commit comments

Comments
 (0)