fix(bun-cache): one invalidation, one CDN purge

packages/bun-cache/README.md

@@ -2681,10 +2681,11 @@ for that invalidation before it reads. Every entry carries its path's tag,
 so `pages.invalidate(path | url | request)` drops a path's queries and
 variants. Either invalidation also runs a second time, in the background,
 once the writes this middleware had in flight for the path have landed (at
-most `coalesce.wait`): an uncoalesced write, or one whose body is still
-being read (a stream, a sent file), is issued after the invalidation and no
-fence covers it, so without the second pass it could put back what was just
-dropped; the `tags` option (a list, or `(req, res) => tags`) and
+most `coalesce.wait`), if one of them stored after it began: an uncoalesced
+write, or one whose body is still being read (a stream, a sent file), is
+issued after the invalidation and no fence covers it, so without the second
+pass it could put back what was just dropped (a write that landed before it
+began, or that a fence refused, needs none, and gets none); the `tags` option (a list, or `(req, res) => tags`) and
 `addCacheTags(res, ...tags)` from the handler add more for
 `cache.invalidateTags()`.
 
@@ -2923,7 +2924,13 @@ and [purge cache](https://developers.cloudflare.com/cache/how-to/purge-cache/)
 tags in plain text, in the process that called `invalidateTags()` (an
 unsafe method's invalidation and `pages.invalidate()` included; not in the
 processes that only hear of it over the bus), so one listener purges each
-invalidation once:
+invalidation once. One case emits twice, on purpose: when
+`pages.invalidate()` or an unsafe method lands while a `GET` of the path is
+still writing its answer (a streamed body is written at its end), and that
+write stores after the invalidation began, the path is invalidated again
+once it has, and `invalidate` fires again. The CDN may have taken that same
+answer, so the second purge is wanted too. A write that had already landed,
+or that the invalidation refused, causes no second pass.
 
 ```ts
 // Fastly: POST /service/{id}/purge, up to 256 keys a call. A soft purge

packages/bun-cache/__tests__/http/methods.test.ts

@@ -6,8 +6,8 @@
  */
 import { describe, expect, test } from "bun:test";
 import type { BunRequest, BunResponse } from "@kingsleyweb/bun-common";
-import { BunRouter } from "@kingsleyweb/bun-common";
-import type { CacheDriver } from "../../lib/index";
+import { BunRouter, createDeferred } from "@kingsleyweb/bun-common";
+import type { Cache, CacheDriver } from "../../lib/index";
 import type { CacheResponseOptions } from "../../lib/http/index";
 import { addCacheTags, cacheResponse } from "../../lib/http/index";
 import { createCache, memoryDriver } from "../../lib/index";
@@ -329,3 +329,119 @@ describe("an invalidation and a write in flight", () => {
     expect(await frameAt(cache, keyFor("/r"))).toBeUndefined();
   });
 });
+
+describe("one invalidation, one pass: the second runs only for a write that landed after", () => {
+  /** Counts `cache.invalidateTags` calls and the `invalidate` events, by tag list. */
+  function counted(cache: Cache) {
+    const calls: string[][] = [];
+    const invalidate = cache.invalidateTags.bind(cache);
+    cache.invalidateTags = async (tags, options) => {
+      calls.push([...tags]);
+      await invalidate(tags, options);
+    };
+    const events: string[][] = [];
+    cache.on("invalidate", ({ tags }) => {
+      events.push([...tags]);
+    });
+    return { calls, events };
+  }
+
+  /** One event-loop turn: every microtask queued before it has run. */
+  const nextTurn = () =>
+    new Promise<void>((resolve) => setImmediate(() => resolve()));
+
+  for (const coalesce of [false, true]) {
+    test(`an invalidation right after the write landed runs once and emits once (coalesce: ${String(coalesce)})`, async () => {
+      const cache = makeHttpCache();
+      const router = new BunRouter();
+      const pages = cacheResponse(cache, { ttl: 60_000, coalesce });
+      router.get("/r", pages, (_req, res) => {
+        res.send("r");
+      });
+      const { calls, events } = counted(cache);
+      const written = nextSet(cache);
+      await router.fetch("/r");
+      // The write has landed; its capture is still settling for a few
+      // microtasks, the window that used to run a second pass.
+      await written;
+      await pages.invalidate("/r");
+      // A second pass would have been called by now: it waits only for the
+      // first and for the write's capture, both settled within this turn.
+      await nextTurn();
+      expect(calls).toEqual([["http:path:/r"]]);
+      expect(events).toEqual([["http:path:/r"]]);
+      expect(await frameAt(cache, keyFor("/r"))).toBeUndefined();
+    });
+  }
+
+  /** A streamed body that ends when `end` settles: its write is issued at its end. */
+  function bodyEndingOn(end: Promise<void>): ReadableStream<Uint8Array> {
+    return new ReadableStream<Uint8Array>({
+      async start(controller) {
+        await end;
+        controller.enqueue(new TextEncoder().encode("r"));
+        controller.close();
+      },
+    });
+  }
+
+  test("a streamed write that lands after the invalidation began still gets the second pass", async () => {
+    const cache = makeHttpCache();
+    const router = new BunRouter();
+    const pages = cacheResponse(cache, {
+      ttl: 60_000,
+      coalesce: false,
+      streams: { buffer: 1024 },
+    });
+    // Its write is issued at the body's end, after the invalidation, so the
+    // core's fence (which refuses a write issued before an invalidation)
+    // does not cover it: only the second pass does.
+    const end = createDeferred<void>();
+    router.get("/r", pages, (_req, res) => {
+      res.send(bodyEndingOn(end.promise));
+    });
+    const { calls, events } = counted(cache);
+    const second = createDeferred<void>();
+    cache.on("invalidate", () => {
+      if (events.length === 2) {
+        second.resolve();
+      }
+    });
+    const written = nextSet(cache);
+    const first = await router.fetch("/r");
+    await pages.invalidate("/r");
+    end.resolve();
+    expect(await first.text()).toBe("r");
+    await written;
+    await second.promise;
+    expect(calls).toEqual([["http:path:/r"], ["http:path:/r"]]);
+    expect(await frameAt(cache, keyFor("/r"))).toBeUndefined();
+  });
+
+  test("a coalesced write in flight is fenced by the invalidation: nothing stored, one pass", async () => {
+    const cache = makeHttpCache();
+    const router = new BunRouter();
+    const pages = cacheResponse(cache, {
+      ttl: 60_000,
+      coalesce: true,
+      streams: { buffer: 1024 },
+    });
+    const end = createDeferred<void>();
+    router.get("/r", pages, (_req, res) => {
+      res.send(bodyEndingOn(end.promise));
+    });
+    const { calls, events } = counted(cache);
+    const fenced = createDeferred<void>();
+    cache.on("fenced", () => fenced.resolve());
+    const first = await router.fetch("/r");
+    await pages.invalidate("/r");
+    end.resolve();
+    expect(await first.text()).toBe("r");
+    // The local invalidation forgot the flight, so its write is refused.
+    await fenced.promise;
+    await nextTurn();
+    expect(calls).toEqual([["http:path:/r"]]);
+    expect(events).toEqual([["http:path:/r"]]);
+    expect(await frameAt(cache, keyFor("/r"))).toBeUndefined();
+  });
+});

packages/bun-cache/__tests__/http/surrogate.test.ts

@@ -614,12 +614,9 @@ describe("purging a CDN by the same tags (the README recipe)", () => {
       .get("surrogate-key")!
       .split(" ");
     await written;
-    // The miss's write stays tracked for a few microtasks after its `set`
-    // event, and an invalidation of the path in that window runs a second
-    // time once it settles, a fourth event. `invalidateTags` below usually
-    // outlasts the window, but not when it skips its 1 ms wait for the clock
-    // (under load the clock has often moved already): let the write settle.
-    await new Promise((resolve) => setImmediate(resolve));
+    // The write has landed, so the invalidations below run once each: no
+    // second pass, and no fourth event (methods.test.ts, "one invalidation,
+    // one pass").
     await f.cache.invalidateTags(["post:7"]);
     await f.pages.invalidate("/posts/7");
     await f.router.fetch("/posts/7", { method: "DELETE" });

packages/bun-cache/lib/http/cacheResponse.ts

@@ -415,6 +415,37 @@ interface CaptureHandle {
   onTransformed: (() => void) | undefined;
 }
 
+/** A capture's write, tracked under its path's tag until it settles. */
+interface TrackedWrite {
+  /** Settles when the write has (or `wait` has passed). */
+  settled: Promise<void>;
+  /** The `set` count at its key's last landing; `undefined` until one, infinite when its key is unknown. */
+  landedAt: number | undefined;
+}
+
+/** The set `map` holds at `key`, created empty when there is none. */
+function setIn<K, V>(map: Map<K, Set<V>>, key: K): Set<V> {
+  let set = map.get(key);
+  if (set === undefined) {
+    set = new Set();
+    map.set(key, set);
+  }
+  return set;
+}
+
+/** Removes `value` from `set`, and `set` from `map` at `key` once it is empty (and still the one there). */
+function dropFrom<K, V>(
+  map: Map<K, Set<V>>,
+  key: K,
+  set: Set<V>,
+  value: V,
+): void {
+  set.delete(value);
+  if (set.size === 0 && map.get(key) === set) {
+    map.delete(key);
+  }
+}
+
 /** A background replay a middleware is waiting on, by its id. */
 interface Replay {
   /** The base key the replayed request must produce. */
@@ -654,12 +685,49 @@ export function cacheResponse(
    * path runs again once they have settled: a plain `set` (an uncoalesced
    * miss) is fenced only against invalidations after it was issued, and a
    * streamed body's is issued when the stream ends, so without the second
-   * pass it could put back what the invalidation just dropped.
+   * pass it could put back what the invalidation just dropped. The second
+   * pass runs only when one of them stored after the invalidation began:
+   * one that had already landed was dropped by the first (its promise
+   * settles a few microtasks after the store, which once made an
+   * invalidation in that gap run, and emit `invalidate`, twice).
    */
-  const writing = new Map<string, Set<Promise<void>>>();
+  const writing = new Map<string, Set<TrackedWrite>>();
+
+  /** The tracked writes by the key they store at, for {@link landings}. */
+  const writingByKey = new Map<string, Set<TrackedWrite>>();
+
+  /**
+   * Counts the cache's `set` events for keys a tracked write stores at: the
+   * store itself, synchronous with it, so it orders a landing against the
+   * start of an invalidation exactly. Another write to the same key marks
+   * the tracked one too, which only ever adds a second pass.
+   */
+  let landings = 0;
+  /** Whether the `set` listener is on: added with the first tracked write, so building the middleware reads nothing of the cache. */
+  let watchingSets = false;
+  const watchSets = (): void => {
+    if (watchingSets) {
+      return;
+    }
+    watchingSets = true;
+    cache.on("set", ({ key }) => {
+      const writes = writingByKey.get(key);
+      if (writes === undefined) {
+        return;
+      }
+      landings++;
+      for (const write of writes) {
+        write.landedAt = landings;
+      }
+    });
+  };
 
-  /** Tracks `settled`, a capture's write, under `tag` until it settles or `wait` passes. */
-  const trackWrite = (tag: string, settled: Promise<void>): void => {
+  /** Tracks `settled`, a capture's write to `key`, under `tag` until it settles or `wait` passes. */
+  const trackWrite = (
+    tag: string,
+    key: string | undefined,
+    settled: Promise<void>,
+  ): void => {
     let timer: ReturnType<typeof setTimeout> | undefined;
     const capped = Promise.race([
       settled,
@@ -670,28 +738,35 @@ export function cacheResponse(
     ]).finally(() => {
       clearTimeout(timer);
     });
-    let writes = writing.get(tag);
-    if (writes === undefined) {
-      writes = new Set();
-      writing.set(tag, writes);
+    // Without its key a landing cannot be seen: always run the second pass.
+    const write: TrackedWrite = {
+      settled: capped,
+      landedAt: key === undefined ? Number.POSITIVE_INFINITY : undefined,
+    };
+    if (key !== undefined) {
+      watchSets();
     }
-    const own = writes;
-    own.add(capped);
+    const own = setIn(writing, tag);
+    own.add(write);
+    const byKey = key === undefined ? undefined : setIn(writingByKey, key);
+    byKey?.add(write);
     void capped.then(() => {
-      own.delete(capped);
-      if (own.size === 0 && writing.get(tag) === own) {
-        writing.delete(tag);
+      dropFrom(writing, tag, own, write);
+      if (key !== undefined && byKey !== undefined) {
+        dropFrom(writingByKey, key, byKey, write);
       }
     });
   };
 
   /**
    * Invalidates `tags`, tracked in {@link pending} until it settles; and
    * again, in the background, once the writes in flight under them have
-   * settled (see {@link writing}). The caller waits for the first pass only.
+   * settled, if one of them stored after this began (see {@link writing}).
+   * The caller waits for the first pass only.
    */
   const invalidateTracked = (tags: readonly string[]): Promise<void> => {
-    const inflight: Promise<void>[] = [];
+    const began = landings;
+    const inflight: TrackedWrite[] = [];
     for (const tag of tags) {
       for (const write of writing.get(tag) ?? []) {
         inflight.push(write);
@@ -702,9 +777,15 @@ export function cacheResponse(
       // cache's throw must not become an unhandled rejection here.
     });
     if (inflight.length > 0) {
-      void Promise.all([done, ...inflight])
+      void Promise.all([done, ...inflight.map((write) => write.settled)])
         .then(async () => {
-          await cache.invalidateTags(tags);
+          if (
+            inflight.some(
+              (write) => write.landedAt !== undefined && write.landedAt > began,
+            )
+          ) {
+            await cache.invalidateTags(tags);
+          }
         })
         .catch(() => {
           // Reported by the failure policy, as above.
@@ -839,11 +920,14 @@ export function cacheResponse(
    * a streamed body is swapped for a copying tap, whose end reaches the
    * client only once the promise `onOutcome` returns has settled (at most
    * `wait`): return the write, so a request after the end finds it.
+   * `key` is the key the write stores at, so an invalidation can tell
+   * whether it landed before it began (see {@link writing}).
    */
   const capture = (
     exchange: Exchange,
     onOutcome: (outcome: Outcome) => Promise<unknown> | undefined,
     stale?: StaleEntry,
+    key?: string,
   ): CaptureHandle => {
     const { req, res } = exchange;
     /** Settles once the outcome has been reported and what `onOutcome` returned has settled. */
@@ -891,7 +975,7 @@ export function cacheResponse(
           }
           // From here the outcome is always reported: an invalidation of
           // the path waits for what it writes.
-          trackWrite(exchange.tag, handled.promise);
+          trackWrite(exchange.tag, key, handled.promise);
           const now = Date.now();
           if (
             stale !== undefined &&
@@ -1192,6 +1276,7 @@ export function cacheResponse(
       (outcome) =>
         outcome.ok ? storeOutcome(exchange, key, variant, outcome) : undefined,
       stale,
+      key,
     );
     armHardTimeout(exchange, handle, stale);
     exchange.next();
@@ -1431,6 +1516,7 @@ export function cacheResponse(
         return undefined;
       },
       stale,
+      key,
     );
     armHardTimeout(exchange, handle, stale, () => {
       if (!settled && stale !== undefined) {
@@ -1696,10 +1782,15 @@ export function cacheResponse(
     replay.exchange = exchange;
     // No wait at a buffered stream's end here: the replay's load, which
     // holds the flight, is what reads that end.
-    replay.handle = capture(exchange, (outcome) => {
-      replay.settle(outcome);
-      return undefined;
-    });
+    replay.handle = capture(
+      exchange,
+      (outcome) => {
+        replay.settle(outcome);
+        return undefined;
+      },
+      undefined,
+      key,
+    );
     exchange.next();
   };