perf(runtime): catch a request up through a window of recent invalidations

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

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<TagWindow> {
+    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<ScopeRevision | undefined> {
     return await this.#beacon?.read();

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<Uint8Array> {
-    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<CacheHostError> {
+    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<TagsDeltaResponse> {
-    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<TagsRecentResponse> {
+    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<InvalidateResponse> {
     return this.#json('POST', '/invalidations', request, invalidateResponseSchema);
   }

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<Uint8Array> {
+  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

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<GatewayEnv>()
   // 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<GatewayEnv>()
   .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) {