feat(pinia-colada,tanstack-query,swr)!: redesign streamed query refetch and SSR semantics

apps/content/blog/nextjs-ssr-orpc-tanstack-query.mdx

@@ -415,10 +415,10 @@ Wire it into the page, inside its own suspense boundary:
 </Suspense>
 ```
 
-Think about what SSR means for a query that never completes. Two problems:
+A query that never completes causes two SSR problems:
 
-1. **On the server**, waiting for the stream to end would block rendering forever.
-2. **In the browser**, the hydrated reading is a static snapshot, and `staleTime` marks it fresh, so TanStack Query sees no reason to refetch. Nothing reopens the stream, and the widget silently freezes at the SSR value.
+1. **On the server**, waiting for the stream to end blocks rendering forever.
+2. **In the browser**, `staleTime` marks the hydrated reading fresh, so nothing reopens the stream. The widget freezes at the SSR value.
 
 `StreamQuerySSRUtilsPlugin` fixes both. Register it once as a [plugin](/docs/integrations/tanstack-query#plugins) in `lib/orpc.ts`:
 
@@ -430,7 +430,7 @@ export const orpc = createTanstackQueryUtils(client, {
 })
 ```
 
-On the server, the plugin settles live queries on their first event and closes the stream, so the HTML ships with a real temperature in it. In the browser, it sets `refetchOnMount: 'always'` on streamed and live queries, so the client reopens the stream after hydration. It also gives streamed queries `initialData: []`, so they render an empty list instead of suspending or reporting `pending` before their stream connects. Per-call options still override anything the plugin sets. The [Streamed and Live Queries](/docs/integrations/tanstack-query#streamed-and-live-queries) docs cover the details.
+On the server, the plugin stops a live query at its first event (a streamed query once it connects) and closes the stream. The HTML ships with a real temperature. The plugin also sets `refetchOnMount: 'always'`, so the browser reopens the stream on mount while the server value stays on screen. See [Streamed and Live Queries](/docs/integrations/tanstack-query#streamed-and-live-queries) for details.
 
 Reload the page one last time and check your work:
 

apps/content/docs/integrations/pinia-colada.mdx

@@ -61,46 +61,63 @@ Options accept plain values only. For reactive inputs, pass a callback to `useQu
 
 ## Streamed Query Options Utility
 
-Use `.streamedOptions` to build streamed query options for an [AsyncIteratorObject](/docs/async-iterator-object). The resulting data is an array of chunks, and each new chunk is appended as it arrives. It works with `useQuery` and any other API that accepts query options.
+Use `.streamedOptions` for a procedure that returns an [AsyncIteratorObject](/docs/async-iterator-object). The data is an array of chunks: each new chunk is appended, and `maxChunks` keeps only the latest.
 
 ```ts
 const query = useQuery(
   orpc.streamed.streamedOptions({
-    input: { id: 123 }, // Specify input if needed
-    context: { cache: true }, // Provide client context if needed
-    fnOptions: {
-      // Configure streamed query behavior
-      refetchMode: 'reset',
-      maxChunks: 3,
-    },
-    // additional options...
+    input: { id: 123 },
+    fnOptions: { refetchMode: 'append', maxChunks: 3 },
   }),
 )
 ```
 
-:::info
-`refetchMode` determines how data is handled when the query is fetched again:
+On refetch, existing data stays until the new stream connects. Then `refetchMode` decides what happens to it:
 
-- `'reset'` _(default)_: Clears existing data and returns the query to a pending state.
-- `'append'`: Adds new streamed chunks to the existing data.
-- `'replace'`: Buffers streamed data and replaces the cache after the stream completes.
+| `refetchMode`         | When the stream connects             |
+| --------------------- | ------------------------------------ |
+| `'reset'` _(default)_ | Data becomes `[]`                    |
+| `'append'`            | Data is kept, trimmed to `maxChunks` |
 
-:::
+:::warning
+`refetchMode` and `maxChunks` are part of the query key (when not the default):
+
+- Queries with different options get separate cache entries and separate streams. To share one, use the same options, and show fewer chunks when reading, e.g. `computed(() => data.value?.slice(-3))`.
+- To read or write the cache by key, pass the same options: `streamedKey({ input, fnOptions: { refetchMode, maxChunks } })`.
+  :::
 
 ## Live Query Options Utility
 
-Use `.liveOptions` to build live query options for an [AsyncIteratorObject](/docs/async-iterator-object). The data always reflects the latest chunk, replacing the previous value whenever a new one arrives. It works with `useQuery` and any other API that accepts query options.
+Use `.liveOptions` for a procedure that returns an [AsyncIteratorObject](/docs/async-iterator-object). The data is always the latest chunk.
 
 ```ts
 const query = useQuery(
   orpc.live.liveOptions({
-    input: { id: 123 }, // Specify input if needed
-    context: { cache: true }, // Provide client context if needed
-    // additional options...
+    input: { id: 123 },
   }),
 )
 ```
 
+### Settle on First Data
+
+By default, a streamed or live fetch resolves only when its stream ends. A stream also closes once the last component using it unmounts. Set `settleOnFirstData` to resolve at the first data instead: when a streamed query connects, or on a live query's first event. The stream then closes and the data is kept.
+
+| `settleOnFirstData` | Settles at first data                               |
+| ------------------- | --------------------------------------------------- |
+| `true`              | Always                                              |
+| `'unobserved'`      | Only if no component or active scope uses the query |
+| `false` _(default)_ | Never                                               |
+
+```ts
+const options = orpc.live.liveOptions({
+  input: { id: 123 },
+  settleOnFirstData: 'unobserved',
+  staleTime: 0,
+})
+```
+
+Use `staleTime: 0` so the stream reopens when a component mounts the query. `settleOnFirstData` is not part of the query key, and is ignored with a custom `query`.
+
 ## Infinite Query Options Utility
 
 Use `.infiniteOptions` to build infinite query options. It works with `useInfiniteQuery` and any other API that accepts infinite query options.
@@ -387,6 +404,39 @@ app.use(PiniaColada, {
 
 </Expandable>
 
+## Streamed and Live Queries in SSR
+
+`StreamQuerySSRUtilsPlugin` renders [streamed](#streamed-query-options-utility) and [live](#live-query-options-utility) queries on the server without waiting for the stream to end, then reopens the streams in the browser.
+
+```ts
+import { StreamQuerySSRUtilsPlugin } from '@orpc/pinia-colada'
+
+export const orpc = createPiniaColadaUtils(client, {
+  plugins: [new StreamQuerySSRUtilsPlugin()],
+})
+```
+
+It sets these defaults on `.streamedOptions` and `.liveOptions`. Per-call and [default options](#default-options) override them.
+
+| Option                                       | Server | Browser        |
+| -------------------------------------------- | ------ | -------------- |
+| [`settleOnFirstData`](#settle-on-first-data) | `true` | `'unobserved'` |
+| `staleTime`                                  | -      | `0`            |
+
+- The server renders the first data.
+- In the browser, mounting the query reopens the stream while the hydrated data stays visible.
+- A fetch no component uses yet, such as `queryCache.fetch` in a navigation guard, resolves at the first data. Pass `settleOnFirstData: false` to wait for the stream to end.
+
+To make this work:
+
+- On the server, call `useQuery` in a component setup, so the fetch runs in `onServerPrefetch`.
+- Keep `refetchOnMount` at its default. `'always'` restarts a stream that is already open.
+- An environment without `window` (or Deno) counts as the server. Pass `isServer` to override.
+
+:::info
+With `staleTime: 0`, a finite stream that has ended refetches when the window regains focus. Set `refetchOnWindowFocus: false` to avoid this.
+:::
+
 ## Client Context
 
 When a client is invoked through the Pinia Colada integration, an **operation context** is automatically added to the [client context](/docs/client/client-side#client-context). You can use this context to configure request behavior, such as selecting the HTTP method for [RPC Link](/docs/rpc/link#request-method).

apps/content/docs/integrations/swr.mdx

@@ -84,10 +84,19 @@ import useSWRSubscription from 'swr/subscription'
 
 const { data, error } = useSWRSubscription(
   orpc.streamed.key({ input: { id: 3 } }),
-  orpc.streamed.subscriber({ context: { cache: true }, maxChunks: 10 }), // Provide client context if needed
+  orpc.streamed.subscriber({ maxChunks: 10 }),
 )
 ```
 
+When the subscription restarts, existing data stays until the new stream connects. Then `refetchMode` decides:
+
+| `refetchMode`         | At connect                           |
+| --------------------- | ------------------------------------ |
+| `'reset'` _(default)_ | Data becomes `[]`                    |
+| `'append'`            | Data is kept, trimmed to `maxChunks` |
+
+Each chunk is then appended. `maxChunks` keeps only the latest chunks.
+
 Use `.liveSubscriber` to subscribe to the latest events without chunking:
 
 ```ts

apps/content/docs/integrations/tanstack-query.mdx

@@ -74,52 +74,72 @@ const query = useQuery(
 
 ## Streamed Query Options
 
-Use `.streamedOptions` to build streamed query options for an [AsyncIteratorObject](/docs/async-iterator-object). The resulting data is an array of events, and each new event is appended as it arrives.
-
-It works with `useQuery`, `useSuspenseQuery`, and `prefetchQuery`, and any other API that accepts query options.
+Use `.streamedOptions` to build query options for an [AsyncIteratorObject](/docs/async-iterator-object). The data is an array of events, and each new event is appended to it. It works with `useQuery`, `useSuspenseQuery`, `prefetchQuery`, and any other API that accepts query options.
 
 ```ts
 const query = useQuery(
   orpc.streamed.streamedOptions({
-    input: { id: 123 }, // Specify input if needed
-    context: { cache: true }, // Provide client context if needed
+    input: { id: 123 },
     queryFnOptions: {
-      // Configure streamed query behavior
-      refetchMode: 'reset',
-      maxChunks: 3,
+      refetchMode: 'append',
+      maxChunks: 3, // Keep only the last 3 events
     },
-    retry: true, // Infinite retry for more reliable streaming
-    // additional options...
+    retry: true, // Retry forever to keep the stream alive
   }),
 )
 ```
 
 :::info
-`refetchMode` determines how data is handled when the query is fetched again:
+On refetch, existing data stays until the new stream connects. Then `refetchMode` decides:
 
-- `'reset'` _(default)_: Clears existing data and returns the query to a pending state.
-- `'append'`: Adds new streamed chunks to the existing data.
-- `'replace'`: Buffers streamed data and replaces the cache after the stream completes.
+- `'reset'` _(default)_: data becomes `[]`.
+- `'append'`: data is kept, trimmed to `maxChunks`.
 
+For a loading state, call `queryClient.resetQueries({ queryKey })` before refetching.
 :::
 
-## Live Query Options
+:::warning
+`refetchMode` and `maxChunks` are part of the query key (when not the default):
+
+- Queries with different options get separate cache entries and separate streams. To share one, use the same options, and show fewer chunks with `select`, e.g. `select: data => data.slice(-3)`.
+- To read or write the cache by key, pass the same options: `streamedKey({ input, queryFnOptions: { refetchMode, maxChunks } })`.
+  :::
 
-Use `.liveOptions` to build live query options for an [AsyncIteratorObject](/docs/async-iterator-object). The data always reflects the latest event, replacing the previous value whenever a new one arrives.
+## Live Query Options
 
-It works with `useQuery`, `useSuspenseQuery`, and `prefetchQuery`, and any other API that accepts query options.
+Use `.liveOptions` to build query options for an [AsyncIteratorObject](/docs/async-iterator-object). The data is always the latest event. An `undefined` event rejects the fetch, since TanStack Query cannot store `undefined`.
 
 ```ts
 const query = useQuery(
   orpc.live.liveOptions({
-    input: { id: 123 }, // Specify input if needed
-    context: { cache: true }, // Provide client context if needed
-    retry: true, // Infinite retry for more reliable streaming
-    // additional options...
+    input: { id: 123 },
+    retry: true, // Retry forever to keep the stream alive
   }),
 )
 ```
 
+### Settle on First Data
+
+By default, a streamed or live query follows its stream until it ends. So awaiting an endless stream (`fetchQuery`, `ensureQueryData`, route loaders) never resolves.
+
+Pass `settleOnFirstData` to settle the fetch at the first data instead: when a streamed query connects, or on a live query's first event. The stream closes and the data stays, marked stale. The stream reopens once the query is observed, for example on mount.
+
+- `true`: always settle.
+- `'unobserved'`: settle only if nothing observes the query, such as a mounted `useQuery`.
+- `false` _(default)_: never settle.
+
+```ts
+const query = useSuspenseQuery(
+  orpc.live.liveOptions({
+    input: { id: 123 },
+    settleOnFirstData: 'unobserved',
+  }),
+)
+```
+
+- Not part of the query key, and ignored with a custom `queryFn`.
+- A stream cut short (cancelled, or its last observer unmounts) is marked stale, so it reopens when observed again.
+
 ## Infinite Query Options
 
 Use `.infiniteOptions` to build infinite query options. It works with `useInfiniteQuery`, `useSuspenseInfiniteQuery`, and `prefetchInfiniteQuery`, and any other API that accepts infinite query options.
@@ -570,7 +590,7 @@ const queryClient = new QueryClient({
 
 ### Streamed and Live Queries
 
-`StreamQuerySSRUtilsPlugin` lets [streamed](#streamed-query-options) and [live](#live-query-options) queries render on the server without blocking forever: streamed queries are cancelled once they connect and live queries on their first event, keeping the data received so far. In the browser, they refetch on mount to reopen the stream after hydration.
+Add `StreamQuerySSRUtilsPlugin` so [streamed](#streamed-query-options) and [live](#live-query-options) queries render on the server without blocking forever.
 
 ```ts
 import { StreamQuerySSRUtilsPlugin } from '@orpc/tanstack-query'
@@ -580,4 +600,19 @@ export const orpc = createTanstackQueryUtils(client, {
 })
 ```
 
-An environment without `window` counts as the server; pass `isServer` to override.
+It sets the [`settleOnFirstData`](#settle-on-first-data) option of `.streamedOptions` and `.liveOptions`:
+
+- Server: `true`. The query is dehydrated with its first data.
+- Browser: `'unobserved'`. A mounted `useQuery` follows the stream, as does `suspense()` in Vue, Svelte, and Solid.
+
+In the browser it also sets `refetchOnMount: 'always'`, so after hydration, mounting reopens the stream (even with a `staleTime`) while the hydrated data stays on screen. A finished finite stream also reruns on each mount; pass `refetchOnMount: true` to respect `staleTime` instead. On the server it sets `refetchOnMount: false`, so components sharing a settled query do not reopen it.
+
+Per-call and [default options](#default-options) override the plugin. The server is detected with TanStack Query's `environmentManager.isServer()`; pass `isServer` to override.
+
+:::warning
+
+- In the browser, `await queryClient.fetchQuery(orpc.x.streamedOptions())` resolves once the stream connects, unless a mounted query already follows that stream, in which case it waits for that stream. Pass `settleOnFirstData: false` to wait for the stream to end.
+- Avoid `staleTime: 'static'`, which stops the stream from reopening on mount.
+- When [streaming](https://tanstack.com/query/latest/docs/framework/react/guides/advanced-ssr#streaming-with-server-components) pending queries, read them with `useSuspenseQuery`. A `useQuery` that mounts before the hydrated data arrives keeps that data until it goes stale.
+- Do not mix suspense and non-suspense consumers of one streamed key across suspense boundaries. A mounted `useQuery` keeps the suspending fetch from settling, so the stream stays open.
+  :::

apps/content/docs/migrations/from-v1.mdx

@@ -1153,10 +1153,25 @@ orpc.live.experimental_liveOptions({ input: {} })
 
 `experimental_defaults` became `scoped`, and the hydration serializer changed from `StandardRPCJsonSerializer` to `RPCJsonSerializer` (see [RPC JSON Serializer](/docs/rpc/serializer#rpc-json-serializer)).
 
+Streamed and live queries changed (see [Streamed Query Options](/docs/integrations/tanstack-query#streamed-query-options)):
+
+- `refetchMode: 'reset'` keeps the old data until the new stream connects, then sets it to `[]`. (v1 cleared it immediately, returning the query to `pending`.)
+- `.liveOptions` rejects an `undefined` event, which TanStack Query cannot store.
+- `refetchMode: 'replace'` was removed. To keep the old data until the stream ends, collect the stream in a custom `queryFn`:
+
+```ts
+orpc.streamed.streamedOptions({
+  input: {},
+  queryFn: async ({ signal }) => Array.fromAsync(await client.streamed({}, { signal })),
+})
+```
+
 ### SWR and Pinia Colada
 
 `@orpc/experimental-react-swr` is now `@orpc/swr` (see [SWR Integration](/docs/integrations/swr)), and `@orpc/vue-colada` is now `@orpc/pinia-colada` with `createORPCVueColadaUtils` renamed to `createPiniaColadaUtils` (see [Pinia Colada Integration](/docs/integrations/pinia-colada)). Both switched from `path` to `prefix`, same as TanStack Query.
 
+In SWR, `.subscriber` now writes the data when the stream connects: `[]` with `refetchMode: 'reset'` (instead of `undefined`), and the existing data trimmed to `maxChunks` with `'append'`.
+
 ### NestJS
 
 Import `implement`, `ORPCError`, and `onError` from `@orpc/server` instead of `@orpc/nest`, and augment `DefaultInitialContext` instead of `ORPCGlobalContext`. See [NestJS Integration](/docs/integrations/nest).

packages/pinia-colada/package.json

@@ -59,7 +59,7 @@
     "zod": "^4.6.7"
   },
   "peerDependencies": {
-    "@pinia/colada": ">=1.2.0",
+    "@pinia/colada": ">=1.4.9",
     "vue": ">=3.5.17"
   }
 }

packages/pinia-colada/src/index.test.ts

@@ -1,7 +1,8 @@
-it('exports createPiniaColadaUtils, PINIA_COLADA_OPERATION_CONTEXT_SYMBOL', async () => {
+it('exports createPiniaColadaUtils, PINIA_COLADA_OPERATION_CONTEXT_SYMBOL, StreamQuerySSRUtilsPlugin', async () => {
   await expect(import('./index.ts')).resolves.toMatchObject({
     createPiniaColadaUtils: expect.any(Function),
     createRouterUtils: expect.any(Function),
     PINIA_COLADA_OPERATION_CONTEXT_SYMBOL: expect.any(Symbol),
+    StreamQuerySSRUtilsPlugin: expect.any(Function),
   })
 })

packages/pinia-colada/src/index.ts

@@ -31,6 +31,8 @@ export {
 } from './router-utils.ts'
 export { createRouterUtils as createPiniaColadaUtils } from './router-utils.ts'
 export { SharedUtils } from './shared-utils.ts'
+export * from './ssr.ts'
+export type { SettleOnFirstDataOptions } from './stream-cache.ts'
 export type { SerializableStreamedQueryOptions } from './stream-query.ts'
 export * from './types.ts'
 export {

packages/pinia-colada/src/key.test.ts

@@ -43,6 +43,24 @@ it('generateOperationKey', () => {
   ).toEqual(['__prefix__', ['path'], { type: 'query', input: { a: 1 } }])
 })
 
+it('generateOperationKey keeps only non-default streamed fnOptions', () => {
+  expect(
+    generateOperationKey(['planet', 'stream'], {
+      type: 'streamed',
+      fnOptions: { refetchMode: 'append', maxChunks: 3 },
+    }),
+  ).toEqual([
+    ['planet', 'stream'],
+    { type: 'streamed', fnOptions: { refetchMode: 'append', maxChunks: 3 } },
+  ])
+  expect(
+    generateOperationKey(['planet', 'stream'], {
+      type: 'streamed',
+      fnOptions: { refetchMode: 'reset', maxChunks: Number.POSITIVE_INFINITY },
+    }),
+  ).toEqual([['planet', 'stream'], { type: 'streamed' }])
+})
+
 describe('generateOperationKey input meta', () => {
   it('keys meta by JSON-encoded path', () => {
     expect(