Skip to content

Commit 64fbff7

Browse files
committed
perf(run-store): take Redis off the run path for a non-resident run
A run's residency is its Redis keyspace, and the test for it lives in the append script, so the store had to complete a round trip just to learn a run was not its own. That put Redis on the path of every transition of every run, resident or not: two percent with a healthy Redis, four times with a slow one, and it never decayed, because a fleet holding no resident runs still asked once per transition. Residency is monotonic, which is what makes a local answer sound. Only a birth creates a keyspace: the script refuses a transition into a dead one, and the repair appends as a transition. So a keyspace the script reports absent is absent for good, and that answer may skip the network. The reverse is a hint only, and a stale one costs a single round trip that returns skippedNoKeyspace. Only the script's own reply may create a negative. A birth path can be re-entered, so a birth that did mirror can reach the not-mirrored branch on a retry once the short-lived override cache has moved, and inferring absence from that local decision would suppress every later transition of a resident run and freeze its head.
1 parent 1cb15ab commit 64fbff7

7 files changed

Lines changed: 351 additions & 1 deletion

File tree

‎internal-packages/run-store/package.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,8 @@
1616
"dependencies": {
1717
"@internal/redis": "workspace:*",
1818
"@trigger.dev/core": "workspace:*",
19-
"@trigger.dev/database": "workspace:*"
19+
"@trigger.dev/database": "workspace:*",
20+
"lru-cache": "^11.2.4"
2021
},
2122
"devDependencies": {
2223
"@internal/run-ops-database": "workspace:*",

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

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@ import {
77
} from "@internal/redis";
88
import { Logger } from "@trigger.dev/core/logger";
99
import type { CompletedWaitpoint } from "@trigger.dev/core/v3/schemas";
10+
import { ResidencyCache } from "./residencyCache.js";
1011

1112
export type SnapshotKeys = { e: string; idx: string; cur: string; seq: string };
1213

@@ -239,6 +240,8 @@ export type RedisSnapshotStoreConnection =
239240

240241
export type RedisSnapshotStoreOptions = RedisSnapshotStoreConnection & {
241242
completedTtlMs: number;
243+
/** Entries in the per-process residency cache. See {@link ResidencyCache}. */
244+
residencyCacheMax?: number;
242245
sinceLimit?: number;
243246
highWater?: { entryBytes?: number; cycleKeyBytes?: number; cycleCount?: number };
244247
metrics?: SnapshotStoreMetrics;
@@ -269,6 +272,13 @@ export class RedisSnapshotStore {
269272
private readonly sinceLimit: number;
270273
private readonly metrics?: SnapshotStoreMetrics;
271274
private readonly highWater: NonNullable<RedisSnapshotStoreOptions["highWater"]>;
275+
/**
276+
* Which runs this process knows the mirror does not own, so a transition for one of them never
277+
* reaches the network. Lives here rather than on the decorator because the decorator is re-minted
278+
* per transaction while this store instance is shared, and because this class owns the append
279+
* replies that are the cache's only ground truth.
280+
*/
281+
readonly #residency: ResidencyCache;
272282
#quit?: Promise<void>;
273283

274284
constructor(options: RedisSnapshotStoreOptions) {
@@ -277,6 +287,9 @@ export class RedisSnapshotStore {
277287
this.sinceLimit = options.sinceLimit ?? 50;
278288
this.metrics = options.metrics;
279289
this.highWater = options.highWater ?? {};
290+
this.#residency = new ResidencyCache({
291+
...(options.residencyCacheMax !== undefined && { max: options.residencyCacheMax }),
292+
});
280293
this.ownsClient = options.client === undefined;
281294
this.redis =
282295
options.client ??
@@ -311,13 +324,20 @@ export class RedisSnapshotStore {
311324
}
312325
}
313326

327+
/** Test seam. */
328+
residencyFor(runId: string): "resident" | "non-resident" | undefined {
329+
return this.#residency.get(runId);
330+
}
331+
314332
/**
315333
* Removes a run's whole keyspace, wait-cycle keys included. The caller must have established that
316334
* the head cannot be trusted and that Postgres still holds the run's rows.
317335
*/
318336
async dropRun(runId: string): Promise<void> {
319337
const keys = snapshotKeys(runId);
320338
await this.redis.dropSnapshotRun(keys.e, keys.idx, keys.cur, keys.seq);
339+
// This process, at least, stops asking. Other processes learn it from their next append.
340+
this.#residency.setNonResident(runId);
321341
}
322342

323343
async append(args: {
@@ -352,6 +372,15 @@ export class RedisSnapshotStore {
352372
"Writing it into the entry JSON breaks byte-comparability with the Postgres row."
353373
);
354374
}
375+
// The whole point of the cache. A transition into a keyspace this process already knows is gone
376+
// is refused without a round trip. A birth deliberately never takes this path: a birth is what
377+
// CREATES residency, so it must reach the script even when the run is currently unknown.
378+
if (args.kind === "transition" && this.#residency.get(args.entry.runId) === "non-resident") {
379+
this.metrics?.recordSkippedNoKeyspace();
380+
this.metrics?.recordAppend("skippedNoKeyspace", "none");
381+
return { outcome: "skippedNoKeyspace" };
382+
}
383+
355384
return this.#timed("append", async () => {
356385
const k = snapshotKeys(args.entry.runId);
357386
const raw = JSON.stringify(args.entry);
@@ -418,15 +447,22 @@ export class RedisSnapshotStore {
418447
runId: string
419448
): AppendResult {
420449
if (reply[0] === SKIPPED) {
450+
// Authoritative and final: the script looked and there is no keyspace. Only a birth could
451+
// create one and this run's birth has already happened.
452+
this.#residency.setNonResident(runId);
421453
this.metrics?.recordSkippedNoKeyspace();
422454
this.metrics?.recordAppend("skippedNoKeyspace", "none");
423455
return { outcome: "skippedNoKeyspace" };
424456
}
425457
if (reply[0] === FORKED) {
458+
// A fork means the script found a keyspace and disagreed about its head, so the run IS
459+
// resident.
460+
this.#residency.setResident(runId);
426461
this.metrics?.recordAppend("forked", "none");
427462
return { outcome: "forked", actualCur: reply[1] ?? "" };
428463
}
429464
if (reply[0] === DUPLICATE) {
465+
this.#residency.setResident(runId);
430466
this.metrics?.recordAppend("duplicate", "none");
431467
return { outcome: "duplicate", seq: Number(reply[1]) };
432468
}
@@ -437,6 +473,8 @@ export class RedisSnapshotStore {
437473
if (cycleMismatch) {
438474
this.metrics?.recordCycleMismatch();
439475
}
476+
// A written entry proves the keyspace exists. For a birth this is what makes the run resident.
477+
this.#residency.setResident(runId);
440478
this.#observeSizes(raw, orderJson, records, cycleSeq, runId);
441479
this.metrics?.recordAppend("written", ttl);
442480
return {
Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
// The keyspace is a run's residency record, and it lives in Redis, so the store had to ask Redis to
2+
// learn a run was NOT its own. That put Redis on the hot path for every transition of every run,
3+
// resident or not: 2 percent with a healthy Redis, 4 times with a slow one, and it never decayed,
4+
// because a fleet with no resident runs still asked once per transition.
5+
//
6+
// The cache is sound because residency is monotonic. Only a birth creates a keyspace: the append
7+
// script refuses `kind: "transition"` into a dead one, and the repair appends as a transition. So
8+
// non-resident is permanent, and a cached negative can never suppress a resident run's append.
9+
import { describe, expect, it } from "vitest";
10+
import { ResidencyCache } from "./residencyCache.js";
11+
12+
describe("ResidencyCache", () => {
13+
it("has no opinion about a run it has never seen", () => {
14+
expect(new ResidencyCache().get("run_a")).toBeUndefined();
15+
});
16+
17+
it("remembers a run that has no keyspace, which is the whole point", () => {
18+
const cache = new ResidencyCache();
19+
cache.setNonResident("run_a");
20+
expect(cache.get("run_a")).toBe("non-resident");
21+
});
22+
23+
it("remembers a run that has one", () => {
24+
const cache = new ResidencyCache();
25+
cache.setResident("run_a");
26+
expect(cache.get("run_a")).toBe("resident");
27+
});
28+
29+
it("never lets a negative be overwritten by a positive", () => {
30+
// Non-resident is permanent, so a positive arriving afterwards is stale information, not news.
31+
// Honouring it would start mirroring a run half way through its life, which is the one thing
32+
// the residency model forbids.
33+
const cache = new ResidencyCache();
34+
cache.setNonResident("run_a");
35+
cache.setResident("run_a");
36+
expect(cache.get("run_a")).toBe("non-resident");
37+
});
38+
39+
it("lets a positive be replaced by a negative", () => {
40+
// The safe direction: a keyspace can go away under a completion expiry, a sweep, or an
41+
// eviction. Believing that costs nothing, because the append script refuses the write anyway.
42+
const cache = new ResidencyCache();
43+
cache.setResident("run_a");
44+
cache.setNonResident("run_a");
45+
expect(cache.get("run_a")).toBe("non-resident");
46+
});
47+
48+
it("stays inside its bound", () => {
49+
const cache = new ResidencyCache({ max: 10 });
50+
for (let i = 0; i < 100; i++) {
51+
cache.setNonResident(`run_${i}`);
52+
}
53+
expect(cache.size).toBeLessThanOrEqual(10);
54+
// An evicted entry is a cache miss, never a wrong answer: the next transition probes Redis once
55+
// and re-learns it.
56+
expect(cache.get("run_0")).toBeUndefined();
57+
expect(cache.get("run_99")).toBe("non-resident");
58+
});
59+
});
Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
import { LRUCache } from "lru-cache";
2+
3+
export type Residency = "resident" | "non-resident";
4+
5+
/** Entries, not bytes. At the default this is roughly 25 MB of run ids. */
6+
const DEFAULT_MAX = 250_000;
7+
8+
/**
9+
* Per-process memory of which runs the mirror owns, so the hot path stops asking Redis.
10+
*
11+
* The keyspace is a run's residency record and it lives in Redis, so learning that a run is NOT
12+
* resident used to cost a round trip, once per transition, forever. Under a brownout that is the
13+
* full retry budget for a question whose answer never changes.
14+
*
15+
* Soundness rests on residency being monotonic. Only a birth creates a keyspace: the append script
16+
* refuses `kind: "transition"` into a dead one, and the repair appends as a transition, so nothing
17+
* else can mint one. Therefore:
18+
*
19+
* - `non-resident` is PERMANENT, and may be trusted to skip the network entirely.
20+
* - `resident` is a hint. A keyspace can still go away under a completion expiry, a sweep, or an
21+
* eviction, so a stale positive costs one round trip that returns `skippedNoKeyspace`. That is
22+
* the safe direction, and it is why the cache never needs a TTL.
23+
*
24+
* There is no negative-cache invalidation and there must not be one. A run that was told it has no
25+
* keyspace has to keep that answer for life, or it would change stores half way through.
26+
*/
27+
export class ResidencyCache {
28+
readonly #entries: LRUCache<string, Residency>;
29+
30+
constructor(options: { max?: number } = {}) {
31+
this.#entries = new LRUCache({ max: options.max ?? DEFAULT_MAX });
32+
}
33+
34+
get(runId: string): Residency | undefined {
35+
return this.#entries.get(runId);
36+
}
37+
38+
/** Hint only. Refused once the run is known non-resident, because that answer is final. */
39+
setResident(runId: string): void {
40+
if (this.#entries.get(runId) === "non-resident") {
41+
return;
42+
}
43+
this.#entries.set(runId, "resident");
44+
}
45+
46+
setNonResident(runId: string): void {
47+
this.#entries.set(runId, "non-resident");
48+
}
49+
50+
get size(): number {
51+
return this.#entries.size;
52+
}
53+
}

0 commit comments

Comments
 (0)