From f315ff5bd52c9bcdf0098ab2eea05a098d53dd14 Mon Sep 17 00:00:00 2001 From: Michel Smola Date: Thu, 24 Sep 2026 11:38:03 +0200 Subject: [PATCH 1/3] fix: debounce revalidation stream --- .changeset/two-worms-take.md | 5 ++++ .../toapi-server/src/revalidation-stream.ts | 27 +++++++++++++++---- 2 files changed, 27 insertions(+), 5 deletions(-) create mode 100644 .changeset/two-worms-take.md diff --git a/.changeset/two-worms-take.md b/.changeset/two-worms-take.md new file mode 100644 index 00000000..ef42b249 --- /dev/null +++ b/.changeset/two-worms-take.md @@ -0,0 +1,5 @@ +--- +"@toapi/server": patch +--- + +debounce and dedupe tags stream diff --git a/packages/toapi-server/src/revalidation-stream.ts b/packages/toapi-server/src/revalidation-stream.ts index 21dc10aa..3de759e6 100644 --- a/packages/toapi-server/src/revalidation-stream.ts +++ b/packages/toapi-server/src/revalidation-stream.ts @@ -1,10 +1,8 @@ -import { - SESSION_COOKIE_NAME, - TAGS_CONTENT_TYPE, -} from "@toapi/common"; +import { SESSION_COOKIE_NAME, TAGS_CONTENT_TYPE } from "@toapi/common"; import type { Cache } from "./cache.js"; const KEEPALIVE_INTERVAL = 10 * 1000; +const THROTTLE_TIME_MS = 500; interface Options { cache: Cache; @@ -16,7 +14,10 @@ export function streamRevalidatedTags({ cache }: Options) { let unsubscribe = () => {}; const stream = new ReadableStream({ async start(controller) { + let queue = new Set(); + let timeout: ReturnType | null = null; const textEncoder = new TextEncoder(); + // subscribe to tag invalidations unsubscribe = cache.subscribe((tags, meta) => { // ignore our own invalidations @@ -27,8 +28,24 @@ export function streamRevalidatedTags({ cache }: Options) { meta.clientId === id ) return; + + for (const tag of tags) queue.add(tag); + // send tags to client - controller.enqueue(textEncoder.encode(`${tags.join(" ")}\n`)); + if (!timeout) { + controller.enqueue( + textEncoder.encode(`${Array.from(queue).join(" ")}\n`), + ); + queue = new Set(); + + timeout = setTimeout(() => { + controller.enqueue( + textEncoder.encode(`${Array.from(queue).join(" ")}\n`), + ); + queue = new Set(); + timeout = null; + }, THROTTLE_TIME_MS); + } }); // keepalive. The first one is sent right away so the response headers From 4f32f63d242f376a8504396420ff153852ea6dca Mon Sep 17 00:00:00 2001 From: Michel Smola Date: Fri, 25 Sep 2026 12:30:20 +0200 Subject: [PATCH 2/3] fix: clear timeouts in streamRevalidatedTags --- .github/workflows/ci.yml | 2 +- packages/toapi-server/src/revalidation-stream.ts | 3 ++- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index becc12d1..5b7e1529 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -53,5 +53,5 @@ jobs: - run: pnpm install - run: pnpm run lint - run: pnpm run build - - run: pnpm exec playwright install --with-deps + - run: pnpm exec playwright install --with-deps --no-progress - run: pnpm run test diff --git a/packages/toapi-server/src/revalidation-stream.ts b/packages/toapi-server/src/revalidation-stream.ts index 3de759e6..3baf895c 100644 --- a/packages/toapi-server/src/revalidation-stream.ts +++ b/packages/toapi-server/src/revalidation-stream.ts @@ -11,11 +11,11 @@ interface Options { export function streamRevalidatedTags({ cache }: Options) { const id = crypto.randomUUID(); let interval: ReturnType | null = null; + let timeout: ReturnType | null = null; let unsubscribe = () => {}; const stream = new ReadableStream({ async start(controller) { let queue = new Set(); - let timeout: ReturnType | null = null; const textEncoder = new TextEncoder(); // subscribe to tag invalidations @@ -57,6 +57,7 @@ export function streamRevalidatedTags({ cache }: Options) { }, cancel() { if (interval) clearInterval(interval); + if (timeout) clearTimeout(timeout); unsubscribe(); }, }); From 7f6501838ba355b9a5c1b6b2779776b4e652fe70 Mon Sep 17 00:00:00 2001 From: Michel Smola Date: Fri, 25 Sep 2026 12:45:20 +0200 Subject: [PATCH 3/3] feat: make timeout magic numbers configurable --- .../src/create-request-handler.ts | 1 + packages/toapi-server/src/define-api.ts | 12 +++++++++-- .../toapi-server/src/revalidation-stream.ts | 20 ++++++++++++++----- 3 files changed, 26 insertions(+), 7 deletions(-) diff --git a/packages/toapi-server/src/create-request-handler.ts b/packages/toapi-server/src/create-request-handler.ts index c3b9c9f8..3f7c8cc8 100644 --- a/packages/toapi-server/src/create-request-handler.ts +++ b/packages/toapi-server/src/create-request-handler.ts @@ -59,6 +59,7 @@ export function createRequestHandler( if (url.pathname === `${basePath}${INVALIDATIONS_ROUTE}`) { return streamRevalidatedTags({ cache: api.cache, + config: api.revalidationStreamConfig, }); } diff --git a/packages/toapi-server/src/define-api.ts b/packages/toapi-server/src/define-api.ts index 7fbc45c0..1995d5f1 100644 --- a/packages/toapi-server/src/define-api.ts +++ b/packages/toapi-server/src/define-api.ts @@ -3,6 +3,7 @@ import type { Logger } from "@toapi/common"; import type { Path as BasePath, StrictParams } from "@toapi/common"; import type { Route } from "@toapi/common"; import { type Cache, PubSub } from "./cache.js"; +import type { RevalidationStreamConfig } from "./revalidation-stream.js"; export interface OasInfo { title: string; @@ -13,10 +14,16 @@ interface Options { cache?: Cache; oas?: OasInfo; logger?: Logger; + revalidationStream?: RevalidationStreamConfig; } export function defineApi(options: Options = {}) { - return new ApiDefinition({}, options?.cache ?? new PubSub(), options?.oas, options?.logger); + return new ApiDefinition( + {}, + options?.cache ?? new PubSub(), + options?.oas, + options?.logger, + ); } export class ApiDefinition> { @@ -25,10 +32,11 @@ export class ApiDefinition> { public cache: Cache, public oas?: OasInfo, public logger?: Logger, + public revalidationStreamConfig?: RevalidationStreamConfig, ) {} async invalidate(tags: string[]) { - await this.cache.delete(tags) + await this.cache.delete(tags); } route< diff --git a/packages/toapi-server/src/revalidation-stream.ts b/packages/toapi-server/src/revalidation-stream.ts index 3baf895c..5491e812 100644 --- a/packages/toapi-server/src/revalidation-stream.ts +++ b/packages/toapi-server/src/revalidation-stream.ts @@ -1,14 +1,24 @@ import { SESSION_COOKIE_NAME, TAGS_CONTENT_TYPE } from "@toapi/common"; import type { Cache } from "./cache.js"; -const KEEPALIVE_INTERVAL = 10 * 1000; -const THROTTLE_TIME_MS = 500; +const DEFAULT_KEEPALIVE_INTERVAL = 10 * 1000; +const DEFAULT_THROTTLE_TIMEOUT = 500; + +export interface RevalidationStreamConfig { + throttleTimeout?: number; + keepaliveInterval?: number; +} interface Options { cache: Cache; + config?: RevalidationStreamConfig; } -export function streamRevalidatedTags({ cache }: Options) { +export function streamRevalidatedTags({ cache, config = {} }: Options) { + const { + throttleTimeout = DEFAULT_THROTTLE_TIMEOUT, + keepaliveInterval = DEFAULT_KEEPALIVE_INTERVAL, + } = config; const id = crypto.randomUUID(); let interval: ReturnType | null = null; let timeout: ReturnType | null = null; @@ -44,7 +54,7 @@ export function streamRevalidatedTags({ cache }: Options) { ); queue = new Set(); timeout = null; - }, THROTTLE_TIME_MS); + }, throttleTimeout); } }); @@ -53,7 +63,7 @@ export function streamRevalidatedTags({ cache }: Options) { // fires — consumers skip empty lines, so this is a no-op for them. const keepalive = () => controller.enqueue(textEncoder.encode("\n")); keepalive(); - interval = setInterval(keepalive, KEEPALIVE_INTERVAL); + interval = setInterval(keepalive, keepaliveInterval); }, cancel() { if (interval) clearInterval(interval);