diff --git a/apps/server/src/routes/threads/data.ts b/apps/server/src/routes/threads/data.ts index f8a741ade..e3c9a3007 100644 --- a/apps/server/src/routes/threads/data.ts +++ b/apps/server/src/routes/threads/data.ts @@ -56,7 +56,6 @@ import type { } from "../../services/threads/timeline-pagination.js"; import { createSlowThreadTimelineBuildLogger } from "../../services/threads/timeline-build-log.js"; import { - buildThreadTimelineCacheKey, buildThreadTimelineParamsKey, createThreadTimelineCache, } from "../../services/threads/timeline-cache.js"; @@ -355,8 +354,9 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void { summaryOnly, includeProviderUnhandledOperations, }; + const paramsKey = buildThreadTimelineParamsKey(keyArgs); const full = timelineCache.getOrBuild( - buildThreadTimelineCacheKey({ ...keyArgs, maxSeq }), + { paramsKey, maxSeq }, () => { const { profile, response } = buildThreadTimelineWithProfile( deps.db, @@ -399,7 +399,6 @@ export function registerThreadDataRoutes(app: Hono, deps: AppDeps): void { query.afterSequence, "afterSequence", ); - const paramsKey = buildThreadTimelineParamsKey(keyArgs); const previous = afterSequence === undefined ? undefined diff --git a/apps/server/src/services/threads/timeline-cache.ts b/apps/server/src/services/threads/timeline-cache.ts index f5e1c903a..a8bca92f9 100644 --- a/apps/server/src/services/threads/timeline-cache.ts +++ b/apps/server/src/services/threads/timeline-cache.ts @@ -13,9 +13,12 @@ import type { ThreadTimelinePageRequest } from "./timeline-pagination.js"; * (detail view + side-chat tabs), debounced realtime invalidations that fire * after the tail already settled, and re-opening a thread. * - * Keying on the thread high-water `maxSeq` makes invalidation implicit: any - * appended event bumps `maxSeq`, producing a new key and a cold rebuild. The - * key MUST also include every other input the projection depends on: + * Entries are keyed by request shape (`paramsKey`) and store the thread + * high-water `maxSeq` they were built at. A request with a different `maxSeq` + * is a miss that *replaces* the slot: `maxSeq` never decreases, so the old + * revision could never be looked up again and keeping it until global LRU + * eviction only pins a dead response per appended event (#2066). The request + * shape MUST include every other input the projection depends on: * `thread.status` (interrupt flips earlier rows), `environmentId` (workspace * root relativizes file paths), provider display name (labels dynamic-provider * diagnostic rows), and the row-shape request flags. Event pruning @@ -23,10 +26,9 @@ import type { ThreadTimelinePageRequest } from "./timeline-pagination.js"; * and never lowers `maxSeq`, so it cannot stale a cached entry. * * Entries with many rows are not cached: an expanded active turn (the streaming - * case) produces hundreds of rows AND a `maxSeq` that changes on every event, - * so caching it only thrashes the LRU and pins large objects for no reuse. Idle - * windows collapse completed turns to a handful of rows regardless of thread - * size, so the cap excludes exactly the entries that would never be reused. + * case) produces hundreds of rows that are rebuilt on every event, so storing + * them pins a large object for no reuse. The per-shape slot bounds the *count* + * of retained revisions; the row cap bounds their *size*. */ const DEFAULT_MAX_ENTRIES = 128; @@ -40,7 +42,7 @@ interface ThreadTimelineCacheOptions { interface ThreadTimelineCache { getOrBuild( - key: string, + key: { paramsKey: string; maxSeq: number }, build: () => ThreadTimelineResponse, ): ThreadTimelineResponse; /** Number of currently cached entries (for tests/metrics). */ @@ -53,21 +55,27 @@ export function createThreadTimelineCache( const maxEntries = options.maxEntries ?? DEFAULT_MAX_ENTRIES; const maxCacheableRows = options.maxCacheableRows ?? DEFAULT_MAX_CACHEABLE_ROWS; - const entries = new Map(); + const entries = new Map< + string, + { maxSeq: number; value: ThreadTimelineResponse } + >(); return { - getOrBuild(key, build) { - const cached = entries.get(key); - if (cached !== undefined) { + getOrBuild({ paramsKey, maxSeq }, build) { + const cached = entries.get(paramsKey); + if (cached?.maxSeq === maxSeq) { // Re-insert to mark most-recently-used. - entries.delete(key); - entries.set(key, cached); - return cached; + entries.delete(paramsKey); + entries.set(paramsKey, cached); + return cached.value; } const value = build(); + // A newer revision supersedes the stored one even when the new value is + // too large to cache: the old one can never hit again. + entries.delete(paramsKey); if (value.rows.length <= maxCacheableRows) { - entries.set(key, value); + entries.set(paramsKey, { maxSeq, value }); while (entries.size > maxEntries) { const oldest = entries.keys().next().value; if (oldest === undefined) { @@ -86,8 +94,6 @@ export function createThreadTimelineCache( export interface ThreadTimelineCacheKeyArgs { threadId: string; - /** Thread high-water event sequence; bumps on every appended event. */ - maxSeq: number; status: ThreadStatus; environmentId: string | null; providerDisplayName?: string; @@ -104,12 +110,12 @@ function pageKeyPart(page: ThreadTimelinePageRequest): string { } /** - * The cache identity *excluding* `maxSeq` — i.e. everything that selects which - * window is being requested, but not which revision of it. Used to track the - * latest-sent rows per request shape for delta computation. + * The request shape: everything that selects which window is being requested, + * but not which revision (`maxSeq`) of it. Shared by the response cache and the + * latest-rows delta cache. */ export function buildThreadTimelineParamsKey( - args: Omit, + args: ThreadTimelineCacheKeyArgs, ): string { return [ args.threadId, @@ -122,9 +128,3 @@ export function buildThreadTimelineParamsKey( args.includeProviderUnhandledOperations ? "1" : "0", ].join("|"); } - -export function buildThreadTimelineCacheKey( - args: ThreadTimelineCacheKeyArgs, -): string { - return `${args.maxSeq}|${buildThreadTimelineParamsKey(args)}`; -} diff --git a/apps/server/test/services/threads/timeline-cache.test.ts b/apps/server/test/services/threads/timeline-cache.test.ts index 141af64fe..7ef779e94 100644 --- a/apps/server/test/services/threads/timeline-cache.test.ts +++ b/apps/server/test/services/threads/timeline-cache.test.ts @@ -2,7 +2,7 @@ import { describe, expect, it, vi } from "vitest"; import type { ThreadTimelineResponse } from "@bb/server-contract"; import type { ThreadTimelinePageRequest } from "../../../src/services/threads/timeline-pagination.js"; import { - buildThreadTimelineCacheKey, + buildThreadTimelineParamsKey, createThreadTimelineCache, type ThreadTimelineCacheKeyArgs, } from "../../../src/services/threads/timeline-cache.js"; @@ -48,7 +48,6 @@ const latestPage: ThreadTimelinePageRequest = { const baseKeyArgs: ThreadTimelineCacheKeyArgs = { threadId: "thr_x", - maxSeq: 10, status: "idle", environmentId: null, page: latestPage, @@ -57,62 +56,92 @@ const baseKeyArgs: ThreadTimelineCacheKeyArgs = { includeProviderUnhandledOperations: false, }; +const k = (paramsKey: string, maxSeq = 1) => ({ paramsKey, maxSeq }); + describe("createThreadTimelineCache", () => { - it("builds once for the same key and serves cached on repeat", () => { + it("builds once for the same shape and revision and serves cached on repeat", () => { const cache = createThreadTimelineCache(); const build = vi.fn(() => makeResponse(3)); - const first = cache.getOrBuild("k", build); - const second = cache.getOrBuild("k", build); + const first = cache.getOrBuild(k("k"), build); + const second = cache.getOrBuild(k("k"), build); expect(build).toHaveBeenCalledTimes(1); expect(second).toBe(first); expect(cache.size).toBe(1); }); - it("rebuilds when the key changes (e.g. new maxSeq)", () => { + it("rebuilds on a new maxSeq and replaces the prior revision of the same shape", () => { const cache = createThreadTimelineCache(); const build = vi.fn(() => makeResponse(3)); - cache.getOrBuild("k1", build); - cache.getOrBuild("k2", build); + cache.getOrBuild(k("k", 10), build); + cache.getOrBuild(k("k", 11), build); expect(build).toHaveBeenCalledTimes(2); + expect(cache.size).toBe(1); + }); + + it("never returns a newer revision to a request for an older maxSeq", () => { + const cache = createThreadTimelineCache(); + const newer = makeResponse(2); + cache.getOrBuild(k("k", 11), () => newer); + const rebuilt = makeResponse(1); + const served = cache.getOrBuild(k("k", 10), () => rebuilt); + expect(served).toBe(rebuilt); + }); + + it("retains separate request shapes independently", () => { + const cache = createThreadTimelineCache(); + const build = vi.fn(() => makeResponse(3)); + + cache.getOrBuild(k("latest", 10), build); + cache.getOrBuild(k("older:5", 10), build); + + expect(cache.size).toBe(2); }); it("does not cache responses above the row cap (streaming expanded turns)", () => { const cache = createThreadTimelineCache({ maxCacheableRows: 5 }); const build = vi.fn(() => makeResponse(50)); - cache.getOrBuild("k", build); - cache.getOrBuild("k", build); + cache.getOrBuild(k("k"), build); + cache.getOrBuild(k("k"), build); expect(build).toHaveBeenCalledTimes(2); expect(cache.size).toBe(0); }); + it("drops a cached revision when its replacement is above the row cap", () => { + const cache = createThreadTimelineCache({ maxCacheableRows: 5 }); + + cache.getOrBuild(k("k", 1), () => makeResponse(3)); + expect(cache.size).toBe(1); + cache.getOrBuild(k("k", 2), () => makeResponse(50)); + expect(cache.size).toBe(0); + }); + it("evicts least-recently-used entries beyond maxEntries", () => { const cache = createThreadTimelineCache({ maxEntries: 2 }); const build = vi.fn(() => makeResponse(1)); - cache.getOrBuild("a", build); // [a] - cache.getOrBuild("b", build); // [a,b] - cache.getOrBuild("a", build); // touch a -> [b,a] - cache.getOrBuild("c", build); // evict b -> [a,c] + cache.getOrBuild(k("a"), build); // [a] + cache.getOrBuild(k("b"), build); // [a,b] + cache.getOrBuild(k("a"), build); // touch a -> [b,a] + cache.getOrBuild(k("c"), build); // evict b -> [a,c] expect(cache.size).toBe(2); const buildAgain = vi.fn(() => makeResponse(1)); - cache.getOrBuild("a", buildAgain); // still cached - cache.getOrBuild("b", buildAgain); // evicted -> rebuild + cache.getOrBuild(k("a"), buildAgain); // still cached + cache.getOrBuild(k("b"), buildAgain); // evicted -> rebuild expect(buildAgain).toHaveBeenCalledTimes(1); }); }); -describe("buildThreadTimelineCacheKey", () => { +describe("buildThreadTimelineParamsKey", () => { it("differs when any projection input differs", () => { - const base = buildThreadTimelineCacheKey(baseKeyArgs); + const base = buildThreadTimelineParamsKey(baseKeyArgs); const variants: ThreadTimelineCacheKeyArgs[] = [ - { ...baseKeyArgs, maxSeq: 11 }, { ...baseKeyArgs, status: "active" }, { ...baseKeyArgs, environmentId: "env_1" }, { ...baseKeyArgs, includeNestedRows: true }, @@ -128,12 +157,12 @@ describe("buildThreadTimelineCacheKey", () => { }, ]; for (const variant of variants) { - expect(buildThreadTimelineCacheKey(variant)).not.toBe(base); + expect(buildThreadTimelineParamsKey(variant)).not.toBe(base); } }); it("distinguishes older-page cursors", () => { - const cursorA = buildThreadTimelineCacheKey({ + const cursorA = buildThreadTimelineParamsKey({ ...baseKeyArgs, page: { kind: "older", @@ -141,7 +170,7 @@ describe("buildThreadTimelineCacheKey", () => { beforeCursor: { anchorSeq: 5, anchorId: "a5" }, }, }); - const cursorB = buildThreadTimelineCacheKey({ + const cursorB = buildThreadTimelineParamsKey({ ...baseKeyArgs, page: { kind: "older",