diff --git a/packages/cache/README.md b/packages/cache/README.md index 5d2dff821..7612552a1 100644 --- a/packages/cache/README.md +++ b/packages/cache/README.md @@ -9,6 +9,14 @@ 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, 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 history of what each change did is a subclass's (`CacheScope.historyOf`). Where a host adds 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 4937ef9a0..f5588c2d6 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, @@ -39,6 +40,8 @@ import { tagRevisionSchema, type TagsDeltaResponse, tagsDeltaResponseSchema, + type TagsRecentResponse, + tagsRecentResponseSchema, uploadArtifactsResponseSchema, } from '@stayingupwind/host/cache'; import { @@ -61,8 +64,8 @@ import { import { ArtifactRoom, functionReadBudget, - projectedLength, readArtifactWhole, + readWhole, saidLength, type ReadBudget, } from './read-budget.ts'; @@ -373,27 +376,24 @@ 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. + * 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 #readWhole( + async #unlessNoSuchRoute( 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(); - } + 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. */ @@ -480,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 ( @@ -541,7 +530,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 +552,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)); }); } @@ -662,9 +651,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)) { @@ -677,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/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 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 d28b17aa8..1cadc4136 100644 --- a/packages/cache/test/tag-authority-deadlines.test.ts +++ b/packages/cache/test/tag-authority-deadlines.test.ts @@ -1,13 +1,20 @@ -import { tagKindOf } from '@stayingupwind/core/cache'; +import { type TagKind, tagKindOf } from '@stayingupwind/core/cache'; import { isDeadlineError } from '@stayingupwind/core/util'; -import type { TagRevision, TagsDeltaResponse } from '@stayingupwind/host/cache'; +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, @@ -16,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) { @@ -239,3 +254,117 @@ describe('an authority delta cursor', () => { expect(send).toHaveBeenCalledTimes(1); }); }); + +/** 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 + * 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), + }); + }); + + 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..a7cd93345 --- /dev/null +++ b/packages/cache/test/tags-recent-route.test.ts @@ -0,0 +1,113 @@ +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']); + // 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`); + + 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/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/host/src/cache/gateway-contract.ts b/packages/host/src/cache/gateway-contract.ts index 016418e48..8f02dcf25 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,54 @@ 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 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; + readonly tags: readonly Pick[]; +}): boolean { + return window.tags.every((tag) => { + return ( + tag.revision <= window.revision && + tag.history?.every((mark) => mark.revision <= tag.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..b0e060f69 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,75 @@ 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 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, + ); + 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/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/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. */ 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/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..d816ded33 100644 --- a/packages/runtime/src/cache/tags.ts +++ b/packages/runtime/src/cache/tags.ts @@ -7,8 +7,27 @@ 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, + 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 @@ -20,24 +39,17 @@ import { TagLosses } from './tag-losses.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; +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; -} +/** + * 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. + */ +const windowless = new WeakSet(); export interface TagStateOptions { /** How long a synced view is trusted before the delta is asked for again. */ @@ -91,28 +103,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 @@ -126,6 +116,22 @@ interface TagKnowledge { 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 @@ -206,6 +212,25 @@ 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; + /** + * The marks windows of recent invalidations brought, held apart from the tags the delta brings + * (`#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). + */ + readonly #recent = new Map(); + /** + * 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; /** * The marks of tags the view has let go of, of whichever kind, counted (`TagLosses`): a mark of a @@ -230,11 +255,14 @@ 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); 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 @@ -289,6 +317,56 @@ export class TagState { }); } + /** + * 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 + * 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.#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.#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); + // 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); + } + } + + /** + * 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); + } + /** * 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. @@ -332,6 +410,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), @@ -355,13 +434,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(this.#recent.get(value)?.recentRevision) && !reachesAny(told?.unknown, timestamp) ); } @@ -379,7 +460,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( @@ -424,6 +508,23 @@ 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 this.#recentLostPast > 0 + ? this.#revision + : 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 @@ -501,6 +602,45 @@ 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 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); + } + 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. * @@ -769,112 +909,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, diff --git a/packages/runtime/test/recent-tags.test.ts b/packages/runtime/test/recent-tags.test.ts new file mode 100644 index 000000000..b76f4531f --- /dev/null +++ b/packages/runtime/test/recent-tags.test.ts @@ -0,0 +1,458 @@ +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; + +/** 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)}`); +} + +/** 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 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('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('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 a whole pull 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 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 + * 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. + */ + 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({