diff --git a/packages/static/src/rsc-client/clientWrapper.tsx b/packages/static/src/rsc-client/clientWrapper.tsx index cb689ec..59b948d 100644 --- a/packages/static/src/rsc-client/clientWrapper.tsx +++ b/packages/static/src/rsc-client/clientWrapper.tsx @@ -2,7 +2,7 @@ import React from "react"; import { createFromFetch } from "@vitejs/plugin-rsc/browser"; import { getModulePathFor } from "../rsc/rscModule"; import { createContext, use } from "react"; -import type { LoadedDeferEntry, DeferRegistry } from "../rsc/defer"; +import type { LoadedDeferEntry, DeferRegistry } from "../rsc/deferRegistry"; import { withBasePath } from "../util/basePath"; interface DeferContextValue { diff --git a/packages/static/src/rsc/defer.tsx b/packages/static/src/rsc/defer.tsx index 82e56a5..1626861 100644 --- a/packages/static/src/rsc/defer.tsx +++ b/packages/static/src/rsc/defer.tsx @@ -1,16 +1,10 @@ import type { ReactElement, ReactNode } from "react"; import { renderToReadableStream } from "@vitejs/plugin-rsc/react/rsc"; import { DeferredComponent } from "#rsc-client"; -import { drainStream } from "../util/drainStream"; +import { DeferRegistry } from "./deferRegistry"; import { getPayloadIDFor } from "./rscModule"; import { rscPayloadDir } from "virtual:funstack/config"; -export interface DeferEntry { - state: DeferEntryState; - name?: string; - drainPromise?: Promise; -} - /** * Options for the defer function. */ @@ -23,28 +17,6 @@ export interface DeferOptions { name?: string; } -export interface LoadedDeferEntry extends DeferEntry { - state: Exclude; - drainPromise: Promise; -} - -type DeferEntryState = - | { - state: "pending"; - element: ReactElement; - } - | { - state: "streaming"; - stream: ReadableStream; - } - | { - state: "ready"; - } - | { - state: "error"; - error: unknown; - }; - /** * Sanitizes a name for use in file paths. * Replaces non-alphanumeric characters with underscores and limits length. @@ -57,125 +29,9 @@ function sanitizeName(name: string): string { .slice(0, 50); } -export class DeferRegistry { - #registry = new Map(); - - register(element: ReactElement, id: string, name?: string) { - this.#registry.set(id, { state: { element, state: "pending" }, name }); - } - - load(id: string): LoadedDeferEntry | undefined { - const entry = this.#registry.get(id); - if (!entry) { - return undefined; - } - return this.#loadEntry(entry); - } - - #loadEntry(entry: DeferEntry): LoadedDeferEntry { - const { state } = entry; - switch (state.state) { - case "pending": { - const stream = renderToReadableStream(state.element); - const [stream1, stream2] = stream.tee(); - entry.state = { state: "streaming", stream: stream1 }; - const drainPromise = drainStream(stream2); - entry.drainPromise = drainPromise; - drainPromise.then( - () => { - entry.state = { state: "ready" }; - }, - (error) => { - entry.state = { state: "error", error }; - }, - ); - return entry as LoadedDeferEntry; - } - case "streaming": - case "ready": - case "error": - return entry as LoadedDeferEntry; - } - } - - has(id: string): boolean { - return this.#registry.has(id); - } - - /** - * Iterates over all entries in parallel. - * Yields results as each stream completes. - * - * Rendering a deferred element may itself call `defer()` (nested defer), - * registering new entries while earlier ones are still draining. Entries - * registered mid-iteration are picked up too, until none are left. - */ - async *loadAll() { - const errors: unknown[] = []; - - type Result = { id: string; data: string; name?: string }; - - // Completion queue - const completed: Array = []; - let waiting: (() => void) | undefined; - let remainingCount = 0; - const started = new Set(); - - // Start loading every entry not started yet and track its drain promise. - // We use drain promises (which drain stream2 from tee) instead of - // draining stream1 directly, because stream1 may have been locked - // by createFromReadableStream during SSR. - const startPending = () => { - for (const [id, entry] of this.#registry) { - if (started.has(id)) continue; - started.add(id); - const loaded = this.#loadEntry(entry); - remainingCount++; - loaded.drainPromise.then( - (data) => { - completed.push({ id, data, name: entry.name }); - remainingCount--; - waiting?.(); - }, - (error) => { - completed.push({ error }); - remainingCount--; - waiting?.(); - }, - ); - } - }; - - startPending(); - - // Yield from queue as results arrive - while (remainingCount > 0 || completed.length > 0) { - if (completed.length === 0) { - await new Promise((r) => { - waiting = r; - }); - waiting = undefined; - } - for (const result of completed.splice(0)) { - if ("error" in result) { - errors.push(result.error); - } else { - yield result; - } - } - // A drained entry may have registered nested entries during its - // render; any registration happens before its parent's drain promise - // resolves, so once remainingCount hits 0 no new entries can appear. - startPending(); - } - - if (errors.length > 0) { - throw new AggregateError(errors); - } - } -} - -export const deferRegistry = new DeferRegistry(); +export const deferRegistry = new DeferRegistry((element) => + renderToReadableStream(element), +); /** * Renders given Server Component into a separate RSC payload. diff --git a/packages/static/src/rsc/deferRegistry.test.ts b/packages/static/src/rsc/deferRegistry.test.ts new file mode 100644 index 0000000..5f190d9 --- /dev/null +++ b/packages/static/src/rsc/deferRegistry.test.ts @@ -0,0 +1,189 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { createElement } from "react"; +import type { ReactElement } from "react"; +import { DeferRegistry, devDeferEvictionOptions } from "./deferRegistry"; + +const { ttlMs } = devDeferEvictionOptions; + +const encoder = new TextEncoder(); + +/** + * Fake renderer that emits the element's `data-payload` prop as the stream + * content, so tests can verify what was rendered without the RSC runtime. + */ +function fakeRender(element: ReactElement): ReadableStream { + const payload = String( + (element.props as { "data-payload"?: string })["data-payload"] ?? "", + ); + return new ReadableStream({ + start(controller) { + controller.enqueue(encoder.encode(payload)); + controller.close(); + }, + }); +} + +function element(payload: string): ReactElement { + return createElement("div", { "data-payload": payload }); +} + +describe("DeferRegistry", () => { + let registry: DeferRegistry; + + beforeEach(() => { + vi.useFakeTimers(); + registry = new DeferRegistry(fakeRender); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("returns undefined for unknown ids", () => { + expect(registry.load("nope")).toBeUndefined(); + }); + + it("loads a registered entry and drains its payload", async () => { + registry.register(element("hello"), "id1"); + const entry = registry.load("id1"); + expect(entry).toBeDefined(); + expect(entry?.state.state).toBe("streaming"); + await expect(entry?.drainPromise).resolves.toBe("hello"); + expect(entry?.state.state).toBe("ready"); + }); + + describe("evictStale", () => { + it("never evicts pending entries by time", () => { + // A pending entry may be fetched arbitrarily late, e.g. deferred + // content inside an accordion that is rarely opened. + registry.register(element("accordion"), "a"); + vi.advanceTimersByTime(ttlMs * 100); + + registry.evictStale(devDeferEvictionOptions); + + expect(registry.has("a")).toBe(true); + }); + + it("evicts settled entries older than the TTL and keeps fresh ones", async () => { + registry.register(element("old"), "old"); + await registry.load("old")?.drainPromise; + vi.advanceTimersByTime(ttlMs + 1); + registry.register(element("fresh"), "fresh"); + await registry.load("fresh")?.drainPromise; + + registry.evictStale(devDeferEvictionOptions); + + expect(registry.has("old")).toBe(false); + expect(registry.has("fresh")).toBe(true); + }); + + it("keeps settled entries within the TTL", async () => { + registry.register(element("a"), "a"); + await registry.load("a")?.drainPromise; + vi.advanceTimersByTime(ttlMs); + + registry.evictStale(devDeferEvictionOptions); + + expect(registry.has("a")).toBe(true); + }); + + it("treats a load as an access that refreshes the TTL", async () => { + registry.register(element("a"), "a"); + await registry.load("a")?.drainPromise; + vi.advanceTimersByTime(ttlMs - 1); + registry.load("a"); + vi.advanceTimersByTime(ttlMs - 1); + + registry.evictStale(devDeferEvictionOptions); + expect(registry.has("a")).toBe(true); + + vi.advanceTimersByTime(2); + registry.evictStale(devDeferEvictionOptions); + expect(registry.has("a")).toBe(false); + }); + + it("does not break an already-loaded entry when it is evicted", async () => { + registry.register(element("in-flight"), "a"); + const entry = registry.load("a"); + + vi.advanceTimersByTime(ttlMs + 1); + registry.evictStale(devDeferEvictionOptions); + expect(registry.has("a")).toBe(false); + + // The response holding the entry can still drain it. + await expect(entry?.drainPromise).resolves.toBe("in-flight"); + }); + + it("caps pending entries, evicting the oldest first", () => { + registry.register(element("p1"), "p1"); + registry.register(element("p2"), "p2"); + registry.register(element("p3"), "p3"); + + registry.evictStale({ ttlMs, maxPending: 2 }); + + expect(registry.has("p1")).toBe(false); + expect(registry.has("p2")).toBe(true); + expect(registry.has("p3")).toBe(true); + }); + + it("does not count settled entries toward the pending cap", async () => { + registry.register(element("s1"), "s1"); + await registry.load("s1")?.drainPromise; + registry.register(element("p1"), "p1"); + registry.register(element("p2"), "p2"); + + registry.evictStale({ ttlMs, maxPending: 2 }); + + expect(registry.has("s1")).toBe(true); + expect(registry.has("p1")).toBe(true); + expect(registry.has("p2")).toBe(true); + }); + }); + + describe("loadAll", () => { + it("yields all registered entries", async () => { + registry.register(element("one"), "id1", "first"); + registry.register(element("two"), "id2"); + + const results = []; + for await (const result of registry.loadAll()) { + results.push(result); + } + + results.sort((a, b) => a.id.localeCompare(b.id)); + expect(results).toEqual([ + { id: "id1", data: "one", name: "first" }, + { id: "id2", data: "two", name: undefined }, + ]); + }); + + it("picks up entries registered while draining (nested defer)", async () => { + const nested = element("nested"); + const parent: ReactElement = createElement("div", { + "data-payload": "parent", + }); + const renderWithNested = (el: ReactElement) => { + if (el === parent) { + return new ReadableStream({ + start(controller) { + // Simulates a nested defer() call during the parent's render. + registry.register(nested, "nested-id"); + controller.enqueue(encoder.encode("parent")); + controller.close(); + }, + }); + } + return fakeRender(el); + }; + registry = new DeferRegistry(renderWithNested); + registry.register(parent, "parent-id"); + + const ids = []; + for await (const result of registry.loadAll()) { + ids.push(result.id); + } + + expect(ids.sort()).toEqual(["nested-id", "parent-id"]); + }); + }); +}); diff --git a/packages/static/src/rsc/deferRegistry.ts b/packages/static/src/rsc/deferRegistry.ts new file mode 100644 index 0000000..46e0a87 --- /dev/null +++ b/packages/static/src/rsc/deferRegistry.ts @@ -0,0 +1,236 @@ +import type { ReactElement } from "react"; +import { drainStream } from "../util/drainStream"; + +export interface DeferEntry { + state: DeferEntryState; + name?: string; + drainPromise?: Promise; + /** + * Timestamp (ms) of the last registration or load of this entry. + * Used by `evictStale` to drop entries no longer reachable by any client. + */ + lastAccessedAt: number; +} + +export interface LoadedDeferEntry extends DeferEntry { + state: Exclude; + drainPromise: Promise; +} + +type DeferEntryState = + | { + state: "pending"; + element: ReactElement; + } + | { + state: "streaming"; + stream: ReadableStream; + } + | { + state: "ready"; + } + | { + state: "error"; + error: unknown; + }; + +/** + * Renders a React element to an RSC payload stream. + * Injected so the registry does not depend on the RSC runtime directly. + */ +export type RenderToStream = ( + element: ReactElement, +) => ReadableStream; + +export interface EvictStaleOptions { + /** + * How long a settled (streaming/ready/error) entry is kept after it was + * last loaded. Once served, the client caches the payload per module ID + * and never legitimately re-fetches it, so a short TTL is safe. + */ + ttlMs: number; + /** + * Maximum number of pending entries to keep (oldest evicted first). + * Pending entries are never evicted by time: a DeferredComponent may + * fetch its payload arbitrarily late (e.g. content inside an accordion + * that is rarely opened), so they stay until enough newer registrations + * push them out. + */ + maxPending: number; +} + +/** + * Eviction policy for the dev server. Each dev render registers fresh + * entries (with new IDs), so old entries become unreachable once the + * client re-renders; without eviction the registry grows unboundedly + * over a dev session. + */ +export const devDeferEvictionOptions: EvictStaleOptions = { + ttlMs: 5 * 60 * 1000, + maxPending: 1000, +}; + +export class DeferRegistry { + #registry = new Map(); + #render: RenderToStream; + + constructor(render: RenderToStream) { + this.#render = render; + } + + register(element: ReactElement, id: string, name?: string) { + this.#registry.set(id, { + state: { element, state: "pending" }, + name, + lastAccessedAt: Date.now(), + }); + } + + load(id: string): LoadedDeferEntry | undefined { + const entry = this.#registry.get(id); + if (!entry) { + return undefined; + } + entry.lastAccessedAt = Date.now(); + return this.#loadEntry(entry); + } + + #loadEntry(entry: DeferEntry): LoadedDeferEntry { + const { state } = entry; + switch (state.state) { + case "pending": { + const stream = this.#render(state.element); + const [stream1, stream2] = stream.tee(); + entry.state = { state: "streaming", stream: stream1 }; + const drainPromise = drainStream(stream2); + entry.drainPromise = drainPromise; + drainPromise.then( + () => { + entry.state = { state: "ready" }; + }, + (error) => { + entry.state = { state: "error", error }; + }, + ); + return entry as LoadedDeferEntry; + } + case "streaming": + case "ready": + case "error": + return entry as LoadedDeferEntry; + } + } + + has(id: string): boolean { + return this.#registry.has(id); + } + + /** + * Drops entries that are no longer expected to be fetched. Called from + * dev server request handlers to keep the registry from growing + * unboundedly across renders; never called during a build. + * + * Settled entries (which retain the rendered payload string) are dropped + * once they have not been loaded within `ttlMs`. Pending entries (which + * retain only the React element) are exempt from the TTL — deferred + * content may be fetched arbitrarily late — and are instead capped at + * `maxPending`, evicting the oldest first. + * + * Evicting an entry does not cancel an in-flight render: responses + * already holding the entry's stream or drain promise are unaffected. + */ + evictStale({ ttlMs, maxPending }: EvictStaleOptions): void { + const now = Date.now(); + let pendingCount = 0; + for (const entry of this.#registry.values()) { + if (entry.state.state === "pending") { + pendingCount++; + } + } + // Map iteration is in insertion order, so the oldest pending + // entries are encountered (and evicted) first. + let pendingToEvict = Math.max(0, pendingCount - maxPending); + for (const [id, entry] of this.#registry) { + if (entry.state.state === "pending") { + if (pendingToEvict > 0) { + this.#registry.delete(id); + pendingToEvict--; + } + } else if (now - entry.lastAccessedAt > ttlMs) { + this.#registry.delete(id); + } + } + } + + /** + * Iterates over all entries in parallel. + * Yields results as each stream completes. + * + * Rendering a deferred element may itself call `defer()` (nested defer), + * registering new entries while earlier ones are still draining. Entries + * registered mid-iteration are picked up too, until none are left. + */ + async *loadAll() { + const errors: unknown[] = []; + + type Result = { id: string; data: string; name?: string }; + + // Completion queue + const completed: Array = []; + let waiting: (() => void) | undefined; + let remainingCount = 0; + const started = new Set(); + + // Start loading every entry not started yet and track its drain promise. + // We use drain promises (which drain stream2 from tee) instead of + // draining stream1 directly, because stream1 may have been locked + // by createFromReadableStream during SSR. + const startPending = () => { + for (const [id, entry] of this.#registry) { + if (started.has(id)) continue; + started.add(id); + const loaded = this.#loadEntry(entry); + remainingCount++; + loaded.drainPromise.then( + (data) => { + completed.push({ id, data, name: entry.name }); + remainingCount--; + waiting?.(); + }, + (error) => { + completed.push({ error }); + remainingCount--; + waiting?.(); + }, + ); + } + }; + + startPending(); + + // Yield from queue as results arrive + while (remainingCount > 0 || completed.length > 0) { + if (completed.length === 0) { + await new Promise((r) => { + waiting = r; + }); + waiting = undefined; + } + for (const result of completed.splice(0)) { + if ("error" in result) { + errors.push(result.error); + } else { + yield result; + } + } + // A drained entry may have registered nested entries during its + // render; any registration happens before its parent's drain promise + // resolves, so once remainingCount hits 0 no new entries can appear. + startPending(); + } + + if (errors.length > 0) { + throw new AggregateError(errors); + } + } +} diff --git a/packages/static/src/rsc/entry.tsx b/packages/static/src/rsc/entry.tsx index 24d6203..f3361d0 100644 --- a/packages/static/src/rsc/entry.tsx +++ b/packages/static/src/rsc/entry.tsx @@ -3,6 +3,7 @@ import { renderToReadableStream } from "@vitejs/plugin-rsc/rsc"; import { devMainRscPath } from "./request"; import { generateAppMarker } from "./marker"; import { deferRegistry } from "./defer"; +import { devDeferEvictionOptions } from "./deferRegistry"; import { extractIDFromModulePath } from "./rscModule"; import { stripBasePath } from "../util/basePath"; import { urlPathToFileCandidates } from "../util/urlPath"; @@ -145,6 +146,10 @@ async function renderEntryToResponse( * Accepts a Request to determine which entry to render based on URL path. */ export async function serveHTML(request: Request): Promise { + // Each dev render registers fresh defer entries; drop stale ones so the + // registry does not grow unboundedly over a long dev session (#144). + deferRegistry.evictStale(devDeferEvictionOptions); + const timings: string[] = []; const entriesStart = performance.now(); @@ -182,6 +187,8 @@ export function isServeRSCError(error: unknown): error is ServeRSCError { * Serves an RSC stream response */ export async function serveRSC(request: Request): Promise { + deferRegistry.evictStale(devDeferEvictionOptions); + const timings: string[] = []; const url = new URL(request.url); const pathname = stripBasePath(url.pathname); diff --git a/packages/static/src/ssr/entry.tsx b/packages/static/src/ssr/entry.tsx index 6e1d45b..fe9ba35 100644 --- a/packages/static/src/ssr/entry.tsx +++ b/packages/static/src/ssr/entry.tsx @@ -7,7 +7,7 @@ import type { RscPayload } from "../rsc/entry"; import { appClientManifestVar } from "../client/globals"; import { rscPayloadPlaceholder } from "../build/rscPath"; import { preload } from "react-dom"; -import type { DeferRegistry } from "../rsc/defer"; +import type { DeferRegistry } from "../rsc/deferRegistry"; import { RegistryContext } from "#rsc-client"; export async function renderHTML(