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/.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/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 21dc10aa..5491e812 100644 --- a/packages/toapi-server/src/revalidation-stream.ts +++ b/packages/toapi-server/src/revalidation-stream.ts @@ -1,22 +1,33 @@ -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 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; let unsubscribe = () => {}; const stream = new ReadableStream({ async start(controller) { + let queue = new Set(); const textEncoder = new TextEncoder(); + // subscribe to tag invalidations unsubscribe = cache.subscribe((tags, meta) => { // ignore our own invalidations @@ -27,8 +38,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; + }, throttleTimeout); + } }); // keepalive. The first one is sent right away so the response headers @@ -36,10 +63,11 @@ 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); + if (timeout) clearTimeout(timeout); unsubscribe(); }, });