From ef5ebf7fad728d4b69c049a6eb8bba3c88773b18 Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Wed, 28 Oct 2026 20:38:07 +0900 Subject: [PATCH 1/8] refactor(runtime): keep a tag's marks apart from the tag view `tags.ts` held the tag view (`TagState`) and, below it, everything a tag's marks say of an entry: the marks known at no revision kept beside each other (`Marks`, `withMark`), and the validity a history or a set of marks gives an entry made at a moment. The file stood at 491 of the 500 lines the lint allows, with the view about to learn a window of recent invalidations. Those helpers move, unchanged, to `tag-marks.ts`, with `TagMark`, which `tags.ts` still exports. Nothing they do changes. --- packages/runtime/src/cache/tag-marks.ts | 150 +++++++++++++++++++++++ packages/runtime/src/cache/tags.ts | 156 +++--------------------- 2 files changed, 164 insertions(+), 142 deletions(-) create mode 100644 packages/runtime/src/cache/tag-marks.ts diff --git a/packages/runtime/src/cache/tag-marks.ts b/packages/runtime/src/cache/tag-marks.ts new file mode 100644 index 000000000..e6d91af88 --- /dev/null +++ b/packages/runtime/src/cache/tag-marks.ts @@ -0,0 +1,150 @@ +import type { TagHistoryMark, Validity } from '@stayingupwind/core/cache'; + +/** + * The marks of an invalidation this isolate holds of a tag, and what they say of an entry made at a + * moment: those known at a revision of the scope, kept as the tag's history, and those known at none, + * kept beside each other (`Marks`). What the tag view keeps of each tag is made of these + * (`TagState`, in `tags.ts`). + */ + +export interface TagMark { + readonly staleAt: number; + readonly hardExpireAt: number | null; + /** + * A revision of the scope the invalidation was recorded at or before, where one is known: the one + * the host says it recorded it at, or, for a mark the delta brought where it does not say, the + * revision that page of the delta was read at; for this isolate's own invalidation, the one the + * host answered it with (`confirmLocal`). What tells a view that held the invalidation from one + * that did not, whatever the clocks that stamped it said: a view brought up to that revision held + * it (`validityOf`). Unknown for an invalidation of this isolate's the host has yet to answer. + */ + readonly revision?: number | undefined; +} + +/** + * Marks of a tag known at no revision, in force here at once, of which none says all that another + * does of every entry (`withMark`): each is asked of, and the worst answer stands (`marksValidity`). + * Nothing tells which of them the host recorded last. Taken as one mark, the one stamped latest + * would hide a deadline another set, and taken by the order of the calls that made them, one the + * host recorded later than it was asked for would be lost. + * + * Kept the one made earliest first, each made later sets a later deadline too — else it would say + * all of the one before — so of those that reach an entry, the first sets the earliest deadline. + */ +export type Marks = readonly TagMark[]; + +/** + * How many marks of a tag known at no revision for good, or of calls counted together, are kept + * apart at the most (`withMark`): more than calls stamped in that order with deadlines in that order + * come to. Past that many, the two made earliest are taken as one that says all of both — the later + * moment, and the earlier deadline — which holds more back than either did, and never less. Nothing + * of either is taken out one by one. The marks of the calls still out counted one by one are kept + * apart however many they are, each taken out with its call (`without`): no more than the calls. + */ +const MAX_MARKS = 16; + +/** Whether `a` says all that `b` does of every entry: made no earlier, with a deadline no later. */ +export function covers(a: TagMark, b: TagMark): boolean { + return a.staleAt >= b.staleAt && deadlineOf(a) <= deadlineOf(b); +} + +/** The moment a mark expires what it reaches, where it sets one; never, where it does not. */ +function deadlineOf(mark: TagMark): number { + return mark.hardExpireAt ?? Infinity; +} + +/** + * Where in `marks` the first of them is that reaches an entry made at `timestamp` (`Marks`): of those + * that do, the one with the earliest deadline. Past the last, where none does. + */ +export function firstReaching(marks: Marks, timestamp: number): number { + let low = 0; + let high = marks.length; + while (low < high) { + const middle = Math.floor((low + high) / 2); + if ((marks[middle]?.staleAt ?? timestamp) < timestamp) { + low = middle + 1; + } else { + high = middle; + } + } + return low; +} + +/** + * Put `mark` in force beside `marks`, where they stand (`Marks`): none kept that another says all of + * (`covers`). Past `most`, the two made earliest are taken as one. + */ +export function addMark(marks: TagMark[], mark: TagMark, most: number): void { + const at = firstReaching(marks, mark.staleAt); + const next = marks[at]; + if (next !== undefined && deadlineOf(next) <= deadlineOf(mark)) { + return; + } + // Those `mark` says all of: the last run of those made before it whose deadlines come no earlier + // than its own, and one made at its very moment. + let from = at; + while (from > 0 && deadlineOf(marks[from - 1] ?? mark) >= deadlineOf(mark)) { + from -= 1; + } + marks.splice(from, (next?.staleAt === mark.staleAt ? at + 1 : at) - from, mark); + const [first, second] = marks; + if (first !== undefined && second !== undefined && marks.length > most) { + marks.splice(0, 2, { staleAt: second.staleAt, hardExpireAt: first.hardExpireAt }); + } +} + +/** `marks` with `mark` in force beside them (`addMark`), past `MAX_MARKS` the earliest taken together. */ +export function withMark(marks: Marks | undefined, mark: TagMark): Marks { + const kept = [...(marks ?? [])]; + addMark(kept, mark, MAX_MARKS); + return kept; +} + +/** Whether any of `marks` reaches an entry made at `timestamp`: the one made last does. */ +export function reachesAny(marks: Marks | undefined, timestamp: number): boolean { + return (marks?.at(-1)?.staleAt ?? -Infinity) >= timestamp; +} + +/** + * What `marks` say of an entry made at `timestamp`, judged at `now`: the worst any of them does, which + * the first of them that reaches it does (`Marks`). + */ +export function marksValidity(marks: Marks | undefined, timestamp: number, now: number): Validity { + return marks === undefined + ? 'fresh' + : markValidity(marks[firstReaching(marks, timestamp)], timestamp, now); +} + +/** Earlier deadlines apply only to values that did not already hold their invalidation. */ +export function historyValidity( + history: readonly TagHistoryMark[] | undefined, + timestamp: number, + now: number, + held: number | undefined, +): Validity { + if (history === undefined) return 'fresh'; + let validity: Validity = 'fresh'; + for (const mark of history) { + if (held !== undefined && mark.revision <= held) continue; + validity = worst(validity, markValidity(mark, held === undefined ? timestamp : -Infinity, now)); + if (validity === 'expired') return validity; + } + return validity; +} + +/** The worse of two answers of what a tag says of an entry. */ +export function worst(a: Validity, b: Validity): Validity { + if (a === 'expired' || b === 'expired') { + return 'expired'; + } + return a === 'stale' || b === 'stale' ? 'stale' : 'fresh'; +} + +/** What one mark says of an entry made at `timestamp`, judged at `now` (`validityOf`). */ +function markValidity(mark: TagMark | undefined, timestamp: number, now: number): Validity { + if (mark === undefined || mark.staleAt < timestamp) { + return 'fresh'; + } + return mark.hardExpireAt !== null && mark.hardExpireAt <= now ? 'expired' : 'stale'; +} diff --git a/packages/runtime/src/cache/tags.ts b/packages/runtime/src/cache/tags.ts index 9a8dc3d64..f6ef9e33c 100644 --- a/packages/runtime/src/cache/tags.ts +++ b/packages/runtime/src/cache/tags.ts @@ -9,6 +9,20 @@ import { Lru } from '@stayingupwind/core/util'; import type { CacheHost, InvalidateOutcome, TagInvalidation } from './host.ts'; import { TagLosses } from './tag-losses.ts'; +import { + addMark, + covers, + firstReaching, + historyValidity, + type Marks, + marksValidity, + reachesAny, + type TagMark, + withMark, + worst, +} from './tag-marks.ts'; + +export type { TagMark } from './tag-marks.ts'; /** * What this isolate knows of the scope's invalidations: the tags the host has recorded an @@ -25,20 +39,6 @@ import { TagLosses } from './tag-losses.ts'; */ export type TagHost = Pick; -export interface TagMark { - readonly staleAt: number; - readonly hardExpireAt: number | null; - /** - * A revision of the scope the invalidation was recorded at or before, where one is known: the one - * the host says it recorded it at, or, for a mark the delta brought where it does not say, the - * revision that page of the delta was read at; for this isolate's own invalidation, the one the - * host answered it with (`confirmLocal`). What tells a view that held the invalidation from one - * that did not, whatever the clocks that stamped it said: a view brought up to that revision held - * it (`validityOf`). Unknown for an invalidation of this isolate's the host has yet to answer. - */ - readonly revision?: number | undefined; -} - export interface TagStateOptions { /** How long a synced view is trusted before the delta is asked for again. */ readonly holdMs: number; @@ -91,28 +91,6 @@ const MAX_TAGS_HELD = 4096; */ const MAX_APPLIED_MARKS = 4096; -/** - * Marks of a tag known at no revision, in force here at once, of which none says all that another - * does of every entry (`withMark`): each is asked of, and the worst answer stands (`marksValidity`). - * Nothing tells which of them the host recorded last. Taken as one mark, the one stamped latest - * would hide a deadline another set, and taken by the order of the calls that made them, one the - * host recorded later than it was asked for would be lost. - * - * Kept the one made earliest first, each made later sets a later deadline too — else it would say - * all of the one before — so of those that reach an entry, the first sets the earliest deadline. - */ -type Marks = readonly TagMark[]; - -/** - * How many marks of a tag known at no revision for good, or of calls counted together, are kept - * apart at the most (`withMark`): more than calls stamped in that order with deadlines in that order - * come to. Past that many, the two made earliest are taken as one that says all of both — the later - * moment, and the earlier deadline — which holds more back than either did, and never less. Nothing - * of either is taken out one by one. The marks of the calls still out counted one by one are kept - * apart however many they are, each taken out with its call (`without`): no more than the calls. - */ -const MAX_MARKS = 16; - /** * What this isolate knows of one tag, and when the host last said so (`checkedAt`): its latest * invalidation revision (`knownRevision`), and the bounded deadline history known at @@ -769,112 +747,6 @@ function overflowWithout( return overflow.count <= 1 ? undefined : { ...overflow, count: overflow.count - 1 }; } -/** Whether `a` says all that `b` does of every entry: made no earlier, with a deadline no later. */ -function covers(a: TagMark, b: TagMark): boolean { - return a.staleAt >= b.staleAt && deadlineOf(a) <= deadlineOf(b); -} - -/** The moment a mark expires what it reaches, where it sets one; never, where it does not. */ -function deadlineOf(mark: TagMark): number { - return mark.hardExpireAt ?? Infinity; -} - -/** - * Where in `marks` the first of them is that reaches an entry made at `timestamp` (`Marks`): of those - * that do, the one with the earliest deadline. Past the last, where none does. - */ -function firstReaching(marks: Marks, timestamp: number): number { - let low = 0; - let high = marks.length; - while (low < high) { - const middle = Math.floor((low + high) / 2); - if ((marks[middle]?.staleAt ?? timestamp) < timestamp) { - low = middle + 1; - } else { - high = middle; - } - } - return low; -} - -/** - * Put `mark` in force beside `marks`, where they stand (`Marks`): none kept that another says all of - * (`covers`). Past `most`, the two made earliest are taken as one. - */ -function addMark(marks: TagMark[], mark: TagMark, most: number): void { - const at = firstReaching(marks, mark.staleAt); - const next = marks[at]; - if (next !== undefined && deadlineOf(next) <= deadlineOf(mark)) { - return; - } - // Those `mark` says all of: the last run of those made before it whose deadlines come no earlier - // than its own, and one made at its very moment. - let from = at; - while (from > 0 && deadlineOf(marks[from - 1] ?? mark) >= deadlineOf(mark)) { - from -= 1; - } - marks.splice(from, (next?.staleAt === mark.staleAt ? at + 1 : at) - from, mark); - const [first, second] = marks; - if (first !== undefined && second !== undefined && marks.length > most) { - marks.splice(0, 2, { staleAt: second.staleAt, hardExpireAt: first.hardExpireAt }); - } -} - -/** `marks` with `mark` in force beside them (`addMark`), past `MAX_MARKS` the earliest taken together. */ -function withMark(marks: Marks | undefined, mark: TagMark): Marks { - const kept = [...(marks ?? [])]; - addMark(kept, mark, MAX_MARKS); - return kept; -} - -/** Whether any of `marks` reaches an entry made at `timestamp`: the one made last does. */ -function reachesAny(marks: Marks | undefined, timestamp: number): boolean { - return (marks?.at(-1)?.staleAt ?? -Infinity) >= timestamp; -} - -/** - * What `marks` say of an entry made at `timestamp`, judged at `now`: the worst any of them does, which - * the first of them that reaches it does (`Marks`). - */ -function marksValidity(marks: Marks | undefined, timestamp: number, now: number): Validity { - return marks === undefined - ? 'fresh' - : markValidity(marks[firstReaching(marks, timestamp)], timestamp, now); -} - -/** Earlier deadlines apply only to values that did not already hold their invalidation. */ -function historyValidity( - history: readonly TagHistoryMark[] | undefined, - timestamp: number, - now: number, - held: number | undefined, -): Validity { - if (history === undefined) return 'fresh'; - let validity: Validity = 'fresh'; - for (const mark of history) { - if (held !== undefined && mark.revision <= held) continue; - validity = worst(validity, markValidity(mark, held === undefined ? timestamp : -Infinity, now)); - if (validity === 'expired') return validity; - } - return validity; -} - -/** The worse of two answers of what a tag says of an entry. */ -function worst(a: Validity, b: Validity): Validity { - if (a === 'expired' || b === 'expired') { - return 'expired'; - } - return a === 'stale' || b === 'stale' ? 'stale' : 'fresh'; -} - -/** What one mark says of an entry made at `timestamp`, judged at `now` (`validityOf`). */ -function markValidity(mark: TagMark | undefined, timestamp: number, now: number): Validity { - if (mark === undefined || mark.staleAt < timestamp) { - return 'fresh'; - } - return mark.hardExpireAt !== null && mark.hardExpireAt <= now ? 'expired' : 'stale'; -} - /** The worse of what the tags say and what the record itself carries. */ export function recordValidity( tags: TagState, From acfaae478fb459db9fc27de57acf11ce57e31957 Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Wed, 28 Oct 2026 20:39:43 +0900 Subject: [PATCH 2/8] refactor(cache): read a record whole beside the read budget The Function's gateway client read a delivery record whole through a private method of its own (`#readWhole`), which used nothing of the client: it holds room in the isolate's read budget for the answer's length and reads the body once it has it. It moves, unchanged, beside the budget's other whole reads in `read-budget.ts` (`readWhole`), so `gateway.ts`, at 495 of the 500 lines the lint allows, has room for a tag read of its own. --- packages/cache/src/function/gateway.ts | 30 +++------------------- packages/cache/src/function/read-budget.ts | 24 +++++++++++++++++ 2 files changed, 27 insertions(+), 27 deletions(-) diff --git a/packages/cache/src/function/gateway.ts b/packages/cache/src/function/gateway.ts index 4937ef9a0..651da5874 100644 --- a/packages/cache/src/function/gateway.ts +++ b/packages/cache/src/function/gateway.ts @@ -61,8 +61,8 @@ import { import { ArtifactRoom, functionReadBudget, - projectedLength, readArtifactWhole, + readWhole, saidLength, type ReadBudget, } from './read-budget.ts'; @@ -372,30 +372,6 @@ export class Gateway { }); } - /** - * The answer's bytes, read whole once the isolate has room for them (`ReadBudget`): the room the - * artifact's copy holds, where it does, and room of the answer's own otherwise. One given up - * while it waits for room lets go of its body unread. - */ - async #readWhole( - response: Response, - signal: AbortSignal, - room: ArtifactRoom, - ): Promise { - let release: () => void; - try { - release = await room.hold(projectedLength(response), signal); - } catch (error) { - releaseStream(response.body, 'an artifact read given up before its body'); - throw error; - } - try { - return new Uint8Array(await response.arrayBuffer()); - } finally { - release(); - } - } - /** Both tag metadata reads cross to the authority, regardless of a nearby read gateway. */ #authorityTagRead( path: string, @@ -541,7 +517,7 @@ export class Gateway { /** * The delivery record of an entry, as the edge reads it; `undefined` when there is none. Given up * past `RECORD_READ_DEADLINE_MS`, and called off once `signal` aborts: either, its body and all. - * Read whole in room of its own, as an artifact is (`#readWhole`), where the gateway says how long + * Read whole in room of its own, as an artifact is (`readWhole`), where the gateway says how long * it is, which both gateways do; an answer that says nothing, a gateway's from before they said * it, is read as it always was, in no room. */ @@ -563,7 +539,7 @@ export class Gateway { if (saidLength(response) === undefined) { return new Uint8Array(await response.arrayBuffer()); } - return this.#readWhole(response, asked, new ArtifactRoom(this.#reading)); + return readWhole(response, asked, new ArtifactRoom(this.#reading)); }); } diff --git a/packages/cache/src/function/read-budget.ts b/packages/cache/src/function/read-budget.ts index 664668354..94af7a633 100644 --- a/packages/cache/src/function/read-budget.ts +++ b/packages/cache/src/function/read-budget.ts @@ -250,6 +250,30 @@ export function projectedLength(response: Response): number { return saidLength(response) ?? FUNCTION_READ_BYTES; } +/** + * The answer's bytes, read whole once the isolate has room for them (`ReadBudget`): the room the + * artifact's copy holds, where it does, and room of the answer's own otherwise. One given up + * while it waits for room lets go of its body unread. + */ +export async function readWhole( + response: Response, + signal: AbortSignal, + room: ArtifactRoom, +): Promise { + let release: () => void; + try { + release = await room.hold(projectedLength(response), signal); + } catch (error) { + releaseStream(response.body, 'an artifact read given up before its body'); + throw error; + } + try { + return new Uint8Array(await response.arrayBuffer()); + } finally { + release(); + } +} + /** * The Function's artifact buffer, at most its shared 32 MiB room. Larger seeded or configured * uploads still deploy and stream elsewhere; this cache read misses. A native body takes one From 96f60b357d30b76bd665056b828e97ed716ceed1 Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Wed, 28 Oct 2026 20:44:31 +0900 Subject: [PATCH 3/8] perf(cache): page the tag delta a thousand tags at a time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fresh isolate brings its view of the tags up by paging through the scope's whole delta, oldest first, one round trip to the authority a page — and a scope keeps a row for every tag it ever invalidated. The Function's client asked for no limit, so each page held the gateway's default of 500 tags, though a page may hold 1000 (`MAX_DELTA_PAGE_TAGS`) and the gateway already holds a larger ask to that. What the authority reads for each row is one lookup of the tag's history. The client now asks each page for 1000 tags: half the round trips for every full pull, which a regeneration waits for before it renders, and which a request behind a recent invalidation waits for up to a second. --- packages/cache/src/function/gateway.ts | 11 ++++++-- .../test/tag-authority-deadlines.test.ts | 27 ++++++++++++++++++- 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/packages/cache/src/function/gateway.ts b/packages/cache/src/function/gateway.ts index 651da5874..96b9a726b 100644 --- a/packages/cache/src/function/gateway.ts +++ b/packages/cache/src/function/gateway.ts @@ -31,6 +31,7 @@ import { type InvalidateRequest, type InvalidateResponse, invalidateResponseSchema, + MAX_DELTA_PAGE_TAGS, NO_SUCH_ENDPOINT, type StartAttemptRequest, type StartAttemptResponse, @@ -638,9 +639,15 @@ export class Gateway { ); } - /** One delta page, including its body, read within the authority deadline. */ + /** + * One delta page, including its body, read within the authority deadline: as many tags as a page + * may hold (`MAX_DELTA_PAGE_TAGS`). A fresh isolate pages through the scope's whole history, one + * round trip a page, and what it reads of each row is one lookup of the authority's: the gateway's + * own 500 a page took twice the round trips for the same history. + */ tagsDelta(since: number): Promise { - return this.#authorityTagRead(`/tags/delta?since=${since}`, async (response, signal) => { + const path = `/tags/delta?since=${since}&limit=${MAX_DELTA_PAGE_TAGS}`; + return this.#authorityTagRead(path, async (response, signal) => { const delta = await this.#result(response, tagsDeltaResponseSchema, undefined, signal); // A repeated page would leave its reader asking for the same invalidations forever. if (!deltaMovesOn(since, delta)) { diff --git a/packages/cache/test/tag-authority-deadlines.test.ts b/packages/cache/test/tag-authority-deadlines.test.ts index d28b17aa8..9b6d7c29b 100644 --- a/packages/cache/test/tag-authority-deadlines.test.ts +++ b/packages/cache/test/tag-authority-deadlines.test.ts @@ -1,6 +1,10 @@ import { tagKindOf } from '@stayingupwind/core/cache'; import { isDeadlineError } from '@stayingupwind/core/util'; -import type { TagRevision, TagsDeltaResponse } from '@stayingupwind/host/cache'; +import { + MAX_DELTA_PAGE_TAGS, + type TagRevision, + type TagsDeltaResponse, +} from '@stayingupwind/host/cache'; import type { FetchLike } from '@stayingupwind/runtime/cache-host'; import { afterEach, describe, expect, it, vi } from 'vitest'; @@ -239,3 +243,24 @@ describe('an authority delta cursor', () => { expect(send).toHaveBeenCalledTimes(1); }); }); + +describe('what a Function asks of the authority’s tags', () => { + /** + * A fresh isolate pages through the scope's whole history, a round trip a page: it asks each for + * as many tags as a page may hold, where the gateway's own default is half of it. + */ + it('asks each page of the delta for as many tags as a page may hold', async () => { + const send = vi.fn().mockResolvedValue(Response.json({ ok: true, result: DELTA })); + const { subject } = gateway(send); + + await subject.tagsDelta(4); + + const asked = new URL(requiredRequest(send).url); + + expect(asked.pathname.endsWith('/tags/delta')).toBe(true); + expect(Object.fromEntries(asked.searchParams)).toStrictEqual({ + since: '4', + limit: String(MAX_DELTA_PAGE_TAGS), + }); + }); +}); From 92205e19d5befc1353bdcdabdb2335e266725b05 Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Wed, 28 Oct 2026 20:49:50 +0900 Subject: [PATCH 4/8] feat(cache): read a scope's recent invalidations in one round trip MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A Function's isolate whose view of the tags is behind its scope can only bring it up through the delta: every tag the scope ever invalidated, oldest first, a round trip a page. The invalidations a request needs the view to hold are the newest, and they come last. The authority already answers the window an edge is told of (`recentDelta`): the invalidations past a revision that took effect from a moment on, each as it was recorded, the newest so many, truncated where the window held more, at the scope's revision. The gateway now answers it to a Function too, as a route of its own (`GET /tags/recent?since&from&limit`), with a shape of its own in the contract (`tagsRecentResponseSchema`): no more tags than a page, none past the revision it was read at, and no rule that a truncated answer ends at its last tag — a window cut at its oldest end is no page of the delta, and the delta's shape refuses one, or, where its newest row stood at the scope's revision, would take it for a page boundary. The Function's client reads it as the delta is read, within the same authority deadline (`Gateway.tagsRecent`); its host leaves platform tags out as `tagsDelta` does (`GatewayHost.tagsRecent`); and the runtime's host contract gains it as an optional call (`CacheHost.tagsRecent`). A gateway from before the route answers 404, which the client reports as `unsupported`, sharing what told a batch upload so. The native cache is left without it: no request there carries where its scope stands, so nothing would ask. --- packages/cache/src/function/cache-host.ts | 11 ++ packages/cache/src/function/gateway.ts | 53 ++++++-- packages/cache/src/gateway/routes/tags.ts | 34 ++++- .../test/tag-authority-deadlines.test.ts | 118 ++++++++++++++++-- packages/cache/test/tags-recent-route.test.ts | 112 +++++++++++++++++ packages/host/src/cache/gateway-contract.ts | 49 +++++++- packages/host/src/cache/index.ts | 2 + .../host/test/cache-data-contract.test.ts | 54 ++++++++ packages/runtime/src/cache/host.ts | 23 +++- 9 files changed, 428 insertions(+), 28 deletions(-) create mode 100644 packages/cache/test/tags-recent-route.test.ts diff --git a/packages/cache/src/function/cache-host.ts b/packages/cache/src/function/cache-host.ts index 978f79946..206881950 100644 --- a/packages/cache/src/function/cache-host.ts +++ b/packages/cache/src/function/cache-host.ts @@ -42,6 +42,7 @@ import { nativeFetch, type TagDelta, type TagInvalidations, + type TagWindow, type UploadedArtifact, } from '@stayingupwind/runtime/cache-host'; @@ -582,6 +583,16 @@ class GatewayHost implements CacheHost { return { ...delta, tags: delta.tags.filter((tag) => tag.kind === tagKindOf(tag.value)) }; } + /** + * The newest invalidations past `since` that took effect at `from` or later, of Next.js's tags + * alone, as `tagsDelta` leaves them: a platform-delivery tag sharing a tag's value expires nothing + * of the data cache. + */ + async tagsRecent(since: number, from: number): Promise { + const recent = await this.#gateway.tagsRecent(since, from); + return { ...recent, tags: recent.tags.filter((tag) => tag.kind === tagKindOf(tag.value)) }; + } + /** Where the scope's tags stand, as the data center's copy says (`ScopeBeacon`), if one is kept. */ async scopeRevision(): Promise { return await this.#beacon?.read(); diff --git a/packages/cache/src/function/gateway.ts b/packages/cache/src/function/gateway.ts index 96b9a726b..f5588c2d6 100644 --- a/packages/cache/src/function/gateway.ts +++ b/packages/cache/src/function/gateway.ts @@ -40,6 +40,8 @@ import { tagRevisionSchema, type TagsDeltaResponse, tagsDeltaResponseSchema, + type TagsRecentResponse, + tagsRecentResponseSchema, uploadArtifactsResponseSchema, } from '@stayingupwind/host/cache'; import { @@ -373,6 +375,27 @@ export class Gateway { }); } + /** + * What a call that failed answered, as the error it stands for — or, where it is a gateway from + * before the route (`what`), `unsupported`, as `CacheHost` says: a 404 that says there is no such + * route (`NO_SUCH_ENDPOINT`), or one with no envelope from whatever stands in front of it. Read by + * its status before its body, which such an answer may not have as JSON; a `not_found` of what the + * call names is the call's to fail on. + */ + async #unlessNoSuchRoute( + response: Response, + what: string, + signal?: AbortSignal, + ): Promise { + const failure = await this.#failureOf(response, signal); + const noSuchRoute = + response.status === HTTP_NOT_FOUND && + (failure.code === 'internal' || failure.message === NO_SUCH_ENDPOINT); + return noSuchRoute + ? new CacheHostError(what, { code: 'unsupported', status: response.status }) + : failure; + } + /** Both tag metadata reads cross to the authority, regardless of a nearby read gateway. */ #authorityTagRead( path: string, @@ -457,21 +480,10 @@ export class Gateway { headers: this.#headers({ 'content-type': JSON_TYPE }), body: JSON.stringify({ fencingToken: first.fencingToken, artifacts }), }); - // Its status before its body, which a gateway's answer may not have as JSON. A gateway from - // before the call has no such route — its answer to one it does not have, or a 404 with no - // envelope from whatever stands in front of it — and the runtime is told so as the contract + // A gateway from before the call has no such route, and the runtime is told so as the contract // says (`unsupported`), where a `not_found` of the attempt is the attempt's to fail on. if (!response.ok) { - const failure = await this.#failureOf(response); - const noSuchRoute = - response.status === HTTP_NOT_FOUND && - (failure.code === 'internal' || failure.message === NO_SUCH_ENDPOINT); - throw noSuchRoute - ? new CacheHostError('the gateway takes no batch of uploads', { - code: 'unsupported', - status: response.status, - }) - : failure; + throw await this.#unlessNoSuchRoute(response, 'the gateway takes no batch of uploads'); } const uploaded = (await this.#result(response, uploadArtifactsResponseSchema)).artifacts; if ( @@ -660,6 +672,21 @@ export class Gateway { }); } + /** + * The newest invalidations past `since` that took effect at `from` or later (`/tags/recent`), as + * many as a page may hold, read within the authority deadline as the delta is; `unsupported` from + * a gateway from before the route (`#unlessNoSuchRoute`). + */ + tagsRecent(since: number, from: number): Promise { + const path = `/tags/recent?since=${since}&from=${from}&limit=${MAX_DELTA_PAGE_TAGS}`; + return this.#authorityTagRead(path, async (response, signal) => { + if (!response.ok) { + throw await this.#unlessNoSuchRoute(response, 'the gateway reads no recent tags', signal); + } + return this.#result(response, tagsRecentResponseSchema, undefined, signal); + }); + } + invalidate(request: InvalidateRequest): Promise { return this.#json('POST', '/invalidations', request, invalidateResponseSchema); } diff --git a/packages/cache/src/gateway/routes/tags.ts b/packages/cache/src/gateway/routes/tags.ts index f323f3271..29b34a6f4 100644 --- a/packages/cache/src/gateway/routes/tags.ts +++ b/packages/cache/src/gateway/routes/tags.ts @@ -12,7 +12,10 @@ import { readRequest, TAGS_BODY_BYTES } from '../body.ts'; import { readTagRecords } from '../read-tags.ts'; import { ok } from '../respond.ts'; -/** Invalidations, and the delta a Worker syncs its view of the tags with. */ +/** + * Invalidations, the delta a Worker syncs its view of the tags with, and the window of recent ones + * it catches up with first. + */ const DEFAULT_DELTA_LIMIT = 500; @@ -21,6 +24,11 @@ function nonNegativeInt(value: string | undefined, fallback: number): number { return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : fallback; } +/** The `limit` a read of the tags asked for, no more than a page may hold (`MAX_DELTA_PAGE_TAGS`). */ +function limitOf(value: string | undefined): number { + return Math.max(Math.min(nonNegativeInt(value, DEFAULT_DELTA_LIMIT), MAX_DELTA_PAGE_TAGS), 1); +} + export const tagRoutes = new Hono() // Not `zValidator`: it parses whatever arrives before anything has measured it. .post('/tags/get', async (c) => { @@ -41,16 +49,30 @@ export const tagRoutes = new Hono() .get('/tags/delta', async (c) => { const since = nonNegativeInt(c.req.query('since'), 0); // No more than a page may hold (`MAX_DELTA_PAGE_TAGS`), which its reader holds it to. - const limit = Math.min( - nonNegativeInt(c.req.query('limit'), DEFAULT_DELTA_LIMIT), - MAX_DELTA_PAGE_TAGS, - ); + const limit = limitOf(c.req.query('limit')); const [delta, retired] = await Promise.all([ - c.get('scope').tagsDelta(since, Math.max(limit, 1)), + c.get('scope').tagsDelta(since, limit), retiredScope(c), ]); return retired ? refuseRetired(c) : ok(c, delta); }) + // The newest invalidations past `since` that took effect from `from` on (`recentDelta`, + // `tagsRecentResponseSchema`): what an isolate behind its scope may find its reads behind on, in + // one round trip however long the scope's history is. Its own route, never the delta's: cut at its + // oldest end where it holds more than its limit, it is no page a reader of the delta may stand at. + // A `from` left out asks from the beginning, which leaves out nothing. + .get('/tags/recent', async (c) => { + const since = nonNegativeInt(c.req.query('since'), 0); + const from = nonNegativeInt(c.req.query('from'), 0); + const limit = limitOf(c.req.query('limit')); + const [recent, retired] = await Promise.all([ + c.get('scope').tagsDelta(since, limit, from), + retiredScope(c), + ]); + return retired + ? refuseRetired(c) + : ok(c, { revision: recent.revision, truncated: recent.truncated, tags: recent.tags }); + }) .post('/invalidations', async (c) => { const request = await readRequest(c, invalidateRequestSchema, TAGS_BODY_BYTES); if (request instanceof Response) { diff --git a/packages/cache/test/tag-authority-deadlines.test.ts b/packages/cache/test/tag-authority-deadlines.test.ts index 9b6d7c29b..1cadc4136 100644 --- a/packages/cache/test/tag-authority-deadlines.test.ts +++ b/packages/cache/test/tag-authority-deadlines.test.ts @@ -1,17 +1,20 @@ -import { tagKindOf } from '@stayingupwind/core/cache'; +import { type TagKind, tagKindOf } from '@stayingupwind/core/cache'; import { isDeadlineError } from '@stayingupwind/core/util'; import { MAX_DELTA_PAGE_TAGS, + NO_SUCH_ENDPOINT, type TagRevision, type TagsDeltaResponse, + type TagsRecentResponse, } from '@stayingupwind/host/cache'; import type { FetchLike } from '@stayingupwind/runtime/cache-host'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { cacheHostFor } from '../src/function/cache-host.ts'; import { Gateway } from '../src/function/gateway.ts'; -type Method = 'revision' | 'delta'; -const METHODS: readonly Method[] = ['revision', 'delta']; +type Method = 'revision' | 'delta' | 'recent'; +const METHODS: readonly Method[] = ['revision', 'delta', 'recent']; const REVISION: TagRevision = { revision: 5, invalidatedAt: 123 }; const DELTA: TagsDeltaResponse = { revision: 5, @@ -20,13 +23,21 @@ const DELTA: TagsDeltaResponse = { tags: [], latest: REVISION, }; +const RECENT: TagsRecentResponse = { revision: 5, truncated: false, tags: [] }; +/** The moment a window of recent invalidations is asked from. */ +const FROM = 1_800_000_000_000; -function resultOf(method: Method): TagRevision | TagsDeltaResponse { - return method === 'revision' ? REVISION : DELTA; +type Answer = TagRevision | TagsDeltaResponse | TagsRecentResponse; + +function resultOf(method: Method): Answer { + return { revision: REVISION, delta: DELTA, recent: RECENT }[method]; } -function read(subject: Gateway, method: Method): Promise { - return method === 'revision' ? subject.tagRevision() : subject.tagsDelta(4); +function read(subject: Gateway, method: Method): Promise { + if (method === 'revision') { + return subject.tagRevision(); + } + return method === 'delta' ? subject.tagsDelta(4) : subject.tagsRecent(4, FROM); } function gateway(send: FetchLike) { @@ -244,6 +255,19 @@ describe('an authority delta cursor', () => { }); }); +/** A tag's record as the authority answers it, of the kind its value names unless said. */ +function recordOf(value: string, revision: number, kind: TagKind = tagKindOf(value)) { + return { + tagId: `tag-${value}`, + kind, + value, + revision, + staleAt: 123, + hardExpireAt: null, + meaning: 'immediate', + } as const; +} + describe('what a Function asks of the authority’s tags', () => { /** * A fresh isolate pages through the scope's whole history, a round trip a page: it asks each for @@ -263,4 +287,84 @@ describe('what a Function asks of the authority’s tags', () => { limit: String(MAX_DELTA_PAGE_TAGS), }); }); + + it('asks a window of recent invalidations from a revision and a moment, as many as a page holds', async () => { + const send = vi.fn().mockResolvedValue(Response.json({ ok: true, result: RECENT })); + const { subject } = gateway(send); + + await subject.tagsRecent(4, FROM); + + const asked = new URL(requiredRequest(send).url); + + expect(asked.origin).toBe('https://authority.invalid'); + expect(asked.pathname.endsWith('/tags/recent')).toBe(true); + expect(Object.fromEntries(asked.searchParams)).toStrictEqual({ + since: '4', + from: String(FROM), + limit: String(MAX_DELTA_PAGE_TAGS), + }); + }); + + /** + * A window cut at its oldest end stands at the scope's revision, past the newest invalidation it + * holds: what no page of the delta may do, and what a window is. + */ + it('takes a truncated window that stands past the newest invalidation it holds', async () => { + const window = { revision: 9, truncated: true, tags: [recordOf('posts', 8)] }; + const send = vi.fn().mockResolvedValue(Response.json({ ok: true, result: window })); + const { subject } = gateway(send); + + await expect(subject.tagsRecent(4, FROM)).resolves.toStrictEqual(window); + }); + + /** + * A gateway from before the window answers a route it does not have, or whatever stands in front + * of it answers 404 with no envelope: the runtime is told the call is unsupported, and catches up + * through the delta alone. A 404 of something else is that refusal. + */ + it.each([ + { + answer: { ok: false, error: { code: 'not_found', message: NO_SUCH_ENDPOINT } }, + code: 'unsupported', + }, + { answer: 'no such route', code: 'unsupported' }, + { + answer: { ok: false, error: { code: 'not_found', message: 'no such scope' } }, + code: 'not_found', + }, + ])('says a window a gateway answers 404 to is $code', async ({ answer, code }) => { + const response = + typeof answer === 'string' + ? new Response(answer, { status: 404 }) + : Response.json(answer, { status: 404 }); + const send = vi.fn().mockResolvedValue(response); + const { subject } = gateway(send); + + await expect(subject.tagsRecent(4, FROM)).rejects.toMatchObject({ code, status: 404 }); + }); + + /** The runtime's tags are Next.js's: a platform-delivery tag sharing a value expires nothing. */ + it('leaves a platform tag out of the window a Function’s host answers', async () => { + const window = { + revision: 9, + truncated: false, + tags: [recordOf('products', 7, 'platform-delivery'), recordOf('products', 8)], + }; + const send = vi.fn().mockResolvedValue(Response.json({ ok: true, result: window })); + const binding = cacheHostFor({ + copyCache: 'tag-window-copies', + copyKeyOrigin: 'https://copies.invalid', + })({ + env: { + ARKOR_CACHE_GATEWAY_URL: 'https://authority.invalid', + ARKOR_CACHE_TOKEN: 'scope-token', + NEXT_DEPLOYMENT_ID: 'scope_tag_window', + }, + fetchImpl: send, + }); + + const recent = await binding?.host.tagsRecent?.(0, FROM); + + expect(recent).toStrictEqual({ ...window, tags: [recordOf('products', 8)] }); + }); }); diff --git a/packages/cache/test/tags-recent-route.test.ts b/packages/cache/test/tags-recent-route.test.ts new file mode 100644 index 000000000..033b69f0e --- /dev/null +++ b/packages/cache/test/tags-recent-route.test.ts @@ -0,0 +1,112 @@ +import { DEPLOYMENT_ID_PREFIX } from '@stayingupwind/core/bundle'; +import { createId } from '@stayingupwind/core/util'; +import { + type InvalidateResponse, + type TagsRecentResponse, + tagsRecentResponseSchema, +} from '@stayingupwind/host/cache'; +import { exports } from 'cloudflare:workers'; +import { describe, expect, it } from 'vitest'; + +/** + * The window of a scope's recent invalidations a Function's isolate catches up with + * (`GET /tags/recent`): the newest past a revision that took effect from a moment on, as many as it + * asked for, oldest first, at the scope's revision, and truncated where the window held more. + */ + +interface Scope { + readonly scopeId: string; + readonly token: string; +} + +/** A scope of its own, seeded through the cache's control entrypoint. */ +async function scope(): Promise { + const scopeId = createId(DEPLOYMENT_ID_PREFIX); + const seeded = await exports.CacheControl.seedScope({ + scope: { + scopeId, + organizationId: 'org_site', + projectId: 'site', + environmentId: 'production', + deploymentId: scopeId, + keySchemaVersion: 1, + buildId: 'build-site', + }, + }); + if (seeded.kind === 'conflict') { + throw new Error('the scope was not seeded'); + } + return { scopeId, token: seeded.token }; +} + +/** The result of a gateway call of `at`'s, which must have succeeded. */ +async function call(at: Scope, method: string, path: string, body?: unknown): Promise { + const response = await exports.default.fetch( + new Request(`https://cache.invalid/v1/scopes/${at.scopeId}${path}`, { + method, + headers: { authorization: `Bearer ${at.token}`, 'content-type': 'application/json' }, + ...(body !== undefined && { body: JSON.stringify(body) }), + }), + ); + const envelope = await response.json<{ readonly ok: boolean; readonly result?: T }>(); + if (!envelope.ok || envelope.result === undefined) { + throw new Error(`the gateway answered ${String(response.status)}`); + } + return envelope.result; +} + +/** An invalidation of `tag` at once, as a Function records one: the revision it was recorded at. */ +async function invalidate(at: Scope, tag: string): Promise { + const recorded = await call(at, 'POST', '/invalidations', { + tags: [{ value: tag }], + cause: { source: 'app-worker', api: 'updateTag' }, + }); + return recorded.revision; +} + +/** The window `query` asks for, as its reader checks it. */ +async function recent(at: Scope, query: string): Promise { + return tagsRecentResponseSchema.parse(await call(at, 'GET', `/tags/recent?${query}`)); +} + +describe('the window of a scope’s recent invalidations', () => { + it('answers the newest it was asked for, oldest first, truncated where there were more', async () => { + const at = await scope(); + const first = await invalidate(at, 'a'); + await invalidate(at, 'b'); + const last = await invalidate(at, 'c'); + + const window = await recent(at, 'since=0&from=0&limit=2'); + + expect(window.truncated).toBe(true); + expect(window.tags.map((tag) => tag.value)).toStrictEqual(['b', 'c']); + expect(window.revision).toBeGreaterThanOrEqual(last); + + // Whole past where it was asked from, and never past the revision it was read at. + const whole = await recent(at, `since=${String(first)}&from=0`); + + expect(whole.truncated).toBe(false); + expect(whole.tags.map((tag) => tag.value)).toStrictEqual(['b', 'c']); + expect(whole.tags.every((tag) => tag.revision <= whole.revision)).toBe(true); + }); + + it('holds nothing that took effect before the moment it is asked from', async () => { + const at = await scope(); + await invalidate(at, 'a'); + + const later = await recent(at, `since=0&from=${String(Date.now() + 60_000)}`); + + expect(later).toMatchObject({ truncated: false, tags: [] }); + }); + + it('answers at least one, however few it is asked for', async () => { + const at = await scope(); + await invalidate(at, 'a'); + await invalidate(at, 'b'); + + const window = await recent(at, 'since=0&from=0&limit=0'); + + expect(window.tags.map((tag) => tag.value)).toStrictEqual(['b']); + expect(window.truncated).toBe(true); + }); +}); diff --git a/packages/host/src/cache/gateway-contract.ts b/packages/host/src/cache/gateway-contract.ts index 016418e48..606393dbf 100644 --- a/packages/host/src/cache/gateway-contract.ts +++ b/packages/host/src/cache/gateway-contract.ts @@ -584,7 +584,7 @@ export function answersTheRead( return answer.entry.kind === read.kind && answer.entry.handler === read.handler; } -/** `GET /tags/delta?since=` */ +/** `GET /tags/delta?since=&limit=` */ const tagHistoryMarkSchema = z.object({ revision: z.number().int().positive(), staleAt: z.number().int(), @@ -680,6 +680,53 @@ export function deltaMovesOn( return !delta.truncated || delta.revision > since; } +/** + * Whether no tag a window of recent invalidations holds stands past the revision it was read at, nor + * any earlier deadline one carries: a reader stands there once it holds the window, and a row past + * it would be an invalidation it took in at a revision it does not stand at. + */ +function standsPastItsTags(window: { + readonly revision: number; + readonly tags: readonly Pick[]; +}): boolean { + return window.tags.every((tag) => { + return ( + tag.revision <= window.revision && + tag.history?.every((mark) => mark.revision <= window.revision) !== false + ); + }); +} + +/** + * `GET /tags/recent?since=&from=&limit=`: the invalidations past `since` + * that took effect at `from` or later, each as it was recorded — a row for each invalidation of a + * tag, or a tag's fold of those retention folded, which counts as all it stands for — oldest first, + * the newest `limit` of them at the most (`MAX_DELTA_PAGE_TAGS`, which a larger one is held to), and + * `truncated` where there were more. What an isolate whose view of the tags is behind its scope may + * still find its reads behind on, in one round trip: the delta brings the scope's whole history, a + * page at a time and oldest first, and the newest last. + * + * Read at the scope's revision (`revision`): every such invalidation up to there is in a window that + * is not truncated. Never a page of the delta: a truncated window is cut at its oldest end, not where + * a revision ends, so it says nothing of where a reader that pages the delta may stand — the delta's + * own shape refuses it (`standsAtItsTags`) — and a reader that holds a window stands at its revision + * for what it judges a request's reads by, never for what it commits (`TagState.seen`, in the + * runtime). Its rows carry no history: each is one invalidation, or one fold. + */ +export const tagsRecentResponseSchema = z + .object({ + /** The scope's revision the window was read at. */ + revision: z.number().int().nonnegative(), + /** Set where the window held more than the answer: the oldest are the ones left out. */ + truncated: z.boolean(), + tags: z.array(tagDeltaEntrySchema).max(MAX_DELTA_PAGE_TAGS), + }) + .refine(standsPastItsTags, { + message: 'a window holds no tag revision past the one it was read at', + path: ['revision'], + }); +export type TagsRecentResponse = z.infer; + /** Regional reads are bounded independently of the number of tags in a deployment. */ export const tagsGetRequestSchema = z.object({ tags: z.array(tagValueSchema).max(MAX_TAGS_PER_CALL), diff --git a/packages/host/src/cache/index.ts b/packages/host/src/cache/index.ts index a97d761ca..d080bf30f 100644 --- a/packages/host/src/cache/index.ts +++ b/packages/host/src/cache/index.ts @@ -160,6 +160,8 @@ export { tagsGetRequestSchema, tagsGetResponseSchema, type TagsGetResponse, + type TagsRecentResponse, + tagsRecentResponseSchema, } from './gateway-contract.ts'; export { type CacheNaming, diff --git a/packages/host/test/cache-data-contract.test.ts b/packages/host/test/cache-data-contract.test.ts index cca584214..5fece62c3 100644 --- a/packages/host/test/cache-data-contract.test.ts +++ b/packages/host/test/cache-data-contract.test.ts @@ -20,6 +20,7 @@ import { tagDeltaEntrySchema, tagsDeltaResponseSchema, tagsGetResponseSchema, + tagsRecentResponseSchema, } from '../src/cache/gateway-contract.ts'; /** An entry's metadata: a `use cache` entry's under the handler that made it, a fetch entry's under none. */ @@ -376,6 +377,59 @@ describe('a page of the delta', () => { }); }); +/** A window of recent invalidations holding `count` records, cut at its oldest end. */ +function recentWindow(count: number) { + return { revision: 1, truncated: true, ...tagsAnswer(count) }; +} + +describe('a window of recent invalidations', () => { + /** + * Cut at its oldest end where it held more, a window stands at the scope's revision, past the + * newest invalidation it holds: what a truncated page of the delta may not do, as a reader of the + * delta would stand past the invalidations the next page brings. + */ + it('stands past the newest invalidation it holds, though more were left out', () => { + const window = { + revision: 9, + truncated: true, + tags: [ + { ...tagRecord('a'), revision: 7 }, + { ...tagRecord('b'), revision: 8 }, + ], + }; + + expect(tagsRecentResponseSchema.safeParse(window).success).toBe(true); + expect(tagsDeltaResponseSchema.safeParse({ ...window, floor: 0 }).success).toBe(false); + }); + + it('holds no invalidation, nor earlier deadline, past the revision it was read at', () => { + const past = { revision: 9, truncated: false, tags: [{ ...tagRecord('a'), revision: 10 }] }; + const pastHistory = { + revision: 9, + truncated: false, + tags: [ + { + ...tagRecord('a'), + revision: 9, + history: [{ revision: 10, staleAt: 1, hardExpireAt: null }], + }, + ], + }; + + expect(tagsRecentResponseSchema.safeParse(past).success).toBe(false); + expect(tagsRecentResponseSchema.safeParse(pastHistory).success).toBe(false); + }); + + it('holds no more invalidations than a page of the delta may', () => { + expect(tagsRecentResponseSchema.safeParse(recentWindow(MAX_DELTA_PAGE_TAGS)).success).toBe( + true, + ); + expect(tagsRecentResponseSchema.safeParse(recentWindow(MAX_DELTA_PAGE_TAGS + 1)).success).toBe( + false, + ); + }); +}); + describe("a tag's record in a delta or a tag read", () => { /** What a runtime keeps each one under, as a key of its tag cache. */ it('names a tag no longer than an invalidation may', () => { diff --git a/packages/runtime/src/cache/host.ts b/packages/runtime/src/cache/host.ts index b180b04eb..de417ea41 100644 --- a/packages/runtime/src/cache/host.ts +++ b/packages/runtime/src/cache/host.ts @@ -120,6 +120,19 @@ export interface CacheHost { */ tagsDelta(since: number): Promise; + /** + * The newest invalidations recorded past `since` that took effect at `from` or later (unix ms, by + * the clock the host stamps them with), as many as the host answers at once, each as it was + * recorded: what an isolate whose view is behind its scope may still find its reads behind on, in + * one call, where `tagsDelta` brings every invalidation the scope ever recorded, oldest first, a + * call a page — the newest last. Not a page of the delta: the view holds one that left nothing out + * for what it judges a request's reads by, never for what a regeneration claims it judged against + * (`TagState.seen`). Optional; a host without it, or one that throws a `CacheHostError` with the + * code `unsupported` — a host whose store has no such call after all — is caught up with through + * `tagsDelta` alone. + */ + tagsRecent?(since: number, from: number): Promise; + /** Record an invalidation against these tags, and say what it made of each. */ invalidate(request: InvalidateRequest): Promise; @@ -138,7 +151,7 @@ export interface CacheHost { * A call the host refused, as the runtime reports it. * * `code` is the host's own word for the refusal, and is logged. One word is agreed, and acted on: - * `unsupported`, from a call the host turns out not to have (`uploadArtifacts`), whose work the + * `unsupported`, from a call the host turns out not to have (`uploadArtifacts`, `tagsRecent`), whose work the * runtime then does as it did before there was such a call. Nothing else it does next depends on * which refusal it was. */ @@ -440,6 +453,14 @@ export interface TagDelta extends TagInvalidations { readonly truncated: boolean; } +/** The newest invalidations past a revision that took effect lately (`CacheHost.tagsRecent`). */ +export interface TagWindow extends TagInvalidations { + /** The revision of the scope the window was read at: no invalidation it holds is past it. */ + readonly revision: number; + /** Set where there were more than the answer holds: the oldest are the ones left out. */ + readonly truncated: boolean; +} + export interface InvalidateRequest { readonly tags: readonly string[]; /** Seconds the stale content may still be served; absent means at once. */ From 9dcbb7e6ad5f63fd17ff98c4cf96524eb1069c53 Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Tue, 3 Nov 2026 11:39:55 +0900 Subject: [PATCH 5/8] perf(runtime): catch a request up through a window of recent invalidations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A request whose scope stands past this isolate's view of the tags, at an invalidation made in the last ten minutes, waits up to a second for the view to catch up before it is routed (`catchUpWithScope`). The only way up was the delta, which a fresh isolate pages through from the scope's beginning, oldest first, a round trip a page: on a scope that has invalidated thousands of tags, every early request of a fresh isolate waited the whole second, and the invalidations the wait was for came last. The request now asks, beside the pull, for the window of the scope's recent invalidations past where the view stands (`CacheHost.tagsRecent`), one shared window in flight per isolate as there is one pull, and goes on once either has brought the view up. What the window brings is taken in as marks that judge what the request reads — by timestamps, as data reads and records are judged — and the view stands at its revision for that (`TagState.seen`), which is what the gate asks now. It moves nothing a regeneration claims. A window holds the newest invalidations and none before them, so: - `revision`, which a commit says its render was judged against, is still what the delta brought the view to; regenerations, captures, speculative renders and commits read it as before, and the pull goes on behind the request as it did, so the next regeneration's pull is short; - a window's marks go into a tag's history but never into what the view knows of the tag (`knownRevision`), which says whether it holds the tag whole (`holdsWhole`, `heldRevisionOf`): a mark past one the view let go of would otherwise have a commit claim the deadline it let go of; - a mark a window brought past the revision a render read against is one that render missed, whatever its time says (`#heldBy`), so a render being published answers no request at once after it; - a read of a tag still asks the host of it (`checkedAt` stays as it was), and a tag let go of is remembered at its window's revision too. A window that left some out takes its marks in but stands the view nowhere, and the request waits for the pull as before; so does one from a host without the call, or a gateway that answers it `unsupported`, which is asked no more. Tests: `recent-tags.test.ts` (the view), a regeneration after a window in `regenerate-held-tags.test.ts`, and the catch-up itself in `edge/test/scope-revision.test.ts`. --- packages/cache/README.md | 7 + packages/edge/test/scope-revision.test.ts | 158 +++++++++- packages/runtime/README.md | 12 +- packages/runtime/src/cache/pulls.ts | 110 ++++++- packages/runtime/src/cache/tags.ts | 109 ++++++- packages/runtime/test/recent-tags.test.ts | 285 ++++++++++++++++++ .../runtime/test/regenerate-held-tags.test.ts | 37 +++ 7 files changed, 696 insertions(+), 22 deletions(-) create mode 100644 packages/runtime/test/recent-tags.test.ts diff --git a/packages/cache/README.md b/packages/cache/README.md index 5d2dff821..47e2ea140 100644 --- a/packages/cache/README.md +++ b/packages/cache/README.md @@ -9,6 +9,13 @@ Each deployment has a scope of its own, kept by one Durable Object with one SQLi Function reads and writes it through a gateway, behind a token signed for that scope; a read that needs no authority is answered from storage by a read gateway placed nowhere in particular. +A Function's isolate brings its view of the scope's invalidations up through the tag delta, oldest +first and a page at a time (`/tags/delta`), which a fresh isolate reads from the scope's beginning. +One that a request finds behind a recent invalidation asks first for the newest invalidations past +where its view stands that took effect lately (`/tags/recent`), in one round trip whatever the +scope's history, and judges the request's reads by them while the delta goes on behind; what a +regeneration claims its render was judged against is still the revision the delta brought it to. + What a host adds is handed in. `AuthorityHost` carries the label its tokens are signed under, where its scopes are placed, a meter of what is read and written, and whom to tell when a scope moves. A history of what each change did is a subclass's (`CacheScope.historyOf`). Where a host adds diff --git a/packages/edge/test/scope-revision.test.ts b/packages/edge/test/scope-revision.test.ts index 5d30ad3b1..a012678c1 100644 --- a/packages/edge/test/scope-revision.test.ts +++ b/packages/edge/test/scope-revision.test.ts @@ -81,6 +81,8 @@ interface Authority { readonly fetchImpl: FetchLike; /** How many times the delta was asked for. */ readonly pulls: () => number; + /** How many times the window of recent invalidations was asked for. */ + readonly windows: () => number; /** Let the delta answer, where it was held. */ readonly release: () => void; } @@ -105,20 +107,42 @@ function deltaAtSeven(): Awaited> { }; } -/** An authority whose delta stands at revision 7, its answers held until released where asked. */ -function authorityAhead(options: { readonly held?: boolean } = {}): Authority { +/** The window of recent invalidations an authority at revision 7 answers: the delta's one mark. */ +function windowAtSeven(truncated = false): { + revision: number; + truncated: boolean; + tags: unknown[]; +} { + return { revision: 7, truncated, tags: deltaAtSeven().tags }; +} + +/** + * An authority whose delta stands at revision 7, its answers held until released where asked; and + * whose window of recent invalidations answers as `window` says, where it says — a gateway that + * fails it otherwise, as one that answers nothing it is not asked of in these tests. + */ +function authorityAhead( + options: { readonly held?: boolean; readonly window?: () => Response } = {}, +): Authority { const gate = Promise.withResolvers(); if (options.held !== true) { gate.resolve(true); } let pulls = 0; + let windows = 0; return { pulls: () => pulls, + windows: () => windows, release: () => { gate.resolve(true); }, fetchImpl: async (request) => { - if (new URL(request.url).pathname.endsWith('/tags/delta')) { + const { pathname } = new URL(request.url); + if (pathname.endsWith('/tags/recent')) { + windows += 1; + return options.window?.() ?? new Response('not a tag route', { status: 500 }); + } + if (pathname.endsWith('/tags/delta')) { pulls += 1; await gate.promise; return Response.json({ @@ -336,6 +360,134 @@ describe('an isolate whose view of the tags is behind its scope', () => { } }); + /** + * A fresh isolate's pull pages through the scope's whole delta, the invalidation the request is + * behind on last: the window of recent invalidations brings it in one round trip, and the request + * goes on with it, the pull going on behind — for what a regeneration claims, which is the pull's. + */ + it('goes on once a window of recent invalidations brings the view up, the pull still out', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }); + const authority = authorityAhead({ + held: true, + window: () => Response.json({ ok: true, result: windowAtSeven() }), + }); + const runtime = runtimeFor(authority.fetchImpl); + const behind: Promise[] = []; + const state = { done: false }; + const going = (async () => { + await caughtUp(runtime, requestAt(7, SECOND), behind); + state.done = true; + })(); + try { + await vi.advanceTimersByTimeAsync(1); + const caughtUpThen = { + done: state.done, + seen: runtime.tags.seen, + revision: runtime.tags.revision, + validity: runtime.tags.validityOf(['products'], T0 - 2 * SECOND, T0), + }; + // The next request behind the same invalidation asks nothing, the pull still out. + const next = caughtUp(runtime, requestAt(7, SECOND)); + await vi.advanceTimersByTimeAsync(1); + const askedThen = authority.windows(); + authority.release(); + await vi.advanceTimersByTimeAsync(1); + await Promise.all([next, ...behind]); + + expect(caughtUpThen).toStrictEqual({ done: true, seen: 7, revision: 0, validity: 'expired' }); + expect(askedThen).toBe(1); + expect(behind).toHaveLength(1); + expect(runtime.tags.revision).toBe(7); + expect({ pulls: authority.pulls(), windows: authority.windows() }).toStrictEqual({ + pulls: 1, + windows: 1, + }); + } finally { + authority.release(); + await going; + await Promise.all(behind); + } + }); + + /** + * A window that left some out holds the newest invalidations and not the older ones the view may + * be behind on: its marks are taken in, and the request waits for the pull as it did before there + * was a window. + */ + it('waits for the pull where the window left some out, its marks taken in meanwhile', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }); + const authority = authorityAhead({ + held: true, + window: () => Response.json({ ok: true, result: windowAtSeven(true) }), + }); + const runtime = runtimeFor(authority.fetchImpl); + const behind: Promise[] = []; + const state = { done: false }; + const going = (async () => { + await caughtUp(runtime, requestAt(7, SECOND), behind); + state.done = true; + })(); + try { + await vi.advanceTimersByTimeAsync(SECOND / 2); + const meanwhile = { + done: state.done, + seen: runtime.tags.seen, + validity: runtime.tags.validityOf(['products'], T0 - 2 * SECOND, T0), + }; + authority.release(); + await vi.advanceTimersByTimeAsync(1); + + expect(meanwhile).toStrictEqual({ done: false, seen: 0, validity: 'expired' }); + expect(state.done).toBe(true); + expect(runtime.tags.revision).toBe(7); + } finally { + authority.release(); + await going; + await Promise.all(behind); + } + }); + + /** + * Where the view has seen the scope's latest invalidation through a window, a request behind it + * asks nothing more, though no pull has brought the view's revision there: a regeneration brings + * that up for itself. + */ + it('asks nothing more of a scope a window brought the view up to, though the pull failed', async () => { + let pulls = 0; + const runtime = runtimeFor((request) => { + if (new URL(request.url).pathname.endsWith('/tags/recent')) { + return Promise.resolve(Response.json({ ok: true, result: windowAtSeven() })); + } + pulls += 1; + return Promise.resolve(new Response('the authority is away', { status: 503 })); + }); + + const behind: Promise[] = []; + await caughtUp(runtime, requestAt(7, SECOND), behind); + await Promise.all(behind); + const first = { seen: runtime.tags.seen, revision: runtime.tags.revision, pulls }; + await caughtUp(runtime, requestAt(7, SECOND), behind); + await Promise.all(behind); + + expect(first).toStrictEqual({ seen: 7, revision: 0, pulls: 2 }); + expect(pulls).toBe(2); + }); + + /** A gateway from before the window answers it 404: the delta brings the view up, as before. */ + it('catches up through the delta with a gateway that has no window, and asks it no more', async () => { + const authority = authorityAhead({ + window: () => new Response('no such route', { status: 404 }), + }); + const runtime = runtimeFor(authority.fetchImpl); + + await caughtUp(runtime, requestAt(7, SECOND)); + const first = { revision: runtime.tags.revision, windows: authority.windows() }; + await caughtUp(runtime, requestAt(9, SECOND)); + + expect(first).toStrictEqual({ revision: 7, windows: 1 }); + expect(authority.windows()).toBe(1); + }); + it('asks its host where the scope stands where the request says nothing', async () => { const authority = authorityAhead(); const scopeRevision = vi.fn(() => Promise.resolve({ revision: 7, invalidatedAt: T0 - SECOND })); diff --git a/packages/runtime/README.md b/packages/runtime/README.md index dc573e2e3..2c1775739 100644 --- a/packages/runtime/README.md +++ b/packages/runtime/README.md @@ -103,9 +103,15 @@ host may also cache. Where the scope's tags stand can come sooner: on the reques hears of it sooner (`SCOPE_REVISION_HEADER`, `;`), or, where the request says nothing, from a host that keeps it nearby (`scopeRevision`). A Function whose view of the tags is behind a recent invalidation — ten minutes, past which what it did is in what the Function reads -anyway — pulls the tags' delta before the request is judged, or joins a pull that is out, and waits -for it a second at the most (`catchUpWithScope`): past that, or where the pull fails, the request is -judged with the view as it stands, and a pull still out goes on behind it for the requests after. +anyway — asks, before the request is judged, for the invalidations of those ten minutes past where +its view stands, where the host answers them (`tagsRecent`), beside a pull of the tags' delta, +joining either where one is out; and it waits a second at the most for one of them to bring the view +up (`catchUpWithScope`). The window is one round trip, where a fresh isolate's pull pages through the +scope's whole history and brings the newest invalidations last; a window that left some out has the +request wait for the pull, as before. Past that second, or where neither brings the view up, the +request is judged with the view as it stands; the pull goes on behind it either way, for +the requests after and for the regeneration a request leads to, which claims only what the pull +brought its view to. In a deployment whose routes the build split across app Functions, every Function routes every request — each carries the whole `runtime.json` — and answers only its own routes. A request that diff --git a/packages/runtime/src/cache/pulls.ts b/packages/runtime/src/cache/pulls.ts index 4218a76d0..a96a6a370 100644 --- a/packages/runtime/src/cache/pulls.ts +++ b/packages/runtime/src/cache/pulls.ts @@ -12,6 +12,8 @@ import type { CacheRuntime } from './runtime.ts'; * A pull of the scope's tag delta that the requests needing this isolate's view at a revision join * while it is out: the segments of a page prefetched together are behind the same invalidation, and * the first of them to find it pulls it for all of them (`renderSpeculative`, in `generations.ts`). + * And the window of the scope's recent invalidations a request behind its scope asks for beside the + * pull, which the requests behind it join as well (`catchUpWithScope`). */ /** @@ -111,6 +113,86 @@ const RECENT_INVALIDATION_MINUTES = 10; */ const RECENT_INVALIDATION_MS = RECENT_INVALIDATION_MINUTES * MS_PER_MINUTE; +/** + * The window of recent invalidations each runtime has out, or had last (`TagState.recent`), when it + * went out on the isolate's own clock, and what it answered once it has: whether it brought the view + * up. Joined while it is out and younger than the hold, as a pull is; and, once it answered that it + * did not — it left some out, could not be read, or the host has none — for the rest of the hold: one + * asked again from where the view stands would answer as it did, and the pull is what brings the + * view up meanwhile. One that brought the view up is joined no more: a request behind a later + * invalidation asks past it. + */ +interface SharedWindow { + readonly window: Promise; + readonly since: number; + answered: boolean | undefined; +} + +const windows = new WeakMap(); + +/** + * A window of the scope's recent invalidations, as far back as a request is brought up for + * (`RECENT_INVALIDATION_MS`), for the requests after it to join (`SharedWindow`). Never rejects. + */ +function startWindow(runtime: CacheRuntime): Promise { + const underWay: SharedWindow & { window: Promise } = { + window: Promise.resolve(false), + since: performance.now(), + answered: undefined, + }; + underWay.window = (async () => { + const brought = await runtime.tags.recent(runtime.host, nowMs() - RECENT_INVALIDATION_MS); + underWay.answered = brought; + return brought; + })(); + windows.set(runtime, underWay); + return underWay.window; +} + +/** + * Bring what this isolate's view holds of the scope's recent invalidations up to `required` + * (`TagState.seen`): joining a window that is out, or asking for one, and again where it left the + * view short — one asked for before the invalidation answers at the revision before it. None is + * asked for or joined past `until`, and none again once one could not bring the view up: the pull is + * what is left then. Never rejects. + */ +async function catchUpRecent( + runtime: CacheRuntime, + required: number, + until: number, +): Promise { + for ( + let tries = 0; + tries < PULLS_TO_CATCH_UP && runtime.tags.seen < required && performance.now() < until; + tries += 1 + ) { + const underWay = windows.get(runtime); + const joinable = + underWay !== undefined && + underWay.answered !== true && + performance.now() - underWay.since < runtime.holdMs; + if (!(await (joinable ? underWay.window : startWindow(runtime)))) { + return; + } + } +} + +/** + * Once the view has seen `required` (`TagState.seen`), through the window or through the pull, or + * once both have settled short of it, whichever comes first. Never rejects. + */ +async function seenOrSettled( + runtime: CacheRuntime, + required: number, + pulled: Promise, + windowed: Promise, +): Promise { + await Promise.race([pulled, windowed]); + if (runtime.tags.seen < required) { + await Promise.all([pulled, windowed]); + } +} + /** * Where the request's scope stands: as the request says (`SCOPE_REVISION_HEADER`), or, where it * says nothing, as the host knows it nearby (`CacheHost.scopeRevision`), for a moment at the most. @@ -140,10 +222,20 @@ async function scopeOf( * anything the request answers is judged against it: another isolate's invalidation reaches this * one through its reads otherwise, which a host may cache for half a minute and more. * - * Only where the view is behind the scope's latest invalidation, and that invalidation is recent - * (`RECENT_INVALIDATION_MS`). For a second at the most (`SCOPE_CATCH_UP_MS`), joining a pull that is - * out (`catchUp`): past it the request goes on with the view as it stands, and the pull goes on - * behind it for as long as the requests after it may join it. + * Only where the view has not seen the scope's latest invalidation (`TagState.seen`), and that + * invalidation is recent (`RECENT_INVALIDATION_MS`). Two things are asked at once, each joined where + * one is out: the window of the scope's recent invalidations (`catchUpRecent`), one round trip + * however long the scope's history — and the pull of its delta (`catchUp`), which a fresh isolate + * pages through from the scope's beginning, oldest first, the invalidations the request is behind + * on coming last. A request's reads are judged by timestamps, which the window's marks say all of, + * so the request goes on once either has brought the view up to the scope; where the window left + * some out, could not be read, or the host has none, that is the pull, as before there was one. + * + * For a second at the most (`SCOPE_CATCH_UP_MS`): past it the request goes on with the view as it + * stands. The pull goes on behind it either way, for as long as the requests after it may join it: + * what a regeneration claims its render was judged against is the revision the pull brings the view + * to (`TagState.revision`), and the one the request leads to brings it up first, whatever the window + * held — a pull the request began finds it close. */ export async function catchUpWithScope( runtime: CacheRuntime, @@ -155,12 +247,14 @@ export async function catchUpWithScope( if ( scope === undefined || invalidatedAt === null || - scope.revision <= runtime.tags.revision || + scope.revision <= runtime.tags.seen || nowMs() - invalidatedAt > RECENT_INVALIDATION_MS ) { return; } - const caughtUp = catchUp(runtime, scope.revision, performance.now() + SCOPE_CATCH_UP_MS); - waitUntil(settledWithin(caughtUp, runtime.holdMs)); - await settledWithin(caughtUp, SCOPE_CATCH_UP_MS); + const until = performance.now() + SCOPE_CATCH_UP_MS; + const pulled = catchUp(runtime, scope.revision, until); + const windowed = catchUpRecent(runtime, scope.revision, until); + waitUntil(settledWithin(Promise.all([pulled, windowed]), runtime.holdMs)); + await settledWithin(seenOrSettled(runtime, scope.revision, pulled, windowed), SCOPE_CATCH_UP_MS); } diff --git a/packages/runtime/src/cache/tags.ts b/packages/runtime/src/cache/tags.ts index f6ef9e33c..bf0f21aa9 100644 --- a/packages/runtime/src/cache/tags.ts +++ b/packages/runtime/src/cache/tags.ts @@ -7,7 +7,12 @@ import { } from '@stayingupwind/core/cache'; import { Lru } from '@stayingupwind/core/util'; -import type { CacheHost, InvalidateOutcome, TagInvalidation } from './host.ts'; +import { + type CacheHost, + CacheHostError, + type InvalidateOutcome, + type TagInvalidation, +} from './host.ts'; import { TagLosses } from './tag-losses.ts'; import { addMark, @@ -34,10 +39,17 @@ export type { TagMark } from './tag-marks.ts'; */ /** - * What this view asks of the host, and no more: the scope's delta, and the state of named tags. - * Named so the two calls it makes are the two a reader has to account for. + * What this view asks of the host, and no more: the scope's delta, the window of its recent + * invalidations where the host has one, and the state of named tags. Named so the calls it makes + * are the ones a reader has to account for. + */ +export type TagHost = Pick; + +/** + * Hosts whose window of recent invalidations answered that there is no such call — one from before + * it (`unsupported`, `CacheHost`) — which are caught up with through the delta alone from then on. */ -export type TagHost = Pick; +const windowless = new WeakSet(); export interface TagStateOptions { /** How long a synced view is trusted before the delta is asked for again. */ @@ -101,6 +113,8 @@ interface TagKnowledge { readonly history?: readonly TagHistoryMark[] | undefined; readonly checkedAt?: number | undefined; readonly knownRevision?: number | undefined; + /** The latest revision a window of recent invalidations brought a mark of the tag at (`#markRecent`). */ + readonly recentRevision?: number | undefined; readonly unknown?: Marks | undefined; } @@ -184,6 +198,8 @@ export class TagState { readonly #holdMs: number; readonly #log: TagStateOptions['log']; #revision = 0; + /** The revision the latest window of recent invalidations that left none out was read at (`seen`). */ + #recentRevision = 0; #syncedAt: number | undefined; /** * The marks of tags the view has let go of, of whichever kind, counted (`TagLosses`): a mark of a @@ -267,13 +283,40 @@ export class TagState { }); } + /** + * What a window of recent invalidations said of a tag (`recent`), beside what else is known of it: + * each mark's deadline in the tag's history, as the delta's are, and its revision as the tag's + * `recentRevision` — never as `knownRevision`, nor as a moment the host was asked of the tag + * (`checkedAt`). + * + * A window holds the newest invalidations and none before them, so it says nothing of the marks of + * the tag the view let go of: taken as `knownRevision`, a mark of it past one the view let go of + * would have the view hold the tag whole through its revision (`holdsWhole`), and a commit claim + * the deadline it let go of was held (`heldRevisionOf`), while what a read of the tag says may be + * behind that deadline still. And a read of the tag still asks the host of it (`syncLocal`), which + * brings its whole history. + */ + #markRecent(tag: TagInvalidation, revision: number): void { + const known = this.#tags.get(tag.value); + let { history } = known ?? {}; + let latest = known?.recentRevision ?? 0; + const row = { revision, staleAt: tag.staleAt, hardExpireAt: tag.hardExpireAt }; + for (const mark of [...(tag.history ?? []), row]) { + history = withTagHistory(history, mark); + latest = Math.max(latest, mark.revision); + } + this.#tags.set(tag.value, { ...known, history, recentRevision: latest }); + } + /** * Remember that the view let go of what it knew of `value` (`#lost`), where that was a mark known - * at a revision; and count it (`losses`), where it held a mark of any kind. + * at a revision — one a window brought too (`#markRecent`); and count it (`losses`), where it held + * a mark of any kind. */ #lose(value: string, known: TagKnowledge): void { - if (known.knownRevision !== undefined) { - this.#lost.set(value, Math.max(this.#lost.get(value) ?? 0, known.knownRevision)); + const revision = Math.max(known.knownRevision ?? 0, known.recentRevision ?? 0); + if (revision > 0) { + this.#lost.set(value, Math.max(this.#lost.get(value) ?? 0, revision)); } if (known.history !== undefined || known.unknown !== undefined) { this.losses.forget(value); @@ -333,13 +376,15 @@ export class TagState { const applied = this.#applied.get(value); const within = (revision: number | undefined): boolean => revision === undefined || revision <= held; - // One known at a revision past `held` is one the view missed, whatever its time says. + // One known at a revision past `held` is one the view missed, whatever its time says — one a + // window brought as well (`#markRecent`), which is no part of `knownRevision`. return ( !reachesAny(applied?.unanswered?.marks, timestamp) && !reachesAny(applied?.overflow?.marks, timestamp) && !reachesAny(applied?.unrevised, timestamp) && within(applied?.answeredRevision) && within(told?.knownRevision) && + within(told?.recentRevision) && !reachesAny(told?.unknown, timestamp) ); } @@ -402,6 +447,21 @@ export class TagState { return this.#revision; } + /** + * The revision of the scope through which this view holds every invalidation a request's reads may + * still be behind on: the one the delta brought it to (`revision`), or the one a window of recent + * invalidations that left none out was read at (`recent`), the later. Every invalidation at or + * before it is a mark in the view, or took effect before the window reached back to — and is in + * what the isolate reads anyway. + * + * What a request is brought up to before it is judged (`catchUpWithScope`), never what a + * regeneration claims its render was judged against: that is `revision`, which a window does not + * move, as it holds the newest invalidations and not every one. + */ + get seen(): number { + return Math.max(this.#revision, this.#recentRevision); + } + /** * The latest revision of the scope an invalidation of `tag` is known here at: of what this isolate * was told of it and of what the host answered of its own, the later; 0 where neither is known at @@ -479,6 +539,39 @@ export class TagState { return this.#pull(host, now); } + /** + * Take in the newest invalidations past where the view stands (`seen`) that took effect at `from` + * or later, as `host` answers them (`CacheHost.tagsRecent`): `true` where the window left none + * out, and the view then stands at the revision it was read at (`seen`); `false` where it left some + * out — its marks are taken in all the same, a mark only ever holding more back — or where it could + * not be read, or the host has no such read, which one that answers `unsupported` is asked of no + * more (`windowless`). Moves neither `revision` nor anything a commit claims (`#markRecent`), and + * never rejects. + */ + async recent(host: TagHost, from: number): Promise { + if (host.tagsRecent === undefined || windowless.has(host)) { + return false; + } + try { + const window = await host.tagsRecent(this.seen, from); + for (const tag of window.tags) { + this.#markRecent(tag, tag.revision ?? window.revision); + } + if (window.truncated) { + return false; + } + this.#recentRevision = Math.max(this.#recentRevision, window.revision); + return true; + } catch (error) { + if (error instanceof CacheHostError && error.code === 'unsupported') { + windowless.add(host); + } else { + this.#log?.('recent tags not read', error instanceof Error ? error.message : String(error)); + } + return false; + } + } + /** * Check only the tags a read uses, through KV in the caller's region. * diff --git a/packages/runtime/test/recent-tags.test.ts b/packages/runtime/test/recent-tags.test.ts new file mode 100644 index 000000000..70b9cc452 --- /dev/null +++ b/packages/runtime/test/recent-tags.test.ts @@ -0,0 +1,285 @@ +import { TtlCache } from '@stayingupwind/core/util'; +import { describe, expect, it, vi } from 'vitest'; + +import { withClock } from '../src/cache/clock.ts'; +import { HeldTags } from '../src/cache/held-tags.ts'; +import { + type CacheHost, + CacheHostError, + type TagDelta, + type TagInvalidation, + type TagWindow, +} from '../src/cache/host.ts'; +import { answeringAtOnce } from '../src/cache/publishing.ts'; +import type { CacheRuntime, Publishing } from '../src/cache/runtime.ts'; +import { type TagHost, TagState } from '../src/cache/tags.ts'; + +/** + * What a window of a scope's recent invalidations (`TagState.recent`) does to an isolate's view of + * the tags: its marks judge what a request reads, and the view stands at its revision for that + * (`seen`) — but it holds the newest invalidations and none before them, so nothing a regeneration + * claims of what its render held (`revision`, `holdsWhole`, `heldRevisionOf`) moves for it, and a + * read of a tag it brought still asks the host of the tag. + */ + +const HOLD_MS = 5000; +const NOW = 1_800_000_000_000; +const SECOND = 1000; +/** How many tags a view keeps (`MAX_TAGS_HELD`): one pull bringing more lets the first go. */ +const TAGS_HELD = 4096; +const TAG = 'product-1'; +/** The revision the delta brings `TAG`'s invalidation at, and the revision it is read at. */ +const INVALIDATED = 5; +const REVISION = 10; +/** A later invalidation of `TAG`, which a window read past the delta brings. */ +const LATER = 12; +const INVALIDATED_AT = NOW - 60 * SECOND; +const LATER_AT = NOW - 10 * SECOND; + +/** `count` tags no read of these tests is under. */ +function othersOf(count: number): string[] { + return Array.from({ length: count }, (_, index) => `other-${String(index)}`); +} + +/** An invalidation of `value` at once, at `revision`, from `at`. */ +function markOf(value: string, revision: number, at: number): TagInvalidation { + return { value, revision, staleAt: at, hardExpireAt: at }; +} + +/** + * A host whose delta brings `TAG`'s invalidation and `later` other tags after it, whose copy of the + * tags says nothing of any, and whose window answers `window`: counting what it is asked. + */ +function hostOf( + later: number, + window: TagWindow | (() => Promise), +): TagHost & { readonly asked: { windows: number; tags: number } } { + const asked = { windows: 0, tags: 0 }; + const delta: TagDelta = { + revision: REVISION, + truncated: false, + tags: [ + markOf(TAG, INVALIDATED, INVALIDATED_AT), + ...othersOf(later).map((value) => markOf(value, 6, 1)), + ], + }; + return { + asked, + tagsDelta: (since) => { + return Promise.resolve( + since < REVISION ? delta : { revision: REVISION, truncated: false, tags: [] }, + ); + }, + getTags: () => { + asked.tags += 1; + return Promise.resolve({ tags: [] }); + }, + tagsRecent: () => { + asked.windows += 1; + return typeof window === 'function' ? window() : Promise.resolve(window); + }, + }; +} + +const LATER_WINDOW: TagWindow = { + revision: LATER, + truncated: false, + tags: [markOf(TAG, LATER, LATER_AT)], +}; + +/** The view a regeneration brings up through `host`'s delta. */ +async function viewOf(host: TagHost): Promise { + const state = new TagState({ holdMs: HOLD_MS }); + await state.sync(host, NOW, { force: true }); + return state; +} + +describe('a window of recent invalidations in the tag view', () => { + /** + * The pull that brought `TAG`'s mark brought as many tags after it, which let the mark go, and the + * host's copy says nothing of it. A window then brings a later invalidation of `TAG`: taken as what + * the view knows of the tag, it would have the view hold the tag whole through its revision, and a + * commit claim the mark it let go of — whose deadline it no longer has — was held. + */ + it('holds no tag whole for a mark a window brought past one the view let go of', async () => { + const host = hostOf(TAGS_HELD, LATER_WINDOW); + const state = await viewOf(host); + const before = { whole: state.holdsWhole(TAG), held: state.heldRevisionOf(TAG) }; + + await expect(state.recent(host, 0)).resolves.toBe(true); + + expect(before).toStrictEqual({ whole: false, held: 0 }); + expect(state.holdsWhole(TAG)).toBe(false); + expect(state.heldRevisionOf(TAG)).toBe(0); + expect(state.knownRevisionOf(TAG)).toBe(0); + expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('expired'); + }); + + /** + * A mark a window brought is one the view lets go of as it lets go of one the delta brought: past + * so many tags, and remembered as let go of at its revision (`#lost`), so that the tag is held + * whole no more by what is read of it after. + */ + it('remembers letting go of a mark a window brought, as of one the delta brought', async () => { + const host = hostOf(0, LATER_WINDOW); + const state = new TagState({ holdMs: HOLD_MS }); + await state.recent(host, 0); + + await state.syncLocal(host, othersOf(TAGS_HELD), NOW); + + expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('fresh'); + expect(state.holdsWhole(TAG)).toBe(false); + expect(state.heldRevisionOf(TAG)).toBe(0); + }); + + /** + * A regeneration's commit claims the revision the delta brought the view to, and what it held of + * each tag through it; a window read past it moves neither, though the view has seen past it. + */ + it('stands a request at the window’s revision, and a commit at the delta’s', async () => { + const host = hostOf(0, LATER_WINDOW); + const state = await viewOf(host); + + await state.recent(host, 0); + const held = new HeldTags(); + + expect({ seen: state.seen, revision: state.revision }).toStrictEqual({ + seen: LATER, + revision: REVISION, + }); + expect(state.heldRevisionOf(TAG)).toBe(REVISION); + expect(held.of([TAG], state.revision)).toStrictEqual({ [TAG]: REVISION }); + }); + + /** + * A window says nothing of when the host was last asked of a tag: a read of the tag asks its copy + * all the same (`syncLocal`), and is told the tag's whole history there. + */ + it('leaves a read of a tag a window brought to ask the host of it', async () => { + const host = hostOf(0, LATER_WINDOW); + const state = new TagState({ holdMs: HOLD_MS }); + + await state.recent(host, 0); + await state.syncLocal(host, [TAG], NOW); + await state.syncLocal(host, [TAG], NOW); + + expect(host.asked.tags).toBe(1); + }); + + /** + * One that left some out stands the view nowhere: the invalidations it left out are older ones it + * may still be behind on. What it brought holds back what it reaches all the same, as a mark only + * ever holds more back. + */ + it('stands the view no further for a window that left some out, though it takes its marks in', async () => { + const host = hostOf(0, { ...LATER_WINDOW, truncated: true }); + const state = new TagState({ holdMs: HOLD_MS }); + + await expect(state.recent(host, 0)).resolves.toBe(false); + + expect(state.seen).toBe(0); + expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('expired'); + }); + + it('asks a window from where the view has seen, and from the moment it is handed', async () => { + const asked: [number, number][] = []; + const host: TagHost = { + ...hostOf(0, LATER_WINDOW), + tagsRecent: (since, from) => { + asked.push([since, from]); + return Promise.resolve(LATER_WINDOW); + }, + }; + const state = await viewOf(host); + + await state.recent(host, NOW - 600 * SECOND); + await state.recent(host, NOW - 590 * SECOND); + + expect(asked).toStrictEqual([ + [REVISION, NOW - 600 * SECOND], + [LATER, NOW - 590 * SECOND], + ]); + }); + + /** A host from before the window says so once, and is caught up with through its delta alone. */ + it('asks a host that has no window no more, and one without the call nothing', async () => { + const unsupported = hostOf(0, () => + Promise.reject(new CacheHostError('no window here', { code: 'unsupported', status: 404 })), + ); + const failing = hostOf(0, () => { + return Promise.reject( + new CacheHostError('the authority is away', { code: 'internal', status: 503 }), + ); + }); + const without: TagHost = { getTags: unsupported.getTags, tagsDelta: unsupported.tagsDelta }; + const state = new TagState({ holdMs: HOLD_MS }); + + const answers = [ + await state.recent(unsupported, 0), + await state.recent(unsupported, 0), + await state.recent(failing, 0), + await state.recent(failing, 0), + await state.recent(without, 0), + ]; + + expect(answers).toStrictEqual([false, false, false, false, false]); + expect({ + unsupported: unsupported.asked.windows, + failing: failing.asked.windows, + }).toStrictEqual({ + unsupported: 1, + failing: 2, + }); + expect(state.seen).toBe(0); + }); +}); + +describe('a render being published, after a window', () => { + /** + * The render read against the delta at `REVISION`, and a window since brings an invalidation of one + * of its tags past it, stamped before the pull the render read against was asked for: known past + * the revision the render's view stood at, it is one the render missed, whatever its time says, and + * the render answers no request of the isolate's at once. + */ + it('answers no request at once once a window brings a mark of its tags past its revision', async () => { + const host = hostOf(0, { + revision: LATER, + truncated: false, + tags: [markOf(TAG, LATER, INVALIDATED_AT - SECOND)], + }); + const tags = await viewOf(host); + const publishing: Publishing = { + render: {} as Publishing['render'], + attemptId: 'attempt', + tags: [TAG], + observedTagRevision: REVISION, + observedLosses: tags.losses.count, + held: new HeldTags(), + observedAt: NOW, + published: new Promise(() => { + // The publish is out while the test looks. + }), + }; + const frozen = (): number => NOW; + const runtime = { + tags, + holdMs: HOLD_MS, + publishing: new TtlCache(HOLD_MS, 4, frozen), + letGo: new WeakSet(), + host: host as unknown as CacheHost, + log: vi.fn(), + } as unknown as CacheRuntime; + runtime.publishing.set('entry', publishing); + + const before = await withClock(NOW, () => + Promise.resolve(answeringAtOnce(runtime, 'expired', 'entry')), + ); + await tags.recent(host, 0); + const after = await withClock(NOW, () => + Promise.resolve(answeringAtOnce(runtime, 'expired', 'entry')), + ); + + expect(before).toBe(publishing); + expect(after).toBeUndefined(); + }); +}); diff --git a/packages/runtime/test/regenerate-held-tags.test.ts b/packages/runtime/test/regenerate-held-tags.test.ts index 08c7fdca8..b5e304457 100644 --- a/packages/runtime/test/regenerate-held-tags.test.ts +++ b/packages/runtime/test/regenerate-held-tags.test.ts @@ -10,6 +10,7 @@ import type { DataRead, TagDelta, TagInvalidation, + TagWindow, } from '../src/cache/host.ts'; import { answeringAtOnce, publishingKeyOf } from '../src/cache/publishing.ts'; import { regenerate } from '../src/cache/regenerate.ts'; @@ -70,6 +71,8 @@ interface Scenario { readonly tags: readonly string[]; /** What it reads behind its answer, in work handed to `waitUntil`, once the render is over. */ readonly behind?: readonly Read[] | undefined; + /** A window of recent invalidations the isolate took in before the regeneration, where one did. */ + readonly window?: TagWindow | undefined; } /** The moment every clock of these tests reads: the isolate's, and what a request acts at. */ @@ -161,6 +164,7 @@ function hostOf(scenario: Scenario, commits: CommitRequest[], hooks: Hooks): Cac getTags: (values) => Promise.resolve({ tags: scenario.copies.filter((tag) => values.includes(tag.value)) }), tagsDelta: (since) => Promise.resolve(since < REVISION ? delta : past), + tagsRecent: () => Promise.resolve(scenario.window ?? past), invalidate: unasked, }; } @@ -236,6 +240,9 @@ async function regenerated(scenario: Scenario): Promise { logged, ); asking.runtime = runtime; + if (scenario.window !== undefined) { + await runtime.tags.recent(runtime.host, 0); + } configureCacheHandlers(runtime); let answers: unknown[] = []; vi.mocked(invokeNodeHandler).mockImplementation(async (input) => { @@ -327,6 +334,36 @@ describe('what a regeneration commits of what its render held of each tag', () = expect(commit?.heldTagRevisions).toMatchObject({ [TAG]: 3, [UNREAD]: REVISION }); }); + /** + * A window of recent invalidations read past where the delta stands moves what the isolate's + * requests are judged by (`TagState.seen`), never what a regeneration claims: it holds the newest + * invalidations and none before them, and the commit says the revision the delta brought its view + * to, as a regeneration's always did. + */ + it('claims the revision the delta brought the view to, though a window was read past it', async () => { + const window: TagWindow = { + revision: PAST_THE_VIEW, + truncated: false, + tags: [ + { value: UNREAD, staleAt: INVALIDATED_AT, hardExpireAt: null, revision: PAST_THE_VIEW }, + ], + }; + + const { commit, logged } = await regenerated({ + later: 0, + copies: [], + entries: {}, + reads: [], + tags: [UNREAD], + window, + }); + + expect(logged).toStrictEqual([]); + expect(commit?.observedTagRevision).toBe(REVISION); + expect(commit?.heldTagRevisions).toMatchObject({ [UNREAD]: REVISION }); + expect(Math.max(...Object.values(commit?.heldTagRevisions ?? {}))).toBe(REVISION); + }); + /** A view that let go of nothing of a tag knows the whole of it up to its revision. */ it('holds a tag the render read with its mark kept through the view’s revision', async () => { const { commit, answers } = await regenerated({ From 85f0fd68d857588271c09fb98c4726229df8406c Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Tue, 3 Nov 2026 12:13:05 +0900 Subject: [PATCH 6/8] fix(runtime): hold a window's marks apart from the tags the delta brings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A window's marks went into the tag view's one bound of tags (`MAX_TAGS_HELD`), ahead of the delta a fresh isolate pages through after them, oldest first: on a scope with more tags than the bound, the delta let the window's marks go first — the newest invalidations, the ones the view stood at the window's revision by (`seen`). A read of such a tag then asked a regional copy of it, which may be behind them still, and invalidated data could be served as fresh again until the delta reached them (Codex). The marks are now held apart (`#recent`), judged with the rest — a request's reads, what a render's view missed, a tag's latest moment — and let go of as the delta brings the view past them. Past as many as are held apart, the first let go of keeps the view at the revision the delta brought it to (`seen`), so a request behind its scope is caught up again, until the delta passes it or a window asked from there brings it back. Also from review (cubic): a window's row may carry no earlier deadline past the row's own revision, as the delta's rows may not; the route test asserts the exact revision; the cache README says the window is asked beside the delta, not before it. --- packages/cache/README.md | 9 +- packages/cache/test/tags-recent-route.test.ts | 3 +- packages/host/src/cache/gateway-contract.ts | 7 +- .../host/test/cache-data-contract.test.ts | 16 ++++ packages/runtime/src/cache/tags.ts | 92 +++++++++++++++---- packages/runtime/test/recent-tags.test.ts | 65 +++++++++++-- 6 files changed, 160 insertions(+), 32 deletions(-) diff --git a/packages/cache/README.md b/packages/cache/README.md index 47e2ea140..7612552a1 100644 --- a/packages/cache/README.md +++ b/packages/cache/README.md @@ -11,10 +11,11 @@ needs no authority is answered from storage by a read gateway placed nowhere in A Function's isolate brings its view of the scope's invalidations up through the tag delta, oldest first and a page at a time (`/tags/delta`), which a fresh isolate reads from the scope's beginning. -One that a request finds behind a recent invalidation asks first for the newest invalidations past -where its view stands that took effect lately (`/tags/recent`), in one round trip whatever the -scope's history, and judges the request's reads by them while the delta goes on behind; what a -regeneration claims its render was judged against is still the revision the delta brought it to. +One that a request finds behind a recent invalidation asks, beside the delta, for the newest +invalidations past where its view stands that took effect lately (`/tags/recent`), in one round +trip whatever the scope's history, and judges the request's reads by them while the delta goes on +behind; what a regeneration claims its render was judged against is still the revision the delta +brought it to. What a host adds is handed in. `AuthorityHost` carries the label its tokens are signed under, where its scopes are placed, a meter of what is read and written, and whom to tell when a scope moves. A diff --git a/packages/cache/test/tags-recent-route.test.ts b/packages/cache/test/tags-recent-route.test.ts index 033b69f0e..a7cd93345 100644 --- a/packages/cache/test/tags-recent-route.test.ts +++ b/packages/cache/test/tags-recent-route.test.ts @@ -80,7 +80,8 @@ describe('the window of a scope’s recent invalidations', () => { expect(window.truncated).toBe(true); expect(window.tags.map((tag) => tag.value)).toStrictEqual(['b', 'c']); - expect(window.revision).toBeGreaterThanOrEqual(last); + // Nothing else invalidates the scope meanwhile: the revision it was read at is the last (cubic). + expect(window.revision).toBe(last); // Whole past where it was asked from, and never past the revision it was read at. const whole = await recent(at, `since=${String(first)}&from=0`); diff --git a/packages/host/src/cache/gateway-contract.ts b/packages/host/src/cache/gateway-contract.ts index 606393dbf..8f02dcf25 100644 --- a/packages/host/src/cache/gateway-contract.ts +++ b/packages/host/src/cache/gateway-contract.ts @@ -682,8 +682,9 @@ export function deltaMovesOn( /** * Whether no tag a window of recent invalidations holds stands past the revision it was read at, nor - * any earlier deadline one carries: a reader stands there once it holds the window, and a row past - * it would be an invalidation it took in at a revision it does not stand at. + * any earlier deadline one carries past the tag's own, as the delta's rows are held (`standsAtItsTags`, + * cubic): a reader stands there once it holds the window, and a row past it would be an + * invalidation it took in at a revision it does not stand at. */ function standsPastItsTags(window: { readonly revision: number; @@ -692,7 +693,7 @@ function standsPastItsTags(window: { return window.tags.every((tag) => { return ( tag.revision <= window.revision && - tag.history?.every((mark) => mark.revision <= window.revision) !== false + tag.history?.every((mark) => mark.revision <= tag.revision) !== false ); }); } diff --git a/packages/host/test/cache-data-contract.test.ts b/packages/host/test/cache-data-contract.test.ts index 5fece62c3..b0e060f69 100644 --- a/packages/host/test/cache-data-contract.test.ts +++ b/packages/host/test/cache-data-contract.test.ts @@ -420,6 +420,22 @@ describe('a window of recent invalidations', () => { expect(tagsRecentResponseSchema.safeParse(pastHistory).success).toBe(false); }); + it('holds no earlier deadline of a tag past the tag’s own revision (cubic)', () => { + const pastItsTag = { + revision: 12, + truncated: false, + tags: [ + { + ...tagRecord('a'), + revision: 9, + history: [{ revision: 10, staleAt: 1, hardExpireAt: null }], + }, + ], + }; + + expect(tagsRecentResponseSchema.safeParse(pastItsTag).success).toBe(false); + }); + it('holds no more invalidations than a page of the delta may', () => { expect(tagsRecentResponseSchema.safeParse(recentWindow(MAX_DELTA_PAGE_TAGS)).success).toBe( true, diff --git a/packages/runtime/src/cache/tags.ts b/packages/runtime/src/cache/tags.ts index bf0f21aa9..dee93e0cc 100644 --- a/packages/runtime/src/cache/tags.ts +++ b/packages/runtime/src/cache/tags.ts @@ -113,11 +113,25 @@ interface TagKnowledge { readonly history?: readonly TagHistoryMark[] | undefined; readonly checkedAt?: number | undefined; readonly knownRevision?: number | undefined; - /** The latest revision a window of recent invalidations brought a mark of the tag at (`#markRecent`). */ - readonly recentRevision?: number | undefined; readonly unknown?: Marks | undefined; } +/** + * How many tags the marks windows of recent invalidations brought are held for, apart from the + * tags the delta brings (`#recent`): a window holds at most a page of them, and they are let go of + * as the delta brings the same marks. + */ +const MAX_RECENT_MARKS = 4096; + +/** + * What windows of recent invalidations said of one tag (`#markRecent`): each mark's deadline, and + * the latest revision one was at (`recentRevision`). + */ +interface RecentKnowledge { + readonly history: readonly TagHistoryMark[]; + readonly recentRevision: number; +} + /** * One call of this Function's that records invalidations of tags, from `applyLocal` until the host * has answered it: the tags it names, each once, and the mark it puts in force here. An answer @@ -200,6 +214,21 @@ export class TagState { #revision = 0; /** The revision the latest window of recent invalidations that left none out was read at (`seen`). */ #recentRevision = 0; + /** + * The marks windows of recent invalidations brought, held apart from the tags the delta brings + * (`#markRecent`) until the delta brings them too (`#pull`): the delta of a fresh isolate brings + * the scope's whole history after them, oldest first, and in the bound of its own tags + * (`MAX_TAGS_HELD`) would let go of the newest marks first — the ones the view stands at the + * window's revision by (`seen`), which a read of the tag from a regional copy may be behind on + * still (Codex). + */ + readonly #recent = new Map(); + /** + * The latest revision of a window's mark let go of before the delta brought it + * (`MAX_RECENT_MARKS`): the view stands no further than the delta brought it (`seen`) until the + * delta passes it. + */ + #recentLostPast = 0; #syncedAt: number | undefined; /** * The marks of tags the view has let go of, of whichever kind, counted (`TagLosses`): a mark of a @@ -227,6 +256,7 @@ export class TagState { this.#remember(tag, tag.revision ?? delta.revision); } this.#revision = Math.max(this.#revision, delta.revision); + this.#passRecent(); truncated = delta.truncated; } // Pulls overlap, and one that began earlier can answer last: what a later one brought is in @@ -284,10 +314,9 @@ export class TagState { } /** - * What a window of recent invalidations said of a tag (`recent`), beside what else is known of it: - * each mark's deadline in the tag's history, as the delta's are, and its revision as the tag's - * `recentRevision` — never as `knownRevision`, nor as a moment the host was asked of the tag - * (`checkedAt`). + * What a window of recent invalidations said of a tag (`recent`), held apart from what else is + * known of it (`#recent`): each mark's deadline, and its revision as the tag's `recentRevision` — + * never as `knownRevision`, nor as a moment the host was asked of the tag (`checkedAt`). * * A window holds the newest invalidations and none before them, so it says nothing of the marks of * the tag the view let go of: taken as `knownRevision`, a mark of it past one the view let go of @@ -297,26 +326,41 @@ export class TagState { * brings its whole history. */ #markRecent(tag: TagInvalidation, revision: number): void { - const known = this.#tags.get(tag.value); - let { history } = known ?? {}; + const known = this.#recent.get(tag.value); + let history = known?.history; let latest = known?.recentRevision ?? 0; const row = { revision, staleAt: tag.staleAt, hardExpireAt: tag.hardExpireAt }; for (const mark of [...(tag.history ?? []), row]) { history = withTagHistory(history, mark); latest = Math.max(latest, mark.revision); } - this.#tags.set(tag.value, { ...known, history, recentRevision: latest }); + this.#recent.delete(tag.value); + this.#recent.set(tag.value, { history: history ?? [], recentRevision: latest }); + for (const [value, held] of this.#recent) { + if (this.#recent.size <= MAX_RECENT_MARKS) { + break; + } + this.#recent.delete(value); + this.#recentLostPast = Math.max(this.#recentLostPast, held.recentRevision); + } + } + + /** Let go of the marks of windows the delta has brought the view past (`#recent`). */ + #passRecent(): void { + for (const [value, held] of this.#recent) { + if (held.recentRevision <= this.#revision) { + this.#recent.delete(value); + } + } } /** * Remember that the view let go of what it knew of `value` (`#lost`), where that was a mark known - * at a revision — one a window brought too (`#markRecent`); and count it (`losses`), where it held - * a mark of any kind. + * at a revision; and count it (`losses`), where it held a mark of any kind. */ #lose(value: string, known: TagKnowledge): void { - const revision = Math.max(known.knownRevision ?? 0, known.recentRevision ?? 0); - if (revision > 0) { - this.#lost.set(value, Math.max(this.#lost.get(value) ?? 0, revision)); + if (known.knownRevision !== undefined) { + this.#lost.set(value, Math.max(this.#lost.get(value) ?? 0, known.knownRevision)); } if (known.history !== undefined || known.unknown !== undefined) { this.losses.forget(value); @@ -353,6 +397,7 @@ export class TagState { const told = this.#tags.get(value); const marks = [ told?.history?.at(-1), + this.#recent.get(value)?.history.at(-1), told?.unknown?.at(-1), applied?.history?.at(-1), applied?.unanswered?.marks.at(-1), @@ -384,7 +429,7 @@ export class TagState { !reachesAny(applied?.unrevised, timestamp) && within(applied?.answeredRevision) && within(told?.knownRevision) && - within(told?.recentRevision) && + within(this.#recent.get(value)?.recentRevision) && !reachesAny(told?.unknown, timestamp) ); } @@ -402,7 +447,10 @@ export class TagState { #validityOfTag(tag: string, timestamp: number, now: number, held: number | undefined): Validity { const applied = this.#applied.get(tag); const known = this.#tags.get(tag); - const byTold = historyValidity(known?.history, timestamp, now, held); + const byTold = worst( + historyValidity(known?.history, timestamp, now, held), + historyValidity(this.#recent.get(tag)?.history, timestamp, now, held), + ); const byAnswered = historyValidity(applied?.history, timestamp, now, held); const byUnrevised = worst( worst( @@ -459,7 +507,9 @@ export class TagState { * move, as it holds the newest invalidations and not every one. */ get seen(): number { - return Math.max(this.#revision, this.#recentRevision); + return this.#recentLostPast > this.#revision + ? this.#revision + : Math.max(this.#revision, this.#recentRevision); } /** @@ -553,7 +603,13 @@ export class TagState { return false; } try { - const window = await host.tagsRecent(this.seen, from); + const since = this.seen; + const window = await host.tagsRecent(since, from); + // Asked from no further than the delta stands, a window that left none out brings back every + // mark of a window let go of past it (`#recentLostPast`). + if (!window.truncated && since <= this.#revision) { + this.#recentLostPast = 0; + } for (const tag of window.tags) { this.#markRecent(tag, tag.revision ?? window.revision); } diff --git a/packages/runtime/test/recent-tags.test.ts b/packages/runtime/test/recent-tags.test.ts index 70b9cc452..34422abb8 100644 --- a/packages/runtime/test/recent-tags.test.ts +++ b/packages/runtime/test/recent-tags.test.ts @@ -116,22 +116,75 @@ describe('a window of recent invalidations in the tag view', () => { }); /** - * A mark a window brought is one the view lets go of as it lets go of one the delta brought: past - * so many tags, and remembered as let go of at its revision (`#lost`), so that the tag is held - * whole no more by what is read of it after. + * A mark a window brought is held apart from the tags the delta brings and a read asks the host of + * (`#recent`): however many of those the view lets go of after it, the request the view stands at + * the window's revision for is judged by the mark, not by a regional copy of the tag that may be + * behind it (Codex). */ - it('remembers letting go of a mark a window brought, as of one the delta brought', async () => { + it('keeps a mark a window brought, however many tags the view lets go of after it', async () => { const host = hostOf(0, LATER_WINDOW); const state = new TagState({ holdMs: HOLD_MS }); await state.recent(host, 0); await state.syncLocal(host, othersOf(TAGS_HELD), NOW); - expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('fresh'); - expect(state.holdsWhole(TAG)).toBe(false); + expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('expired'); + expect(state.seen).toBe(LATER); expect(state.heldRevisionOf(TAG)).toBe(0); }); + /** + * A fresh isolate's delta brings the scope's whole history after the window, oldest first: more + * tags than the view keeps, which let the first of its own go. The window's marks are not among + * them (Codex). + */ + it('keeps a window’s marks while the delta brings the scope’s history after them', async () => { + const host = hostOf(TAGS_HELD, LATER_WINDOW); + const state = new TagState({ holdMs: HOLD_MS }); + await state.recent(host, 0); + + await state.sync(host, NOW, { force: true }); + + expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('expired'); + expect({ seen: state.seen, revision: state.revision }).toStrictEqual({ + seen: LATER, + revision: REVISION, + }); + }); + + /** + * Past as many marks of windows as are held apart, the first are let go of: the view stands no + * further than the delta brought it (`seen`) — a request behind its scope is caught up again — + * until the delta passes them. + */ + it('stands where the delta brought it once a window’s mark is let go of, until the delta passes it', async () => { + const many: TagWindow = { + revision: LATER, + truncated: false, + tags: [markOf(TAG, LATER, LATER_AT), ...othersOf(TAGS_HELD).map((v) => markOf(v, 11, 1))], + }; + const host: TagHost = { + tagsDelta: (since) => { + return Promise.resolve({ + revision: LATER, + truncated: false, + tags: since < LATER ? [markOf(TAG, LATER, LATER_AT)] : [], + }); + }, + getTags: () => Promise.resolve({ tags: [] }), + tagsRecent: () => Promise.resolve(many), + }; + const state = new TagState({ holdMs: HOLD_MS }); + + await state.recent(host, 0); + const letGo = state.seen; + await state.sync(host, NOW, { force: true }); + + expect(letGo).toBe(0); + expect(state.seen).toBe(LATER); + expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('expired'); + }); + /** * A regeneration's commit claims the revision the delta brought the view to, and what it held of * each tag through it; a window read past it moves neither, though the view has seen past it. From b5c00d76c2aa4d560bd51b12862ddd120a01b114 Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Tue, 3 Nov 2026 12:27:58 +0900 Subject: [PATCH 7/8] fix(runtime): keep a window's mark until the delta brings its tag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A window's mark was let go of once the delta's revision passed it. But the delta answers each tag's latest row, so a tag invalidated again since the window comes only at its later revision, past pages that reach beyond the window's mark without it: the mark went, the view still stood at the window's revision (`seen`), and a read of the tag judged by a regional copy could call invalidated data fresh until the page with the tag arrived (Codex). The mark now goes once the delta brings the tag itself at its revision or later — and never on a regional read. And a mark let go of past the bound of those held apart said nothing of the loss: the tag was still held whole and a render that began before it could answer again (`answeringAtOnce`), with `seen` lowered but read by neither check (Codex, Greptile). It is now lost as a mark the delta brought is (`#lose`): recorded against the tag's revision and counted in `losses`. --- packages/runtime/src/cache/tags.ts | 34 ++++++---- packages/runtime/test/recent-tags.test.ts | 77 +++++++++++++++++++++++ 2 files changed, 99 insertions(+), 12 deletions(-) diff --git a/packages/runtime/src/cache/tags.ts b/packages/runtime/src/cache/tags.ts index dee93e0cc..75dbf74d9 100644 --- a/packages/runtime/src/cache/tags.ts +++ b/packages/runtime/src/cache/tags.ts @@ -216,9 +216,9 @@ export class TagState { #recentRevision = 0; /** * The marks windows of recent invalidations brought, held apart from the tags the delta brings - * (`#markRecent`) until the delta brings them too (`#pull`): the delta of a fresh isolate brings - * the scope's whole history after them, oldest first, and in the bound of its own tags - * (`MAX_TAGS_HELD`) would let go of the newest marks first — the ones the view stands at the + * (`#markRecent`) until the delta brings their tags too (`#broughtByDelta`): the delta of a fresh + * isolate brings the scope's whole history after them, oldest first, and in the bound of its own + * tags (`MAX_TAGS_HELD`) would let go of the newest marks first — the ones the view stands at the * window's revision by (`seen`), which a read of the tag from a regional copy may be behind on * still (Codex). */ @@ -253,10 +253,11 @@ export class TagState { while (truncated) { const delta = await host.tagsDelta(this.#revision); for (const tag of delta.tags) { - this.#remember(tag, tag.revision ?? delta.revision); + const revision = tag.revision ?? delta.revision; + this.#remember(tag, revision); + this.#broughtByDelta(tag.value, revision); } this.#revision = Math.max(this.#revision, delta.revision); - this.#passRecent(); truncated = delta.truncated; } // Pulls overlap, and one that began earlier can answer last: what a later one brought is in @@ -341,17 +342,26 @@ export class TagState { break; } this.#recent.delete(value); + // Knowledge lost, as a mark the delta brought is when the view lets go of it (`#lose`): the + // tag is held whole no more until the delta brings it past this revision (`holdsWhole`), a + // render that began before it answers no more (`losses`), and the view stands no further than + // the delta brought it (`seen`) — Codex, Greptile. + this.#lose(value, { knownRevision: held.recentRevision, history: held.history }); this.#recentLostPast = Math.max(this.#recentLostPast, held.recentRevision); } } - /** Let go of the marks of windows the delta has brought the view past (`#recent`). */ - #passRecent(): void { - for (const [value, held] of this.#recent) { - if (held.recentRevision <= this.#revision) { - this.#recent.delete(value); - } - } + /** + * The delta brought `value` at `revision`: what a window brought of it at no later revision is in + * the view's own knowledge of the tag now, and is let go of (`#recent`). Not where the delta's + * revision only passed the window's mark: the delta answers each tag's latest row, so a tag + * invalidated again since the window comes at its later revision, past pages that reach beyond + * the window's mark without it (Codex). Nor on a regional read of the tag (`syncLocal`), which may + * be behind the window. + */ + #broughtByDelta(value: string, revision: number): void { + if (revision >= (this.#recent.get(value)?.recentRevision ?? Infinity)) + this.#recent.delete(value); } /** diff --git a/packages/runtime/test/recent-tags.test.ts b/packages/runtime/test/recent-tags.test.ts index 34422abb8..af58ebe12 100644 --- a/packages/runtime/test/recent-tags.test.ts +++ b/packages/runtime/test/recent-tags.test.ts @@ -36,6 +36,11 @@ const LATER = 12; const INVALIDATED_AT = NOW - 60 * SECOND; const LATER_AT = NOW - 10 * SECOND; +/** What a test leaves out, which it does not wait for. */ +function ignore(): void { + // Nothing to report: it is never answered while the test looks. +} + /** `count` tags no read of these tests is under. */ function othersOf(count: number): string[] { return Array.from({ length: count }, (_, index) => `other-${String(index)}`); @@ -185,6 +190,78 @@ describe('a window of recent invalidations in the tag view', () => { expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('expired'); }); + /** + * The delta answers each tag's latest row: a tag invalidated again since the window comes at its + * later revision, past pages that reach beyond the window's mark without it. The window's mark is + * kept until the delta brings the tag itself (Codex). + */ + it('keeps a window’s mark until the delta brings its tag, not until it passes its revision', async () => { + const window: TagWindow = { + revision: 20, + truncated: false, + tags: [markOf(TAG, INVALIDATED, INVALIDATED_AT)], + }; + // The page before the tag's: past the window's mark, without the tag, whose row is at 30 now. + const first: TagDelta = { revision: 10, truncated: true, tags: [markOf('other', 10, 1)] }; + const asked: number[] = []; + const host: TagHost = { + tagsDelta: (since) => { + asked.push(since); + // The page that brings the tag is still out once the first is in: judged between them. + return since === 0 + ? Promise.resolve(first) + : new Promise(() => { + // Never answered while the test looks. + }); + }, + getTags: () => Promise.resolve({ tags: [] }), + tagsRecent: () => Promise.resolve(window), + }; + const state = new TagState({ holdMs: HOLD_MS }); + await state.recent(host, 0); + + void state.sync(host, NOW, { force: true }).catch(ignore); + await vi.waitFor(() => { + expect(asked).toStrictEqual([0, 10]); + }); + + expect(state.revision).toBe(10); + expect(state.seen).toBe(20); + expect(state.validityOf([TAG], INVALIDATED_AT - SECOND, NOW)).toBe('expired'); + }); + + /** + * A window's mark let go of past the bound is knowledge lost, as a mark the delta brought is: the + * tag is held whole no more, and a render that began before it answers no more (Codex, Greptile). + */ + it('counts a window’s mark let go of past the bound as lost', async () => { + let windows = 0; + const host: TagHost = { + tagsDelta: () => Promise.resolve({ revision: 0, truncated: false, tags: [] }), + getTags: () => Promise.resolve({ tags: [] }), + tagsRecent: () => { + windows += 1; + return Promise.resolve({ + revision: LATER, + truncated: false, + tags: + windows === 1 + ? [markOf(TAG, LATER, LATER_AT)] + : othersOf(TAGS_HELD).map((v) => markOf(v, 11, 1)), + }); + }, + }; + const state = new TagState({ holdMs: HOLD_MS }); + await state.recent(host, 0); + const before = state.losses.count; + + await state.recent(host, 0); + + expect(state.holdsWhole(TAG)).toBe(false); + expect(state.losses.lostSince(TAG, before)).toBe(true); + expect(state.seen).toBe(0); + }); + /** * A regeneration's commit claims the revision the delta brought the view to, and what it held of * each tag through it; a window read past it moves neither, though the view has seen past it. From e8f8c0f5e373c03ecab0b3d5a5cd5743581cc7d3 Mon Sep 17 00:00:00 2001 From: "soleil.colza" Date: Tue, 3 Nov 2026 12:35:49 +0900 Subject: [PATCH 8/8] fix(runtime): stand at the delta's revision until a whole pull follows a lost mark MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A window's mark let go of past the bound kept the view at the delta's revision (`seen`) only until the delta's revision passed the mark's. But the delta answers each tag's latest row: a tag invalidated again since comes at its later revision, so a pull that has only passed the mark has not brought the tag, and the view went back to the window's revision with the tag's invalidation missing — a request at that revision then skipped its catch-up (Codex). The view now stands at the delta's revision until a pull has run to its end past the lost mark, or a window asked from where the delta stands brings the marks back. --- packages/runtime/src/cache/tags.ts | 11 ++++-- packages/runtime/test/recent-tags.test.ts | 45 ++++++++++++++++++++++- 2 files changed, 51 insertions(+), 5 deletions(-) diff --git a/packages/runtime/src/cache/tags.ts b/packages/runtime/src/cache/tags.ts index 75dbf74d9..d816ded33 100644 --- a/packages/runtime/src/cache/tags.ts +++ b/packages/runtime/src/cache/tags.ts @@ -224,9 +224,11 @@ export class TagState { */ readonly #recent = new Map(); /** - * The latest revision of a window's mark let go of before the delta brought it - * (`MAX_RECENT_MARKS`): the view stands no further than the delta brought it (`seen`) until the - * delta passes it. + * The latest revision of a window's mark let go of before the delta brought its tag + * (`MAX_RECENT_MARKS`), 0 for none: the view stands no further than the delta brought it (`seen`) + * until a pull of the delta has run to its end past it — the delta answers each tag's latest row, + * so only a whole pull has brought every tag, one invalidated again since among them (Codex) — or + * a window asked from where the delta stands brings the marks back (`recent`). */ #recentLostPast = 0; #syncedAt: number | undefined; @@ -260,6 +262,7 @@ export class TagState { this.#revision = Math.max(this.#revision, delta.revision); truncated = delta.truncated; } + if (this.#revision >= this.#recentLostPast) this.#recentLostPast = 0; // Pulls overlap, and one that began earlier can answer last: what a later one brought is in // the view already, and its hold stands. So does a page another pull took of a delta it has // not finished: this pull's last answer held every invalidation the host had made when this @@ -517,7 +520,7 @@ export class TagState { * move, as it holds the newest invalidations and not every one. */ get seen(): number { - return this.#recentLostPast > this.#revision + return this.#recentLostPast > 0 ? this.#revision : Math.max(this.#revision, this.#recentRevision); } diff --git a/packages/runtime/test/recent-tags.test.ts b/packages/runtime/test/recent-tags.test.ts index af58ebe12..b76f4531f 100644 --- a/packages/runtime/test/recent-tags.test.ts +++ b/packages/runtime/test/recent-tags.test.ts @@ -162,7 +162,7 @@ describe('a window of recent invalidations in the tag view', () => { * further than the delta brought it (`seen`) — a request behind its scope is caught up again — * until the delta passes them. */ - it('stands where the delta brought it once a window’s mark is let go of, until the delta passes it', async () => { + it('stands where the delta brought it once a window’s mark is let go of, until a whole pull passes it', async () => { const many: TagWindow = { revision: LATER, truncated: false, @@ -190,6 +190,49 @@ describe('a window of recent invalidations in the tag view', () => { expect(state.validityOf([TAG], LATER_AT - SECOND, NOW)).toBe('expired'); }); + /** + * A pull that has only passed a let-go mark's revision has not brought its tag, which may have + * been invalidated again since and come later: the view stands where the delta brought it until a + * pull runs to its end (Codex). + */ + it('stands where the delta brought it while a pull past a let-go mark is still out', async () => { + let windows = 0; + const asked: number[] = []; + const host: TagHost = { + tagsDelta: (since) => { + asked.push(since); + return since === 0 + ? Promise.resolve({ revision: 10, truncated: true, tags: [markOf('other', 10, 1)] }) + : new Promise(() => { + // The page with the tag, at 60 now, is never answered while the test looks. + }); + }, + getTags: () => Promise.resolve({ tags: [] }), + tagsRecent: () => { + windows += 1; + return Promise.resolve({ + revision: 50, + truncated: false, + tags: + windows === 1 + ? [markOf(TAG, INVALIDATED, INVALIDATED_AT)] + : othersOf(TAGS_HELD).map((v) => markOf(v, 11, 1)), + }); + }, + }; + const state = new TagState({ holdMs: HOLD_MS }); + await state.recent(host, 0); + await state.recent(host, 0); + + void state.sync(host, NOW, { force: true }).catch(ignore); + await vi.waitFor(() => { + expect(asked).toStrictEqual([0, 10]); + }); + + expect(state.revision).toBe(10); + expect(state.seen).toBe(10); + }); + /** * The delta answers each tag's latest row: a tag invalidated again since the window comes at its * later revision, past pages that reach beyond the window's mark without it. The window's mark is