From bcac17e079f587dc7773ea1eada51fe473d3f305 Mon Sep 17 00:00:00 2001 From: Okafor Obinna Date: Sun, 8 Nov 2026 16:46:54 +0100 Subject: [PATCH] fix(bun-cache): one invalidation, one CDN purge cacheResponse() ran a second invalidation pass (and a second invalidate event) after every tracked write that was settling when an invalidation began, so the README's CDN-purge recipe purged twice. Each tracked write now records, through the cache's synchronous set event, whether it was stored after the invalidation began, and only then runs the second pass; a refused or already-landed write no longer does. The listener is added with the first tracked write (bun-nest's interceptor builds one against a throwing probe). Develop fails 3 of the 4 new tests. Approved by the user, 2026-11-08. --- packages/bun-cache/README.md | 17 ++- .../bun-cache/__tests__/http/methods.test.ts | 120 +++++++++++++++- .../__tests__/http/surrogate.test.ts | 9 +- packages/bun-cache/lib/http/cacheResponse.ts | 135 +++++++++++++++--- 4 files changed, 246 insertions(+), 35 deletions(-) diff --git a/packages/bun-cache/README.md b/packages/bun-cache/README.md index 4087ebe0..3d4b11b5 100644 --- a/packages/bun-cache/README.md +++ b/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 diff --git a/packages/bun-cache/__tests__/http/methods.test.ts b/packages/bun-cache/__tests__/http/methods.test.ts index 256ad333..3e0ea9f8 100644 --- a/packages/bun-cache/__tests__/http/methods.test.ts +++ b/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((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): ReadableStream { + return new ReadableStream({ + 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(); + router.get("/r", pages, (_req, res) => { + res.send(bodyEndingOn(end.promise)); + }); + const { calls, events } = counted(cache); + const second = createDeferred(); + 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(); + router.get("/r", pages, (_req, res) => { + res.send(bodyEndingOn(end.promise)); + }); + const { calls, events } = counted(cache); + const fenced = createDeferred(); + 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(); + }); +}); diff --git a/packages/bun-cache/__tests__/http/surrogate.test.ts b/packages/bun-cache/__tests__/http/surrogate.test.ts index 4bcd4c77..261f1b50 100644 --- a/packages/bun-cache/__tests__/http/surrogate.test.ts +++ b/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" }); diff --git a/packages/bun-cache/lib/http/cacheResponse.ts b/packages/bun-cache/lib/http/cacheResponse.ts index 8c4a66ee..9ee9d4ad 100644 --- a/packages/bun-cache/lib/http/cacheResponse.ts +++ b/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; + /** 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(map: Map>, key: K): Set { + 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( + map: Map>, + key: K, + set: Set, + 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>>(); + const writing = new Map>(); + + /** The tracked writes by the key they store at, for {@link landings}. */ + const writingByKey = new Map>(); + + /** + * 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 => { + /** 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 => { let timer: ReturnType | undefined; const capped = Promise.race([ settled, @@ -670,17 +738,22 @@ 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); } }); }; @@ -688,10 +761,12 @@ export function cacheResponse( /** * 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 => { - const inflight: Promise[] = []; + 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 | 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(); };