diff --git a/.changeset/quiet-queries-stream.md b/.changeset/quiet-queries-stream.md new file mode 100644 index 00000000000..6e16379f775 --- /dev/null +++ b/.changeset/quiet-queries-stream.md @@ -0,0 +1,5 @@ +--- +'@tanstack/router-ssr-query-core': patch +--- + +Dehydrate only queries for SSR transport, preserve pending-query promises, batch render-time query streaming, and clean up request-owned QueryClient resources. diff --git a/e2e/react-start/streaming-ssr/tests/query-heavy.spec.ts b/e2e/react-start/streaming-ssr/tests/query-heavy.spec.ts index 5e90d1f527b..8e735040c2e 100644 --- a/e2e/react-start/streaming-ssr/tests/query-heavy.spec.ts +++ b/e2e/react-start/streaming-ssr/tests/query-heavy.spec.ts @@ -147,4 +147,22 @@ test.describe('Query heavy route (9 useSuspenseQuery)', () => { expect(html.slice(lastScriptOpen, endIndex)).toContain('.return(void 0)') expect(endIndex).toBeLessThan(html.indexOf('')) }) + + test('batches same-turn queries into one stream chunk', async ({ + request, + }) => { + const response = await request.get('/query-heavy') + const html = await response.text() + const scripts = Array.from( + html.matchAll(/]*>([\s\S]*?)<\/script>/g), + (match) => match[1]!, + ) + const syncQueryChunk = scripts.find( + (script) => script.includes('.next(') && script.includes('sync-value-1'), + ) + + expect(syncQueryChunk).toBeDefined() + expect(syncQueryChunk).toContain('sync-value-2') + expect(syncQueryChunk).toContain('sync-value-3') + }) }) diff --git a/packages/router-ssr-query-core/INTERNALS.md b/packages/router-ssr-query-core/INTERNALS.md new file mode 100644 index 00000000000..7439cb10aab --- /dev/null +++ b/packages/router-ssr-query-core/INTERNALS.md @@ -0,0 +1,456 @@ +# Internals of SSR Query Integration + +## Purpose + +This package connects server-side rendering (SSR) in TanStack Router to a TanStack Query `QueryClient`. + +The integration sends the initial Query state with Router dehydration data. It sends render-time queries through a `ReadableStream`. + +The React, Solid, and Vue packages call the same core function. This document gives details of that shared implementation. + +Query Core details refer to version `5.102.0`, which this workspace installs. A newer version of Query Core can change these details. + +The package accepts `@tanstack/query-core >=5.102.0`. + +## Supported Lifecycle + +The supported lifecycle has this order: + +1. The application creates a Router and a request-owned `QueryClient`. +2. The application calls `setupCoreRouterSsrQueryIntegration`. +3. Router attaches the server SSR utilities. +4. Router loads the routes. +5. The request handler calls `serverSsr.dehydrate()`. +6. Seroval starts to serialize Router data, Query data, promises, and streams. +7. The framework renders the application. +8. A framework adapter or Router stream transform reports render completion. +9. Router releases the resources for the request. + +The supported use requires integration setup before Router attaches the server SSR utilities. The usual `getRouter()` flow obeys this requirement. + +The one-time `onServerSsrAttach` event lets the integration register Query cleanup for the request. + +Standard Router and Start handlers call `serverSsr.dehydrate()` one time. Router enforces this lifecycle and rejects a second completed call. + +Direct dehydration calls occur sequentially in the supported lifecycle. + +## Transport Shape + +The integration adds one `query` property to the Router dehydration data: + +```ts +type DehydratedRouterQueryState = { + query: { + initial?: Array + stream: ReadableStream> + } +} +``` + +The `query` property groups the Query transport data. The `stream` property is always present in data that the client receives. + +When available, `initial` contains an array of queries from initial dehydration. + +The integration adds `initial` only when initial dehydration selects one or more queries. + +The transport contains Query entries exclusively. + +`RouterSsrQueryOptions.dehydrateOptions` keeps the `DehydrateOptions` type from Query Core. + +The transport uses `shouldDehydrateQuery`, `serializeData`, and `shouldRedactErrors` from these options. + +Each initial entry and each stream entry has the `DehydratedQuery` shape. + +Thus, the initial payload and all stream chunks use the same query data shape. + +Router serializes the stream and pending promises with Seroval. The browser receives a separate reconstructed `ReadableStream`. + +Server request cleanup controls the lifetime of the server stream. + +## Server Setup + +The server branch keeps the dehydration function that Router already has. The integration calls this function first. + +Then the integration reads the Query cache and calls `dehydrateQuery` for each selected query. + +Query writes from the Router function can enter the initial Query state. + +The integration adds an `onServerSsrAttach` listener. This listener registers Query cleanup before Router loads the routes. + +A redirect or loader error before dehydration can stop the request. The cleanup listener clears the request-owned `QueryClient`. + +`shouldDehydrateAllQueries` is the fallback Query filter. + +This filter selects every Query status, including `pending`. An explicit filter or a Query filter in the current defaults replaces this fallback. + +## Initial Dehydration + +Initial dehydration uses the usual precedence of TanStack Query options: + +- The explicit `dehydrateOptions.shouldDehydrateQuery` filter +- The Query filter in the default options at that time +- The `shouldDehydrateAllQueries` fallback. + +The integration uses the first available filter. + +The integration selects this filter after it awaits the Router dehydration function. Thus, changes to Query defaults in that function apply. + +At the same time, the integration selects `serializeData` and `shouldRedactErrors`. Explicit options have priority over the current Query defaults. + +The integration uses these selected options for the initial queries and the Query stream. + +The initial Query state can contain completed queries and pending queries. A selected pending query can contain an active retryer promise. + +For every selected pending query, `dehydrateQuery` adds the `promise` property. + +An active retryer supplies the property value. An inactive retryer supplies `undefined`. + +The integration calls `dehydrateQuery` for each selected query. This preserves query state, pending promises, `meta`, and `queryType`. + +The initial loop records each selected query hash in `sentQueries`. + +The integration associates each hash with its first transported query version for the request. + +After the initial Query state exists, the integration creates the Query stream. Then it subscribes before framework rendering starts. + +When one or more initial queries exist, the returned `query.initial` contains the initial Query state. + +The selected payload contains Query state and can contain `meta` and `queryType`. + +The application configures general Query options and cache configuration separately. + +The stream remains present for render-time queries. + +## Stream State + +`QueryStreamState` owns these active server resources: + +- The stream controller +- The set of sent query hashes +- The function that unsubscribes from the Query cache +- The optional pending-query map. + +`streamState` contains one full `QueryStreamState` object or `undefined`. The `undefined` value represents an inactive Query stream. + +This type keeps all active stream resources together. The controller and unsubscribe function always exist together. + +`finalizeQueryStream` sets `streamState` to `undefined`. Then it unsubscribes from the Query cache. + +If `finalizeQueryStream` receives an error, the function puts the stream in an error state. Otherwise, the function closes the stream. + +A cancelled stream is already terminal. Finalization still removes the Query cache subscription and releases the stream state. + +A queued microtask sees the inactive state after finalization and exits. + +## Query Selection + +The Query cache can emit multiple events for one query. The integration stores one `Query` reference for each hash. + +The pending-query map removes duplicate events in one batch. The set of sent query hashes removes duplicate transport across all payloads. + +Query can install its promise after the first cache event. + +The next applicable event exposes `event.query.promise`. The integration then adds the query to the pending-query map. + +The stream path dehydrates the stored `Query` references directly. Its work is proportional to the selected render-time queries. + +A larger Query cache increases the work that direct dehydration of stored `Query` references avoids. + +## Stream Filter + +The stream path uses the `shouldDehydrateQuery` filter that initial dehydration selected. + +The stream path uses the `serializeData` and `shouldRedactErrors` values that initial dehydration selected. + +The selected filter, serialization function, and error-redaction function stay fixed for the stream lifetime. + +Query Core uses `shouldRedactErrors` for a dehydrated pending promise that later rejects. Existing `state.error` values retain their original value. + +## Batching + +The first eligible event creates the pending-query map. The integration queues one microtask for that map. + +More applicable events can occur before the microtask operates. These events enter the same map and the same stream chunk. + +The microtask dehydrates the queries before a timer callback runs. + +Thus, dehydration occurs before React continues the related Suspense boundaries. + +The React E2E suite examines final server-origin values and Query payload placement before the Router serialization-end marker. + +The core unit test examines array batching directly. + +`flushPendingQueries` removes the pending-query map first. Then the function dehydrates the stored queries. + +A query that settles after this flush can create a new pending-query map. This map produces a new stream chunk. + +`flushPendingQueries` sends one array when one or more queries pass the stream filter. + +## Render Completion + +The integration registers `finishRendering` with `serverSsr.onRenderFinished`. A framework adapter or Router stream transform reports render completion. + +`finishRendering` calls `flushPendingQueries` synchronously and supersedes the scheduled microtask. + +Then `finishRendering` removes the Query cache subscription. Finally, it closes the Query stream. + +If streamed dehydration fails, the integration removes the subscription. Then the integration puts the stream in an error state. + +Request cleanup clears the `QueryClient` after render completion. + +Seroval serialization can continue after rendering completes. A transformed streaming response waits for application output and Router serialization. + +This streaming path can transport Query output that Seroval produces after render completion. + +Built-in string renderers use the HTML buffered at render completion. Then they release the request resources. + +Transport output depends on filters, first-hash selection, errors, aborts, and lifetime limits. + +## Request Cleanup + +Request cleanup sets `cleanedUp` to `true`. Then cleanup removes the Query cache subscription. + +Then cleanup closes the Query stream. + +Cleanup releases the pending-query map and its unflushed data. + +Finally, cleanup calls `QueryClient.clear()`. This call removes the cache entries for the request. + +Query removal clears GC timers. It cancels active retryers. It also aborts the query-function signal. + +A query function that obeys the signal stops after the abort. + +SSR must use a request-owned Router and `QueryClient`. Setup occurs one time for this request-owned pair. + +A request-owned client isolates the request data and cleanup. + +A reused integrated Router keeps `cleanedUp` set to `true`. Subsequent Query dehydration from that Router returns `undefined`. + +Cleanup can occur before the Router dehydration function completes. An aborted request is one cause of this race. + +The integration uses a `finally` block around the Router dehydration function. + +The block clears entries that the Router dehydration function creates after cleanup. + +If cleanup occurs first and Router dehydration succeeds, the integration returns `undefined`. + +If Router dehydration fails, the error propagates. If cleanup starts first, cleanup takes priority over successful dehydration data. + +## Client Hydration + +The client branch first awaits the hydration function that Router already has. Then it restores the initial Query cache synchronously. + +It passes the query-only object `{ queries: query.initial }` to Query Core. + +Then the integration gets the stream reader. Then it schedules the first read. + +Router hydration resolves after reader setup. Stream chunks and transported pending promises continue independently. + +If the stream buffer contains the first chunk, chunk hydration can occur before Router resumes. Otherwise, Router can resume before the chunk arrives. + +The reader processes one chunk at a time. It wraps each chunk as `{ queries: value }` for `hydrateQueryClient`. + +The reader starts the next read after it hydrates the current chunk. The reader is the only reader for this stream. + +The reader holds the stream lock for the lifetime of the read chain. + +If a read or chunk hydration fails, the reader stops. The integration writes `Error reading query stream:` and the error to the console. + +## Pending Promise Hydration + +A pending query can contain a serialized promise. Query Core first examines that promise for an inline result. + +Router reconstructs a native promise. Thus, Router normally reports the promise result asynchronously. + +If the timestamp and cache-state tests pass, Query Core can use an unresolved promise as `initialPromise`. + +The first fetch attempt uses that transported promise. A rejected promise can enter normal retry behavior. + +A subsequent retry can call an available query function. + +An active retryer supplies the transported promise. + +For an `undefined` promise value, a later client fetch uses the available query function. + +## Redirect Errors + +The integration enables client redirect handling by default. The client branch installs these handlers. + +The client branch installs redirect handlers on the QueryClient caches. It preserves all other cache configuration. + +Redirect handlers operate after Query Core completes its retry process. They store the Router location at that time in `error.options._fromLocation`. + +Then the integration resolves the redirect. Then it calls `router.navigate`. + +The redirect path uses the integration handler in place of the previous cache error function. + +`handleRedirects: false` preserves the original error functions for both caches. + +## Error Paths + +An error from the Router dehydration function propagates before Query stream creation. + +An error during initial Query dehydration also propagates to Router. Router request cleanup then clears the request-owned `QueryClient`. + +If streamed dehydration fails, the integration puts the stream in an error state. Finalization also removes the Query cache subscription. + +An initial hydration error rejects Router hydration. This category includes an error from initial Router or Query hydration. + +After the read chain starts, a stream read error only writes a console error. A chunk hydration error has the same behavior. + +An accepted pending-promise rejection uses Query retry and cache error behavior after Router hydration resolves. + +Query Core consults `shouldRedactErrors` when a dehydrated pending promise rejects. + +Only `false` preserves the original rejection error. Other values cause `Error('redacted')`. + +## Ownership Boundaries + +This package owns these resources: + +- The Query cache subscription for the render phase +- The set of sent query hashes +- The pending-query map +- The controller for the server Query stream +- The cleanup action for the request-owned `QueryClient`. + +Router core owns these resources: + +- Seroval serialization of streams and promises +- `ServerSsr` listener dispatch and cleanup idempotence +- Backpressure, cancellation, and lifetime limits for the transformed stream. + +TanStack Start or `createRequestHandler` owns the standard request order and request-signal connection. + +Framework adapters and the Router stream transform report render completion. The host runtime owns the HTTP connection. + +The application owns `QueryClient` creation. SSR must use a new `QueryClient` for each request. + +## Assumptions + +These assumptions are necessary for the implementation: + +- Setup occurs before server SSR attachment. +- Standard handlers call `serverSsr.dehydrate()` one time. +- Direct dehydration calls occur sequentially. +- Initial Query dehydration callbacks complete before cleanup starts. +- The first transported query version for each hash is sufficient for the request. + +These assumptions keep the lifecycle small. The component that defines each lifecycle rule controls that rule. + +## Core Test Coverage + +The core unit tests cover these behaviors: + +- Explicit serialization and filtering for initial and streamed dehydration +- Explicit deserialization for initial and streamed hydration +- Default transport of pending queries and their promises +- Query-only initial cache access +- Creation and removal of the Query cache subscription +- Cleanup during asynchronous dehydration +- Same-turn batching from stored `Query` references +- A default serializer changed after setup +- A streamed serialization error +- Cleanup during an active render-time query +- Cleanup after Query stream cancellation +- Cleanup before dehydration +- Query cancellation during cleanup +- Cleanup registration during server SSR attachment. + +Additional validation can cover these behaviors: + +- Duplicate-hash suppression after transport +- `shouldRedactErrors` +- Redirect handling +- Errors from a detached client stream +- Cleanup from an HTTP request abort +- Terminal serialization for built-in string rendering +- Cleanup that starts inside an initial dehydration callback +- Reuse of an integrated Router. + +## E2E Coverage + +The E2E suite for the React Query integration includes an awaited loader query and an unawaited loader query. + +The suite also covers render-time `useSuspenseQuery`. In its plain `useQuery` case, the browser calls the query function. + +The Solid and Vue suites include the two loader cases and loader-prefetched `useQuery`. + +These suites examine final server-origin or client-origin values. + +The React query-heavy suite includes nine render-time `useSuspenseQuery` calls. Three queries return immediately. + +Six queries return after delays. + +The suite examines server-origin values, browser hydration, and client navigation. + +It also finds the `slow-async-3` payload before Router emits the `$_TSR.e()` serialization-end marker. + +One emitted script with `.next(` contains all three immediate values. Script buffering can combine more than one Seroval operation. + +The script assertion provides serialization-placement evidence. The core unit test provides direct array-batching evidence. + +## Performance Tests + +`tests/dehydrate.bench.ts` first compares the output of two dehydration methods. It removes `dehydratedAt` from the comparison because this value changes. + +Then it compares full-cache filtering with direct query-reference dehydration. The cache sizes are 10, 100, 1,000, and 10,000 queries. + +`tests/server-lifecycle.bench.ts` measures six synthetic operations: + +- Stream creation and closure +- Query cache subscription and removal +- Setup and cleanup before dehydration +- Setup and dehydration for an empty request +- One hundred `setQueryData` writes in a baseline `QueryClient` +- One hundred `setQueryData` writes with the integration. + +The lifecycle benchmark measures these synthetic operations directly. + +These benchmarks provide informational measurements. + +## Maintainer Commands + +After a Query Core update, reexamine this document. + +Do the core unit tests from the repository root: + +```sh +CI=1 NX_DAEMON=false pnpm nx run @tanstack/router-ssr-query-core:test:unit --outputStyle=stream --skipRemoteCache -- tests/index.test.ts +``` + +Do the optional GC test with the Nx local cache disabled: + +```sh +RUN_SSR_GC_TESTS=1 CI=1 NX_DAEMON=false pnpm nx run @tanstack/router-ssr-query-core:test:unit --outputStyle=stream --skipRemoteCache --skipNxCache -- tests/index.test.ts +``` + +Do the performance tests: + +```sh +CI=1 NX_DAEMON=false pnpm nx run @tanstack/router-ssr-query-core:test:perf --outputStyle=stream --skipRemoteCache +``` + +Do the E2E tests for the React Query integration: + +```sh +CI=1 NX_DAEMON=false pnpm nx run tanstack-react-start-e2e-query-integration:test:e2e --outputStyle=stream --skipRemoteCache +``` + +Do the E2E tests for the Solid Query integration: + +```sh +CI=1 NX_DAEMON=false pnpm nx run tanstack-solid-start-e2e-query-integration:test:e2e --outputStyle=stream --skipRemoteCache +``` + +Do the E2E tests for the Vue Query integration: + +```sh +CI=1 NX_DAEMON=false pnpm nx run tanstack-vue-start-e2e-query-integration:test:e2e --outputStyle=stream --skipRemoteCache +``` + +Do the E2E tests for the React query-heavy suite: + +```sh +CI=1 NX_DAEMON=false pnpm nx run tanstack-react-start-e2e-streaming-ssr:test:e2e --outputStyle=stream --skipRemoteCache -- tests/query-heavy.spec.ts +``` diff --git a/packages/router-ssr-query-core/src/index.ts b/packages/router-ssr-query-core/src/index.ts index 685653dea0c..830e6037932 100644 --- a/packages/router-ssr-query-core/src/index.ts +++ b/packages/router-ssr-query-core/src/index.ts @@ -1,6 +1,6 @@ import { - dehydrate as queryDehydrate, - hydrate as queryHydrate, + dehydrateQuery, + hydrate as hydrateQueryClient, } from '@tanstack/query-core' import { isRedirect } from '@tanstack/router-core' import { isServer } from '@tanstack/router-core/isServer' @@ -8,10 +8,21 @@ import type { AnyRouter } from '@tanstack/router-core' import type { DehydrateOptions, HydrateOptions, + Query, QueryClient, - DehydratedState as QueryDehydratedState, } from '@tanstack/query-core' +type DehydratedQuery = ReturnType + +const shouldDehydrateAllQueries = () => true + +type QueryStreamState = { + controller: ReadableStreamDefaultController> + sentQueries: Set + unsubscribe: () => void + pendingQueries?: Map +} + export type RouterSsrQueryOptions = { router: TRouter queryClient: QueryClient @@ -28,8 +39,10 @@ export type RouterSsrQueryOptions = { } type DehydratedRouterQueryState = { - dehydratedQueryClient?: QueryDehydratedState - queryStream: ReadableStream + query: { + initial?: Array + stream: ReadableStream> + } } export function setupCoreRouterSsrQueryIntegration({ @@ -39,235 +52,234 @@ export function setupCoreRouterSsrQueryIntegration({ hydrateOptions, handleRedirects = true, }: RouterSsrQueryOptions) { - const ogHydrate = router.options.hydrate - const ogDehydrate = router.options.dehydrate - if (isServer ?? router.isServer) { - const sentQueries = new Set() - const queryStream = createPushableStream() - let unsubscribe: (() => void) | undefined = undefined - let cleanupRegistered = false - let tornDown = false + const originalDehydrate = router.options.dehydrate + let streamState: QueryStreamState | undefined + let cleanedUp = false - const teardown = () => { - if (tornDown) return - tornDown = true - try { - unsubscribe?.() - } catch { - // ignore - } - unsubscribe = undefined - try { - if (!queryStream.isClosed()) queryStream.close() - } catch { - // ignore - } - // Cancel any in-flight queries and clear the cache. Removing queries - // cancels their gcTime setTimeout handles which would otherwise pin - // the queryClient (and transitively the router via router.context) - // alive for the full gcTime window (default 5min) per SSR request. - try { - queryClient.cancelQueries() - } catch { - // ignore + const finalizeQueryStream = (failure?: { error: unknown }) => { + const state = streamState + streamState = undefined + if (!state) { + return } + + state.unsubscribe() + try { - queryClient.clear() + if (failure) { + state.controller.error(failure.error) + } else { + state.controller.close() + } } catch { - // ignore + // The stream consumer can already have cancelled the stream. } - sentQueries.clear() + } + + const teardown = () => { + cleanedUp = true + finalizeQueryStream() + // Clearing destroys queries, aborts in-flight work, and cancels gcTime + // handles that would otherwise retain request state for up to 5 minutes. + queryClient.clear() } // Register teardown as soon as SSR attaches. attachRouterServerSsrUtils() // runs before router.load(), so this covers redirects/errors thrown before // router.options.dehydrate() can run. - const registerCleanup = (serverSsr = router.serverSsr) => { - if (cleanupRegistered) return - if (!serverSsr) return - serverSsr.onCleanup(teardown) - cleanupRegistered = true - } router.serverSsrLifecycle = { ...router.serverSsrLifecycle, onServerSsrAttach: [ ...(router.serverSsrLifecycle?.onServerSsrAttach ?? []), - registerCleanup, + (serverSsr) => serverSsr.onCleanup(teardown), ], } - router.options.dehydrate = - async (): Promise => { - router.serverSsr!.onRenderFinished(() => { - if (!queryStream.isClosed()) queryStream.close() - unsubscribe?.() - unsubscribe = undefined - }) - const ogDehydrated = await ogDehydrate?.() - - const dehydratedRouter = { - ...ogDehydrated, - // prepare the stream for queries coming up during rendering - queryStream: queryStream.stream, - } - - const dehydratedQueryClient = queryDehydrate( - queryClient, - dehydrateOptions, - ) - if (dehydratedQueryClient.queries.length > 0) { - dehydratedQueryClient.queries.forEach((query) => { - sentQueries.add(query.queryHash) - }) - dehydratedRouter.dehydratedQueryClient = dehydratedQueryClient + router.options.dehydrate = async (): Promise< + DehydratedRouterQueryState | undefined + > => { + let originalDehydrated: Awaited< + ReturnType> + > + try { + originalDehydrated = await originalDehydrate?.() + } finally { + if (cleanedUp) { + queryClient.clear() } - - return dehydratedRouter } - const ogClientOptions = queryClient.getDefaultOptions() - queryClient.setDefaultOptions({ - ...ogClientOptions, - dehydrate: { - shouldDehydrateQuery: () => true, - ...ogClientOptions.dehydrate, - }, - }) - - unsubscribe = queryClient.getQueryCache().subscribe((event) => { - // before rendering starts, we do not stream individual queries - // instead we dehydrate the entire query client in router's dehydrate() - // if attachRouterServerSsrUtils() has not been called yet, `router.serverSsr` will be undefined and we also do not stream - if (!router.serverSsr?.isDehydrated()) { - return - } - if (sentQueries.has(event.query.queryHash)) { - return - } - // promise not yet set on the query, so we cannot stream it yet - if (!event.query.promise) { + if (cleanedUp) { return } - if (queryStream.isClosed()) { - console.warn( - `tried to stream query ${event.query.queryHash} after stream was already closed`, - ) - return - } - const dehydratedQuery = queryDehydrate(queryClient, { - ...dehydrateOptions, - shouldDehydrateQuery: (query) => { - if (query.queryHash !== event.query.queryHash) { - return false - } - return ( - (ogClientOptions.dehydrate?.shouldDehydrateQuery?.(query) ?? - true) && - (dehydrateOptions?.shouldDehydrateQuery?.(query) ?? true) + const currentDehydrateOptions = queryClient.getDefaultOptions().dehydrate + const shouldDehydrateQuery = + dehydrateOptions?.shouldDehydrateQuery ?? + currentDehydrateOptions?.shouldDehydrateQuery ?? + shouldDehydrateAllQueries + const serializeData = + dehydrateOptions?.serializeData ?? + currentDehydrateOptions?.serializeData + const shouldRedactErrors = + dehydrateOptions?.shouldRedactErrors ?? + currentDehydrateOptions?.shouldRedactErrors + const initialQueries = new Array() + const sentQueries = new Set() + + for (const query of queryClient.getQueryCache().getAll()) { + if (shouldDehydrateQuery(query)) { + initialQueries.push( + dehydrateQuery(query, serializeData, shouldRedactErrors), ) + sentQueries.add(query.queryHash) + } + } + + let controller!: ReadableStreamDefaultController> + const stream = new ReadableStream>({ + start(value) { + controller = value }, }) + const flushPendingQueries = () => { + const state = streamState + const queries = state?.pendingQueries + if (!state || !queries) { + return + } + state.pendingQueries = undefined - if (dehydratedQuery.queries.length === 0) { - return - } - - sentQueries.add(event.query.queryHash) - queryStream.enqueue(dehydratedQuery) - }) - // on the client - } else { - router.options.hydrate = async (dehydrated: DehydratedRouterQueryState) => { - await ogHydrate?.(dehydrated) - // hydrate the query client with the dehydrated data (if it was dehydrated on the server) - if (dehydrated.dehydratedQueryClient) { - queryHydrate( - queryClient, - dehydrated.dehydratedQueryClient, - hydrateOptions, - ) - } + const dehydratedQueries = new Array() - // read the query stream and hydrate the queries as they come in - const reader = dehydrated.queryStream.getReader() - reader - .read() - .then(async function handle({ done, value }) { - queryHydrate(queryClient, value, hydrateOptions) - if (done) { - return - } - const result = await reader.read() - return handle(result) - }) - .catch((err) => { - console.error('Error reading query stream:', err) - }) - } - if (handleRedirects) { - const ogMutationCacheConfig = queryClient.getMutationCache().config - queryClient.getMutationCache().config = { - ...ogMutationCacheConfig, - onError: (error, ...rest) => { - if (isRedirect(error)) { - error.options._fromLocation = router.stores.location.get() - return router.navigate(router.resolveRedirect(error).options) + for (const query of queries.values()) { + if ( + state.sentQueries.has(query.queryHash) || + !shouldDehydrateQuery(query) + ) { + continue } - return ogMutationCacheConfig.onError?.(error, ...rest) - }, + dehydratedQueries.push( + dehydrateQuery(query, serializeData, shouldRedactErrors), + ) + state.sentQueries.add(query.queryHash) + } + + if (dehydratedQueries.length > 0) { + state.controller.enqueue(dehydratedQueries) + } } + const unsubscribe = queryClient.getQueryCache().subscribe((event) => { + const state = streamState + if (!state) { + return + } + if ( + state.sentQueries.has(event.query.queryHash) || + // The promise is not set yet for the first query-cache event. + !event.query.promise + ) { + return + } - const ogQueryCacheConfig = queryClient.getQueryCache().config - queryClient.getQueryCache().config = { - ...ogQueryCacheConfig, - onError: (error, ...rest) => { - if (isRedirect(error)) { - error.options._fromLocation = router.stores.location.get() - return router.navigate(router.resolveRedirect(error).options) - } + if (!state.pendingQueries) { + state.pendingQueries = new Map() + // Flush before React resumes resolved Suspense boundaries while + // still batching queries that settle in the same turn. + queueMicrotask(() => { + try { + flushPendingQueries() + } catch (error) { + finalizeQueryStream({ error }) + } + }) + } + state.pendingQueries.set(event.query.queryHash, event.query) + }) + streamState = { controller, sentQueries, unsubscribe } - return ogQueryCacheConfig.onError?.(error, ...rest) + const finishRendering = () => { + try { + flushPendingQueries() + } catch (error) { + finalizeQueryStream({ error }) + return + } + finalizeQueryStream() + } + + router.serverSsr!.onRenderFinished(finishRendering) + return { + ...originalDehydrated, + query: { + ...(initialQueries.length > 0 && { + initial: initialQueries, + }), + stream, }, } } + return } -} + const originalHydrate = router.options.hydrate + router.options.hydrate = async (dehydrated: DehydratedRouterQueryState) => { + await originalHydrate?.(dehydrated) -type PushableStream = { - stream: ReadableStream - enqueue: (chunk: unknown) => void - close: () => void - isClosed: () => boolean - error: (err: unknown) => void -} + const query = dehydrated.query + if (query.initial) { + hydrateQueryClient( + queryClient, + { queries: query.initial }, + hydrateOptions, + ) + } -function createPushableStream(): PushableStream { - let controllerRef: ReadableStreamDefaultController - const stream = new ReadableStream({ - start(controller) { - controllerRef = controller - }, - }) - let _isClosed = false + const reader = query.stream.getReader() + reader + .read() + .then(function handle({ + done, + value, + }: ReadableStreamReadResult< + Array + >): void | Promise { + if (done) { + return + } + hydrateQueryClient(queryClient, { queries: value }, hydrateOptions) + return reader.read().then(handle) + }) + .catch((error) => { + console.error('Error reading query stream:', error) + }) + } + if (handleRedirects) { + const originalMutationCacheConfig = queryClient.getMutationCache().config + queryClient.getMutationCache().config = { + ...originalMutationCacheConfig, + onError: (error, ...rest) => { + if (isRedirect(error)) { + error.options._fromLocation = router.stores.location.get() + return router.navigate(router.resolveRedirect(error).options) + } + + return originalMutationCacheConfig.onError?.(error, ...rest) + }, + } - return { - stream, - enqueue: (chunk) => { - if (!_isClosed) controllerRef.enqueue(chunk) - }, - close: () => { - if (_isClosed) return - controllerRef.close() - _isClosed = true - }, - isClosed: () => _isClosed, - error: (err: unknown) => { - if (_isClosed) return - _isClosed = true - controllerRef.error(err) - }, + const originalQueryCacheConfig = queryClient.getQueryCache().config + queryClient.getQueryCache().config = { + ...originalQueryCacheConfig, + onError: (error, ...rest) => { + if (isRedirect(error)) { + error.options._fromLocation = router.stores.location.get() + return router.navigate(router.resolveRedirect(error).options) + } + + return originalQueryCacheConfig.onError?.(error, ...rest) + }, + } } } diff --git a/packages/router-ssr-query-core/tests/dehydrate.bench.ts b/packages/router-ssr-query-core/tests/dehydrate.bench.ts new file mode 100644 index 00000000000..7e90ab4a71d --- /dev/null +++ b/packages/router-ssr-query-core/tests/dehydrate.bench.ts @@ -0,0 +1,85 @@ +// @vitest-environment node + +import { QueryClient, dehydrate, dehydrateQuery } from '@tanstack/query-core' +import { afterAll, bench, describe, expect } from 'vitest' +import type { Query } from '@tanstack/query-core' + +type DehydratedQuery = ReturnType + +let benchmarkSink = 0 + +describe.each([ + { cachedQueryCount: 10, selectedQueryCount: 1 }, + { cachedQueryCount: 100, selectedQueryCount: 10 }, + { cachedQueryCount: 1_000, selectedQueryCount: 10 }, + { cachedQueryCount: 10_000, selectedQueryCount: 10 }, +])( + 'dehydrate $selectedQueryCount of $cachedQueryCount cached queries', + ({ cachedQueryCount, selectedQueryCount }) => { + const queryClient = new QueryClient() + + for (let index = 0; index < cachedQueryCount; index++) { + queryClient.setQueryData(['query', index], `data-${index}`) + } + + const selectedQueries = new Map() + for ( + let index = cachedQueryCount - selectedQueryCount; + index < cachedQueryCount; + index++ + ) { + const query = queryClient.getQueryCache().find({ + queryKey: ['query', index], + })! + selectedQueries.set(query.queryHash, query) + } + const selectedQueryHashes = new Set(selectedQueries.keys()) + + const scannedQueries = dehydrate(queryClient, { + shouldDehydrateQuery: (query) => selectedQueryHashes.has(query.queryHash), + }).queries + const directlyDehydratedQueries = Array.from( + selectedQueries.values(), + (query) => dehydrateQuery(query), + ) + + expect(directlyDehydratedQueries.map(comparableQuery)).toEqual( + scannedQueries.map(comparableQuery), + ) + + afterAll(() => queryClient.clear()) + + bench('filter full query cache', () => { + const dehydratedQueries = dehydrate(queryClient, { + shouldDehydrateQuery: (query) => + selectedQueryHashes.has(query.queryHash), + }).queries + benchmarkSink = consumeQueries(dehydratedQueries) + }) + + bench('dehydrate direct query references', () => { + const dehydratedQueries = new Array() + for (const query of selectedQueries.values()) { + dehydratedQueries.push(dehydrateQuery(query)) + } + benchmarkSink = consumeQueries(dehydratedQueries) + }) + }, +) + +function comparableQuery({ dehydratedAt: _, ...query }: DehydratedQuery) { + return query +} + +function consumeQueries(queries: Array) { + let value = 0 + for (const query of queries) { + value += (query.dehydratedAt ?? 0) + query.queryHash.length + if (typeof query.state.data === 'string') { + value += query.state.data.length + } + } + return value +} + +void benchmarkSink diff --git a/packages/router-ssr-query-core/tests/index.test.ts b/packages/router-ssr-query-core/tests/index.test.ts index a167dbd03e1..4adb7a822cd 100644 --- a/packages/router-ssr-query-core/tests/index.test.ts +++ b/packages/router-ssr-query-core/tests/index.test.ts @@ -9,7 +9,6 @@ type TestRouter = { hydrate?: (dehydrated: any) => unknown | Promise } serverSsr?: { - isDehydrated: () => boolean onRenderFinished: (listener: () => void) => void onCleanup: (listener: () => void) => void } @@ -20,56 +19,50 @@ type TestRouter = { } } -type ServerRouterFixture = { - router: TestRouter - finishRender: () => void - triggerCleanup: () => void - attachServerSsr: () => void - setDehydrated: (value: boolean) => void - cleanupListenerCount: () => number -} - -function createServerRouter(): ServerRouterFixture { +function createServerRouter() { const renderFinishedListeners = new Array<() => void>() const cleanupListeners = new Array<() => void>() - let dehydrated = false + let cleanedUp = false const serverSsr = { - isDehydrated: () => dehydrated, onRenderFinished: (listener: () => void) => { - renderFinishedListeners.push(listener) + if (!cleanedUp) { + renderFinishedListeners.push(listener) + } }, onCleanup: (listener: () => void) => { - cleanupListeners.push(listener) + if (!cleanedUp) { + cleanupListeners.push(listener) + } }, } + const router: TestRouter = { + isServer: true, + options: {}, + } - const result: ServerRouterFixture = { - router: { - isServer: true, - options: {}, - serverSsr, + return { + router, + attachServerSsr() { + router.serverSsr = serverSsr + router.serverSsrLifecycle?.onServerSsrAttach.forEach((listener) => { + listener(serverSsr) + }) }, - finishRender: () => { - renderFinishedListeners.splice(0).forEach((listener) => listener()) + finishRender() { + if (!cleanedUp) { + renderFinishedListeners.splice(0).forEach((listener) => listener()) + } }, - triggerCleanup: () => { + triggerCleanup() { + if (cleanedUp) { + return + } + cleanedUp = true cleanupListeners.splice(0).forEach((listener) => listener()) + renderFinishedListeners.length = 0 + router.serverSsr = undefined }, - attachServerSsr: () => { - result.router.serverSsr = serverSsr - result.router.serverSsrLifecycle?.onServerSsrAttach.forEach( - (listener) => { - listener(serverSsr) - }, - ) - }, - setDehydrated: (value: boolean) => { - dehydrated = value - }, - cleanupListenerCount: () => cleanupListeners.length, } - - return result } async function readStream(stream: ReadableStream): Promise> { @@ -78,11 +71,9 @@ async function readStream(stream: ReadableStream): Promise> { while (true) { const result = await reader.read() - if (result.done) { return chunks } - chunks.push(result.value) } } @@ -92,11 +83,7 @@ function createDeferred() { const promise = new Promise((res) => { resolve = res }) - - return { - promise, - resolve, - } + return { promise, resolve } } function createDehydratedQueryState(data: string) { @@ -116,10 +103,6 @@ function createDehydratedQueryState(data: string) { } } -// Track QueryClients per-test and clear them in afterEach. Without this, -// queries created in tests keep their gcTime setTimeout handles open (5min -// default in jsdom), pinning QueryClient + QueryCache + this test's router -// alive across the whole suite. cancelQueries() + clear() drops them. const trackedQueryClients = new Set() function track(client: T): T { trackedQueryClients.add(client) @@ -127,390 +110,562 @@ function track(client: T): T { } afterEach(() => { - vi.clearAllMocks() + vi.restoreAllMocks() for (const client of trackedQueryClients) { - try { - client.cancelQueries() - } catch { - // ignore - } - try { - client.clear() - } catch { - // ignore - } + client.clear() } trackedQueryClients.clear() }) describe('setupCoreRouterSsrQueryIntegration', () => { - it('uses custom dehydrate options for the initial payload and streamed queries', async () => { - const queryClient = track(new QueryClient()) - const { router, finishRender, attachServerSsr, setDehydrated } = + it('uses custom dehydration options for initial and streamed queries', async () => { + const queryClient = track( + new QueryClient({ + defaultOptions: { + dehydrate: { shouldDehydrateQuery: () => false }, + }, + }), + ) + const { router, attachServerSsr, finishRender, triggerCleanup } = createServerRouter() - router.serverSsr = undefined setupCoreRouterSsrQueryIntegration({ router: router as any, queryClient, dehydrateOptions: { serializeData: (data) => `${data}-serialized`, - shouldDehydrateQuery: (query) => query.queryKey[0] !== 'skip', + shouldDehydrateQuery: (query) => + !String(query.queryKey[0]).startsWith('skip'), }, }) attachServerSsr() - queryClient.setQueryData(['include'], 'initial') queryClient.setQueryData(['skip'], 'ignored') const dehydrated = (await router.options.dehydrate?.()) as { - dehydratedQueryClient?: { - queries: Array<{ queryKey: Array; state: { data: unknown } }> + query: { + initial?: Array<{ + queryKey: Array + state: { data: unknown } + }> + stream: ReadableStream< + Array<{ queryKey: Array; state: { data: unknown } }> + > } - queryStream: ReadableStream<{ - queries: Array<{ queryKey: Array; state: { data: unknown } }> - }> } - expect(dehydrated.dehydratedQueryClient?.queries).toHaveLength(1) - expect(dehydrated.dehydratedQueryClient?.queries[0]?.queryKey).toEqual([ - 'include', + expect(dehydrated.query.initial).toMatchObject([ + { queryKey: ['include'], state: { data: 'initial-serialized' } }, ]) - expect(dehydrated.dehydratedQueryClient?.queries[0]?.state.data).toBe( - 'initial-serialized', - ) - const streamedQueriesPromise = readStream(dehydrated.queryStream) + const streamedQueriesPromise = readStream(dehydrated.query.stream) const includedDeferred = createDeferred() const skippedDeferred = createDeferred() - - setDehydrated(true) - const includedPromise = queryClient.fetchQuery({ + const included = queryClient.fetchQuery({ queryKey: ['streamed'], queryFn: () => includedDeferred.promise, }) - const skippedPromise = queryClient.fetchQuery({ - queryKey: ['skip'], + const skipped = queryClient.fetchQuery({ + queryKey: ['skip-streamed'], queryFn: () => skippedDeferred.promise, }) - await Promise.resolve() includedDeferred.resolve('next') - skippedDeferred.resolve('still-ignored') - await Promise.all([includedPromise, skippedPromise]) + skippedDeferred.resolve('ignored') + await Promise.all([included, skipped]) finishRender() - const streamedQueries = await streamedQueriesPromise - - expect(streamedQueries).toHaveLength(1) - expect(streamedQueries[0]?.queries).toHaveLength(1) - expect(streamedQueries[0]?.queries[0]?.queryKey).toEqual(['streamed']) - expect(streamedQueries[0]?.queries[0]?.state.data).toBe('next-serialized') + expect(await streamedQueriesPromise).toMatchObject([ + [{ queryKey: ['streamed'], state: { data: 'next-serialized' } }], + ]) + triggerCleanup() }) - it('uses custom hydrate options for the initial payload and streamed queries', async () => { + it('dehydrates pending queries by default', async () => { const queryClient = track(new QueryClient()) - const router: TestRouter = { - isServer: false, - options: {}, - } + const { router, attachServerSsr, finishRender, triggerCleanup } = + createServerRouter() + const queryStarted = createDeferred() + const queryData = createDeferred() setupCoreRouterSsrQueryIntegration({ router: router as any, queryClient, - hydrateOptions: { - defaultOptions: { - deserializeData: (data) => `${String(data)}-hydrated`, - }, + }) + attachServerSsr() + const pendingQuery = queryClient.fetchQuery({ + queryKey: ['pending'], + queryFn: () => { + queryStarted.resolve() + return queryData.promise }, }) + await queryStarted.promise + + const dehydrated = (await router.options.dehydrate?.()) as { + query: { + initial: Array<{ + promise?: Promise + queryKey: Array + state: { status: string } + }> + stream: ReadableStream> + } + } + + expect(dehydrated.query.initial).toMatchObject([ + { queryKey: ['pending'], state: { status: 'pending' } }, + ]) + expect(dehydrated.query.initial[0]?.promise).toBeInstanceOf(Promise) + + const streamedQueriesPromise = readStream(dehydrated.query.stream) + queryData.resolve('data') + await pendingQuery + finishRender() + + expect(await streamedQueriesPromise).toEqual([]) + triggerCleanup() + }) + it('uses custom hydration options for initial and streamed queries', async () => { + const queryClient = track(new QueryClient()) + const router: TestRouter = { isServer: false, options: {} } const stream = new ReadableStream({ start(controller) { - controller.enqueue({ - mutations: [], - queries: [ - { - queryHash: '["streamed"]', - queryKey: ['streamed'], - state: createDehydratedQueryState('stream'), - }, - ], - }) + controller.enqueue([ + { + queryHash: '["streamed"]', + queryKey: ['streamed'], + state: createDehydratedQueryState('stream'), + }, + { + queryHash: '["streamed-batch"]', + queryKey: ['streamed-batch'], + state: createDehydratedQueryState('batch'), + }, + ]) controller.close() }, }) + setupCoreRouterSsrQueryIntegration({ + router: router as any, + queryClient, + hydrateOptions: { + defaultOptions: { + deserializeData: (data) => `${String(data)}-hydrated`, + }, + }, + }) await router.options.hydrate?.({ - dehydratedQueryClient: { - mutations: [], - queries: [ + query: { + initial: [ { queryHash: '["initial"]', queryKey: ['initial'], state: createDehydratedQueryState('initial'), }, ], + stream, }, - queryStream: stream, }) - await Promise.resolve() await Promise.resolve() expect(queryClient.getQueryData(['initial'])).toBe('initial-hydrated') expect(queryClient.getQueryData(['streamed'])).toBe('stream-hydrated') + expect(queryClient.getQueryData(['streamed-batch'])).toBe('batch-hydrated') }) -}) -// GC reclamation tests are non-deterministic by nature (V8 makes no -// guarantee about WeakRef collection timing). Run them on demand only: -// RUN_SSR_GC_TESTS=1 pnpm vitest -// The vite config gates --expose-gc on the same env var; outside of that -// gc() is unavailable and the describe is skipped. -const gcAvailable = typeof (globalThis as any).gc === 'function' -const gcTestsEnabled = process.env.RUN_SSR_GC_TESTS === '1' && gcAvailable - -async function forceGc() { - // Multiple passes; V8 may need several GC cycles to collect WeakRef - // targets, especially with closure chains. - for (let i = 0; i < 6; i++) { - ;(globalThis as any).gc() - await new Promise((r) => setTimeout(r, 0)) - } -} + it('subscribes after initial dehydration and releases after rendering', async () => { + const queryClient = track(new QueryClient()) + const { router, attachServerSsr, finishRender, triggerCleanup } = + createServerRouter() -// Reproduces TanStack/router#7402: per-request Router + QueryClient must be -// reclaimable by GC after SSR cleanup. Without the onCleanup teardown, the -// queryCache subscriber closure + gcTime setTimeout handles pin the -// QueryClient (and transitively the Router via router.context) for the full -// gcTime window (default 5min) per request. -describe.runIf(gcTestsEnabled)('SSR memory: GC reclamation', () => { - it('queryClient + router are reclaimable after cleanup', async () => { - let queryClient: QueryClient | null = new QueryClient({ - defaultOptions: { queries: { gcTime: 5 * 60 * 1000 } }, + setupCoreRouterSsrQueryIntegration({ + router: router as any, + queryClient, }) - let serverRouter: ReturnType | null = + expect(queryClient.getQueryCache().hasListeners()).toBe(false) + + attachServerSsr() + const dehydrated = (await router.options.dehydrate?.()) as { + query: { stream: ReadableStream> } + } + expect(queryClient.getQueryCache().hasListeners()).toBe(true) + + const streamedQueriesPromise = readStream(dehydrated.query.stream) + finishRender() + expect(await streamedQueriesPromise).toEqual([]) + expect(queryClient.getQueryCache().hasListeners()).toBe(false) + triggerCleanup() + }) + + it('does not read or dehydrate mutations during initial dehydration', async () => { + const shouldDehydrateMutation = vi.fn(() => true) + const queryClient = track( + new QueryClient({ + defaultOptions: { + dehydrate: { shouldDehydrateMutation }, + }, + }), + ) + const { router, attachServerSsr, finishRender, triggerCleanup } = createServerRouter() + const getAllMutations = vi.spyOn(queryClient.getMutationCache(), 'getAll') - serverRouter.router.serverSsr = undefined + queryClient.getMutationCache().build( + queryClient, + { mutationKey: ['paused'] }, + { + context: undefined, + data: undefined, + error: null, + failureCount: 0, + failureReason: null, + isPaused: true, + status: 'pending', + variables: undefined, + submittedAt: 1, + }, + ) + queryClient.setQueryData(['query'], 'data') setupCoreRouterSsrQueryIntegration({ - router: serverRouter.router as any, + router: router as any, queryClient, }) - serverRouter.attachServerSsr() - - // Populate cache w/ active gcTime timers (the leak anchor). - queryClient.setQueryData(['a'], 'data-a') - queryClient.setQueryData(['b'], 'data-b') + attachServerSsr() - // Run a full SSR cycle: dehydrate (registers onCleanup) -> finishRender. - await serverRouter.router.options.dehydrate?.() - serverRouter.setDehydrated(true) - serverRouter.finishRender() + const dehydrated = (await router.options.dehydrate?.()) as { + query: { + initial: Array<{ queryKey: Array }> + stream: ReadableStream> + } + } + const streamedQueriesPromise = readStream(dehydrated.query.stream) + finishRender() - const qcRef = new WeakRef(queryClient) - const routerRef = new WeakRef(serverRouter.router) - const cacheRef = new WeakRef(queryClient.getQueryCache()) + expect(dehydrated.query.initial).toMatchObject([{ queryKey: ['query'] }]) + expect(getAllMutations).not.toHaveBeenCalled() + expect(shouldDehydrateMutation).not.toHaveBeenCalled() + expect(await streamedQueriesPromise).toEqual([]) + triggerCleanup() + }) - // Simulate full request teardown. - serverRouter.triggerCleanup() + it('returns no query data when cleanup occurs during dehydration', async () => { + const queryClient = track(new QueryClient()) + const { router, attachServerSsr, triggerCleanup } = createServerRouter() + const deferred = createDeferred() - // Drop all strong refs. - queryClient = null - serverRouter = null + router.options.dehydrate = async () => { + await deferred.promise + queryClient.setQueryData(['late'], 'data') + return { original: true } + } + setupCoreRouterSsrQueryIntegration({ + router: router as any, + queryClient, + }) + attachServerSsr() - await forceGc() + const dehydrating = router.options.dehydrate?.() + triggerCleanup() + deferred.resolve() - expect(qcRef.deref(), 'QueryClient should be GCd').toBeUndefined() - expect(routerRef.deref(), 'Router should be GCd').toBeUndefined() - expect(cacheRef.deref(), 'QueryCache should be GCd').toBeUndefined() + await expect(dehydrating).resolves.toBeUndefined() + expect(queryClient.getQueryCache().getAll()).toEqual([]) + expect(queryClient.getQueryCache().hasListeners()).toBe(false) }) - it('without cleanup, queryClient is retained (control)', async () => { - let queryClient: QueryClient | null = new QueryClient({ - defaultOptions: { queries: { gcTime: 5 * 60 * 1000 } }, - }) - let serverRouter: ReturnType | null = + it('batches same-turn query settlements without scanning the cache', async () => { + const queryClient = track(new QueryClient()) + const { router, attachServerSsr, finishRender, triggerCleanup } = createServerRouter() - serverRouter.router.serverSsr = undefined setupCoreRouterSsrQueryIntegration({ - router: serverRouter.router as any, + router: router as any, queryClient, }) - serverRouter.attachServerSsr() + attachServerSsr() + const dehydrated = (await router.options.dehydrate?.()) as { + query: { + stream: ReadableStream< + Array<{ queryHash: string; queryKey: Array }> + > + } + } + const getAll = vi.spyOn(queryClient.getQueryCache(), 'getAll') + const streamedQueriesPromise = readStream(dehydrated.query.stream) + const firstDeferred = createDeferred() + const secondDeferred = createDeferred() + const laterDeferred = createDeferred() + + const first = queryClient.fetchQuery({ + queryKey: ['first'], + queryFn: () => firstDeferred.promise, + }) + const second = queryClient.fetchQuery({ + queryKey: ['second'], + queryFn: () => secondDeferred.promise, + }) + firstDeferred.resolve('first-data') + secondDeferred.resolve('second-data') + await Promise.all([first, second]) + await new Promise((resolve) => setTimeout(resolve, 0)) + + const later = queryClient.fetchQuery({ + queryKey: ['later'], + queryFn: () => laterDeferred.promise, + }) + laterDeferred.resolve('later-data') + await later + finishRender() - queryClient.setQueryData(['a'], 'data-a') - await serverRouter.router.options.dehydrate?.() + expect(getAll).not.toHaveBeenCalled() + expect(await streamedQueriesPromise).toMatchObject([ + [{ queryKey: ['first'] }, { queryKey: ['second'] }], + [{ queryKey: ['later'] }], + ]) + triggerCleanup() + }) - const qcRef = new WeakRef(queryClient) + it('snapshots QueryClient dehydration defaults after Router dehydration', async () => { + const queryClient = track(new QueryClient()) + const { router, attachServerSsr, finishRender, triggerCleanup } = + createServerRouter() - // Drop strong refs WITHOUT triggering cleanup. - queryClient = null - serverRouter = null - - try { - await forceGc() - - // Subscriber closure + gcTime timers keep it alive. This is the bug - // we are guarding against; if this ever passes (returns undefined) the - // production retention chain has changed and the cleanup-based test - // above may also need re-validation. - expect(qcRef.deref()).toBeDefined() - } finally { - // Avoid leaving the retained client + gcTime timers alive for up to - // 5 minutes after the test finishes. - qcRef.deref()?.clear() + setupCoreRouterSsrQueryIntegration({ + router: router as any, + queryClient, + }) + attachServerSsr() + queryClient.setDefaultOptions({ + ...queryClient.getDefaultOptions(), + dehydrate: { + ...queryClient.getDefaultOptions().dehydrate, + serializeData: (data) => `${data}-current`, + }, + }) + queryClient.setQueryData(['initial'], 'initial') + + const dehydrated = (await router.options.dehydrate?.()) as { + query: { + initial: Array<{ state: { data: unknown } }> + stream: ReadableStream> + } } + const streamedQueriesPromise = readStream(dehydrated.query.stream) + queryClient.setDefaultOptions({ + ...queryClient.getDefaultOptions(), + dehydrate: { + ...queryClient.getDefaultOptions().dehydrate, + serializeData: (data) => `${data}-later`, + }, + }) + await queryClient.fetchQuery({ + queryKey: ['streamed'], + queryFn: () => 'streamed', + }) + finishRender() + + expect(dehydrated.query.initial[0]?.state.data).toBe('initial-current') + expect((await streamedQueriesPromise)[0]?.[0]?.state.data).toBe( + 'streamed-current', + ) + triggerCleanup() }) -}) -// ===================================================================== -// CI-stable cleanup behavior tests. These do not rely on GC timing; they -// assert the observable side-effects that make GC reclamation possible. -// ===================================================================== -describe('SSR cleanup: deterministic behavior', () => { - it('teardown runs when cleanup fires before dehydrate (loader redirect/error case)', async () => { + it('errors and unsubscribes when streamed serialization throws', async () => { const queryClient = track(new QueryClient()) - const { router, triggerCleanup, attachServerSsr, setDehydrated } = + const { router, attachServerSsr, finishRender, triggerCleanup } = createServerRouter() + const error = new Error('serialize failed') - router.serverSsr = undefined setupCoreRouterSsrQueryIntegration({ router: router as any, queryClient, + dehydrateOptions: { + serializeData: () => { + throw error + }, + }, }) attachServerSsr() - - // Cleanup registration happens when server SSR attaches, before loaders. - setDehydrated(true) + const dehydrated = (await router.options.dehydrate?.()) as { + query: { stream: ReadableStream> } + } + const streamedQueriesPromise = readStream(dehydrated.query.stream) await queryClient.fetchQuery({ - queryKey: ['early'], + queryKey: ['throws'], queryFn: () => 'data', }) + finishRender() + + await expect(streamedQueriesPromise).rejects.toBe(error) + expect(queryClient.getQueryCache().hasListeners()).toBe(false) + triggerCleanup() + }) - expect(queryClient.getQueryData(['early'])).toBe('data') + it('closes the stream and aborts an in-flight render query on cleanup', async () => { + const queryClient = track(new QueryClient()) + const { router, attachServerSsr, triggerCleanup } = createServerRouter() + const queryStarted = createDeferred() + let aborted = false - // Trigger cleanup as createRequestHandler's finally block would. + setupCoreRouterSsrQueryIntegration({ + router: router as any, + queryClient, + }) + attachServerSsr() + const dehydrated = (await router.options.dehydrate?.()) as { + query: { stream: ReadableStream> } + } + const streamedQueriesPromise = readStream(dehydrated.query.stream) + const query = queryClient.fetchQuery({ + queryKey: ['in-flight'], + queryFn: ({ signal }) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => { + aborted = true + reject(new Error('aborted')) + }) + queryStarted.resolve() + }), + }) + query.catch(() => {}) + await queryStarted.promise triggerCleanup() - // After cleanup the cache must be cleared (gcTime timers gone). - expect(queryClient.getQueryData(['early'])).toBeUndefined() + expect(await streamedQueriesPromise).toEqual([]) + expect(queryClient.getQueryCache().getAll()).toEqual([]) + expect(queryClient.getQueryCache().hasListeners()).toBe(false) + expect(aborted).toBe(true) }) - it('cancels in-flight queries on cleanup (signal aborted)', async () => { + it('clears the QueryClient when the stream is already cancelled', async () => { const queryClient = track(new QueryClient()) - const { router, triggerCleanup, attachServerSsr, setDehydrated } = - createServerRouter() + const { router, attachServerSsr, triggerCleanup } = createServerRouter() + + setupCoreRouterSsrQueryIntegration({ + router: router as any, + queryClient, + }) + attachServerSsr() + queryClient.setQueryData(['cancelled-stream'], 'data') + const dehydrated = (await router.options.dehydrate?.()) as { + query: { stream: ReadableStream> } + } + await dehydrated.query.stream.cancel() + + expect(() => triggerCleanup()).not.toThrow() + expect(queryClient.getQueryCache().getAll()).toEqual([]) + expect(queryClient.getQueryCache().hasListeners()).toBe(false) + }) +}) + +describe('SSR cleanup', () => { + it('clears queries when a request ends before dehydration', () => { + const queryClient = track(new QueryClient()) + const { router, attachServerSsr, triggerCleanup } = createServerRouter() - router.serverSsr = undefined setupCoreRouterSsrQueryIntegration({ router: router as any, queryClient, }) attachServerSsr() - setDehydrated(true) + queryClient.setQueryData(['loader'], 'data') + triggerCleanup() - let observedAborted = false + expect(queryClient.getQueryCache().getAll()).toEqual([]) + }) + + it('aborts in-flight queries', async () => { + const queryClient = track(new QueryClient()) + const { router, attachServerSsr, triggerCleanup } = createServerRouter() const queryStarted = createDeferred() - const inflight = queryClient.fetchQuery({ + let aborted = false + + setupCoreRouterSsrQueryIntegration({ + router: router as any, + queryClient, + }) + attachServerSsr() + const query = queryClient.fetchQuery({ queryKey: ['slow'], queryFn: ({ signal }) => new Promise((_resolve, reject) => { signal.addEventListener('abort', () => { - observedAborted = true + aborted = true reject(new Error('aborted')) }) - // Signal the test once the queryFn has actually started and the - // abort listener is wired up. No real timers required. queryStarted.resolve() }), }) - // Swallow the rejection from fetchQuery - inflight.catch(() => {}) - + query.catch(() => {}) await queryStarted.promise triggerCleanup() - // Flush microtasks so the abort event handler runs. - await Promise.resolve() await Promise.resolve() - expect(observedAborted).toBe(true) + expect(aborted).toBe(true) }) - it('cleanup is idempotent: listener runs exactly once even with repeated triggers', async () => { + it('registers cleanup when Router attaches server SSR', () => { const queryClient = track(new QueryClient()) - const { - router, - triggerCleanup, - attachServerSsr, - setDehydrated, - cleanupListenerCount, - } = createServerRouter() - - router.serverSsr = undefined + const { router, attachServerSsr, triggerCleanup } = createServerRouter() + setupCoreRouterSsrQueryIntegration({ router: router as any, queryClient, }) + queryClient.setQueryData(['before-attach'], 'data') attachServerSsr() - setDehydrated(true) - - // Cleanup was registered at setup because serverSsr was already attached. - await queryClient.fetchQuery({ - queryKey: ['x'], - queryFn: () => 'data', - }) - expect(cleanupListenerCount()).toBe(1) - expect(queryClient.getQueryData(['x'])).toBe('data') - triggerCleanup() - // Cache cleared by teardown. - expect(queryClient.getQueryData(['x'])).toBeUndefined() - // Second trigger after listeners already drained: no throw, no - // re-registration (cleanupRegistered flag prevents re-subscribe). - expect(() => triggerCleanup()).not.toThrow() - expect(cleanupListenerCount()).toBe(0) + expect(queryClient.getQueryCache().getAll()).toEqual([]) }) +}) - it('registers cleanup when serverSsr attaches after subscriber setup', async () => { - // Simulate user code prepopulating the cache inside getRouter() BEFORE - // attachRouterServerSsrUtils() runs. queryCache subscriber fires while - // serverSsr is undefined; cleanup still registers at attach time, before - // router.load() can throw in beforeLoad. - const queryClient = track(new QueryClient()) - const { - router, - triggerCleanup, - attachServerSsr, - setDehydrated, - cleanupListenerCount, - } = createServerRouter() - // Detach to simulate pre-attach state. - router.serverSsr = undefined +const gcAvailable = typeof (globalThis as any).gc === 'function' +const gcTestsEnabled = process.env.RUN_SSR_GC_TESTS === '1' && gcAvailable + +async function forceGc() { + for (let index = 0; index < 6; index++) { + ;(globalThis as any).gc() + await new Promise((resolve) => setTimeout(resolve, 0)) + } +} + +describe.runIf(gcTestsEnabled)('SSR memory', () => { + it('releases the request QueryClient and Router after cleanup', async () => { + let queryClient: QueryClient | null = new QueryClient({ + defaultOptions: { queries: { gcTime: 5 * 60 * 1000 } }, + }) + let fixture: ReturnType | null = + createServerRouter() setupCoreRouterSsrQueryIntegration({ - router: router as any, + router: fixture.router as any, queryClient, }) + fixture.attachServerSsr() + queryClient.setQueryData(['data'], 'value') + const dehydrated = (await fixture.router.options.dehydrate?.()) as { + query: { stream: ReadableStream> } + } + const streamedQueriesPromise = readStream(dehydrated.query.stream) + fixture.finishRender() + await streamedQueriesPromise - // Prepopulate before attach; queryCache event fires while serverSsr is - // undefined. This must not be the registration point. - await queryClient.fetchQuery({ - queryKey: ['pre'], - queryFn: () => 'early', - }) - expect(cleanupListenerCount()).toBe(0) + const queryClientRef = new WeakRef(queryClient) + const routerRef = new WeakRef(fixture.router) + fixture.triggerCleanup() + queryClient = null + fixture = null - // attachRouterServerSsrUtils equivalent. No onBeforeLoad/dehydrate needed. - attachServerSsr() - setDehydrated(true) - expect(cleanupListenerCount()).toBe(1) + await forceGc() - triggerCleanup() - expect(queryClient.getQueryData(['pre'])).toBeUndefined() + expect(queryClientRef.deref()).toBeUndefined() + expect(routerRef.deref()).toBeUndefined() }) }) diff --git a/packages/router-ssr-query-core/tests/server-lifecycle.bench.ts b/packages/router-ssr-query-core/tests/server-lifecycle.bench.ts new file mode 100644 index 00000000000..bf115f61fba --- /dev/null +++ b/packages/router-ssr-query-core/tests/server-lifecycle.bench.ts @@ -0,0 +1,141 @@ +// @vitest-environment node + +import { QueryClient } from '@tanstack/query-core' +import { afterAll, bench, describe } from 'vitest' +import { setupCoreRouterSsrQueryIntegration } from '../src' + +let benchmarkSink = 0 + +describe('server request lifecycle', () => { + bench('create and close a ReadableStream', () => { + let controller!: ReadableStreamDefaultController + const stream = new ReadableStream({ + start(value) { + controller = value + }, + }) + controller.close() + benchmarkSink += stream.locked ? 1 : 0 + }) + + const subscriptionClient = new QueryClient() + const queryCache = subscriptionClient.getQueryCache() + const listener = () => {} + afterAll(() => subscriptionClient.clear()) + + bench('subscribe and unsubscribe from QueryCache', () => { + const unsubscribe = queryCache.subscribe(listener) + unsubscribe() + }) + + bench('setup and cleanup before dehydrate', () => { + const queryClient = new QueryClient() + const fixture = createServerRouter() + + setupCoreRouterSsrQueryIntegration({ + router: fixture.router as any, + queryClient, + }) + fixture.attach() + fixture.cleanup() + benchmarkSink += fixture.cleanupListenerCount() + }) + + bench('setup and dehydrate an empty request', async () => { + const queryClient = new QueryClient() + const fixture = createServerRouter() + + setupCoreRouterSsrQueryIntegration({ + router: fixture.router as any, + queryClient, + }) + fixture.attach() + const dehydrated = await fixture.router.options.dehydrate?.() + fixture.finishRender() + fixture.cleanup() + benchmarkSink += dehydrated?.query.stream.locked ? 1 : 0 + }) + + bench('write 100 loader queries without integration', () => { + const queryClient = new QueryClient() + populateQueryClient(queryClient, 100) + benchmarkSink += queryClient.getQueryCache().getAll().length + queryClient.clear() + }) + + bench('write 100 loader queries before dehydrate', () => { + const queryClient = new QueryClient() + const fixture = createServerRouter() + + setupCoreRouterSsrQueryIntegration({ + router: fixture.router as any, + queryClient, + }) + fixture.attach() + populateQueryClient(queryClient, 100) + benchmarkSink += queryClient.getQueryCache().getAll().length + fixture.cleanup() + }) +}) + +function populateQueryClient(queryClient: QueryClient, queryCount: number) { + for (let index = 0; index < queryCount; index++) { + queryClient.setQueryData(['query', index], `data-${index}`) + } +} + +function createServerRouter() { + const renderFinishedListeners = new Array<() => void>() + const cleanupListeners = new Array<() => void>() + const serverSsr = { + onRenderFinished: (listener: () => void) => { + renderFinishedListeners.push(listener) + }, + onCleanup: (listener: () => void) => { + cleanupListeners.push(listener) + }, + } + const router = { + isServer: true, + options: {} as { + dehydrate?: () => + | { + query: { stream: ReadableStream } + } + | Promise<{ + query: { stream: ReadableStream } + }> + }, + serverSsr: undefined as typeof serverSsr | undefined, + serverSsrLifecycle: undefined as + | { + onServerSsrAttach: Array<(value: typeof serverSsr) => void> + } + | undefined, + } + + return { + router, + attach() { + router.serverSsr = serverSsr + router.serverSsrLifecycle?.onServerSsrAttach.forEach((listener) => { + listener(serverSsr) + }) + }, + finishRender() { + for (const listener of renderFinishedListeners.splice(0)) { + listener() + } + }, + cleanup() { + for (const listener of cleanupListeners.splice(0)) { + listener() + } + }, + cleanupListenerCount() { + return cleanupListeners.length + }, + } +} + +void benchmarkSink