diff --git a/CHANGELOG.md b/CHANGELOG.md index a28dc41..b837b13 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,27 @@ # Changelog +## Unreleased + +- Require Convex 1.39 or newer. +- Use `@convex-dev/stream` for all newly created streams while preserving + existing stream IDs and persisted text without a migration. +- Keep every existing component, client, and React signature unchanged. +- Add `readStream(ctx, streamId, streamArgs)` and the matching `useStream` + option. Expose it from an app-owned query so followers subscribe to bounded + append-only pages instead of re-reading the whole body on every append. + Authorize that query as you already authorize `getStreamBody`. +- Elect one producer atomically; a duplicate drive request receives `205` and + reads durably instead of starting a second producer. +- Bound persistence by time and bytes, serialize writes, and publish React text + at a fixed cadence. +- Recover a failed drive connection through the same durable read, without + rewinding text already on screen. +- Stop restarting the transport when an auth token rotates mid-stream. +- Cap drive retries so a persistently failing endpoint falls back instead of + polling indefinitely. +- Retain `getStreamBody` and the React query argument for apps that do not adopt + `readStream`. + ## 0.3.3 - Update ctx types for convex@1.41+ diff --git a/README.md b/README.md index a261408..914387f 100644 --- a/README.md +++ b/README.md @@ -4,214 +4,290 @@ -This Convex component enables persistent text streaming. It provides a React -hook for streaming text from HTTP actions while simultaneously storing the data -in the database. This persistence allows the text to be accessed after the -stream ends or by other users. +Persistent Text Streaming sends generated text to one browser with low latency, +stores it in Convex, and lets other browsers follow the durable stream. It is +designed for AI chat, but it works with any text producer. -The most common use case is for AI chat applications. The example app (found in -the `example` directory) is a just such a simple chat app that demonstrates use -of the component. +The component uses two paths behind the existing API: -Here's what you'll end up with! The left browser window is streaming the chat -body to the client, and the right browser window is subscribed to the chat body -via a database query. The message is only updated in the database on sentence -boundaries, whereas the HTTP stream sends tokens as they come: +- The driving browser receives raw text over HTTP as the producer emits it. +- Everyone else -- followers, reloads, and recovery -- subscribes to bounded + append-only pages over the normal Convex websocket. No HTTP request, and no + resending the whole body on every append. -![example-animation](./anim.gif) +An atomic claim allows only one request to run the producer. A duplicate driving +request receives the existing `205` response and reads durably instead. -## Pre-requisite: Convex +![example-animation](./anim.gif) -You'll need an existing Convex project to use the component. Convex is a hosted -backend platform, including a database, serverless functions, and a ton more you -can learn about [here](https://docs.convex.dev/get-started). +## Prerequisite -Run `npm create convex` or follow any of the -[quickstarts](https://docs.convex.dev/home) to set one up. +Add this component to an existing [Convex](https://convex.dev) project. You can +create one with `npm create convex` or follow a +[Convex quickstart](https://docs.convex.dev/home). ## Installation -See [`example/`](./example/convex/) for a working demo. - -1. Install the Persistent Text Streaming component: +Install the package: ```bash npm install @convex-dev/persistent-text-streaming ``` -2. Create a [`convex.config.ts`](./example/convex/convex.config.ts) file in your - app's `convex/` folder and install the component by calling `use`: +This release requires Convex 1.39 or newer. + +Register the component in `convex/convex.config.ts`: ```ts -// convex/convex.config.ts -import { defineApp } from "convex/server"; import persistentTextStreaming from "@convex-dev/persistent-text-streaming/convex.config.js"; +import { defineApp } from "convex/server"; const app = defineApp(); app.use(persistentTextStreaming); export default app; ``` -## Usage +See [`example/`](./example/) for a complete app. -Here's a simple example of how to use the component: +## Backend setup -In `convex/chat.ts`: +Create one client for the component: ```ts -const persistentTextStreaming = new PersistentTextStreaming( +// convex/streaming.ts +import { + PersistentTextStreaming, + StreamId, + StreamIdValidator, +} from "@convex-dev/persistent-text-streaming"; +import { streamQueryArgsValidator } from "@convex-dev/stream"; +import { components } from "./_generated/api"; +import { query } from "./_generated/server"; + +export const streaming = new PersistentTextStreaming( components.persistentTextStreaming, ); -// Create a stream using the component and store the id in the database with -// our chat message. -export const createChat = mutation({ - args: { - prompt: v.string(), - }, - handler: async (ctx, args) => { - const streamId = await persistentTextStreaming.createStream(ctx); - const chatId = await ctx.db.insert("chats", { - title: "...", - prompt: args.prompt, - stream: streamId, - }); - return chatId; - }, +// The full-body query remains part of the public API for history and server +// logic. The React hook uses it only when `readStream` is not provided. +export const getStreamBody = query({ + args: { streamId: StreamIdValidator }, + handler: async (ctx, { streamId }) => + streaming.getStreamBody(ctx, streamId as StreamId), }); -// Create a query that returns the chat body. -export const getChatBody = query({ - args: { - streamId: StreamIdValidator, - }, - handler: async (ctx, args) => { - return await persistentTextStreaming.getStreamBody( - ctx, - args.streamId as StreamId, - ); +// Followers and recovery subscribe here. Authorize the caller exactly as you +// do for getStreamBody -- this query returns persisted assistant text. +export const readStream = query({ + args: { streamId: StreamIdValidator, streamArgs: streamQueryArgsValidator }, + handler: async (ctx, { streamId, streamArgs }) => + streaming.readStream(ctx, streamId as StreamId, streamArgs), +}); +``` + +Create a stream and store its opaque ID in an app-owned record: + +```ts +export const createChat = mutation({ + args: { prompt: v.string() }, + handler: async (ctx, { prompt }) => { + const streamId = await streaming.createStream(ctx); + return ctx.db.insert("chats", { prompt, streamId }); }, }); +``` + +Create the HTTP action that produces text: -// Create an HTTP action that generates chunks of the chat body -// and uses the component to stream them to the client and save them to the database. +```ts export const streamChat = httpAction(async (ctx, request) => { - const body = (await request.json()) as { streamId: string }; - const generateChat = async (ctx, request, streamId, chunkAppender) => { - await chunkAppender("Hi there!"); - await chunkAppender("How are you?"); - await chunkAppender("Pretend I'm an AI or something!"); - }; - - const response = await persistentTextStreaming.stream( + const { streamId } = (await request.json()) as { streamId: string }; + + // Required in production: authenticate the caller and verify that this + // stream belongs to a record the caller may read or generate. Do this before + // calling stream(). The component cannot infer your app's ownership rules. + + const response = await streaming.stream( ctx, request, - body.streamId as StreamId, - generateChat, + streamId as StreamId, + async (_ctx, _request, _streamId, append) => { + await append("Hi there! "); + await append("How are you?"); + }, ); - // Set CORS headers appropriately. response.headers.set("Access-Control-Allow-Origin", "*"); response.headers.set("Vary", "Origin"); return response; }); ``` -You need to expose this HTTP endpoint in your backend, so in `convex/http.ts`: +`stream()` keeps the same signature: ```ts -http.route({ - path: "/chat-stream", - method: "POST", - handler: streamChat, -}); +stream(ctx, request, streamId, writer): Promise +``` + +Only the driving browser calls this route, and only to generate text. It is not +a read endpoint. + +## HTTP route and CORS + +Register POST and OPTIONS routes: + +```ts +// convex/http.ts +import { httpRouter } from "convex/server"; +import { httpAction } from "./_generated/server"; +import { streamChat } from "./chat"; + +const http = httpRouter(); + +http.route({ path: "/chat-stream", method: "POST", handler: streamChat }); -// Handle CORS preflight requests so browsers will allow the POST above when -// your app is served from a different origin than your Convex deployment. http.route({ path: "/chat-stream", method: "OPTIONS", - handler: httpAction(async (_, request) => { - const headers = request.headers; - if ( - headers.get("Origin") !== null && - headers.get("Access-Control-Request-Method") !== null && - headers.get("Access-Control-Request-Headers") !== null - ) { - return new Response(null, { + handler: httpAction( + async () => + new Response(null, { headers: new Headers({ "Access-Control-Allow-Origin": "*", "Access-Control-Allow-Methods": "POST", "Access-Control-Allow-Headers": "Content-Type, Digest, Authorization", "Access-Control-Max-Age": "86400", }), - }); - } else { - return new Response(); - } - }), + }), + ), }); + +export default http; ``` -Finally, in your app, you can now create chats and them subscribe to them via -stream and/or database query as optimal: +Add every custom request header to `Access-Control-Allow-Headers`. Only the +driving browser reaches this route; followers never leave the websocket, so no +additional CORS configuration is required to read a stream. + +For production, replace `*` with your frontend origin. CORS controls browser +access; it does not authenticate or authorize a request. + +## React ```ts -// chat-input.tsx, maybe? -const createChat = useMutation(api.chat.createChat); -const formSubmit = async (e: React.FormEvent) => { - e.preventDefault(); - const chatId = await createChat({ - prompt: inputValue, - }); -}; - -// chat-message.tsx, maybe? import { useStream } from "@convex-dev/persistent-text-streaming/react"; -// ... - -// In our component: const { text, status } = useStream( - api.chat.getChatBody, // The query to call for the full stream body - new URL(`${convexSiteUrl}/chat-stream`), // The HTTP endpoint for streaming - driven, // True if this browser session created this chat and should generate the stream - chat.streamId as StreamId, // The streamId from the chat database record + api.streaming.getStreamBody, + new URL(`${convexSiteUrl}/chat-stream`), + driven, + chat.streamId as StreamId, + { + readStream: api.streaming.readStream, + authToken, + headers: { "X-Workspace": workspaceId }, + }, ); ``` -## Design Philosophy - -This component balances HTTP streaming with database persistence to try to -maximize the benefits of both. To understand why this balance is beneficial, -let's examine each approach in isolation. - -- **HTTP streaming only**: If your app _only_ uses HTTP streaming, then the - original browser that made the request will have a great, high-performance - streaming experience. But if that HTTP connection is lost, if the browser - window is reloaded, if other users want to view the same chat, or this users - wants to revisit the conversation later, it won't be possible. The - conversation is only ephemeral because it was never stored on the server. - -- **Database Persistence Only**: If your app _only_ uses database persistence, - it's true that the conversation will be available for as long as you want. - Additionally, Convex's subscriptions will ensure the chat message is updated - as new text chunks are generated. However, there are a few downsides: one, the - entire chat body needs to be resent every time it is changed, which is a lot - redundant bandwidth to push into the database and over the websockets to all - connected clients. Two, you'll need to make a difficult tradeoff between - interactivity and efficiency. If you write every single small chunk to the - database, this will get quite slow and expensive. But if you batch up the - chunks into, say, paragraphs, then the user experience will feel laggy. - -This component combines the best of both worlds. The original browser that makes -the request will still have a great, high-performance streaming experience. But -the chat body is also stored in the database, so it can be accessed by the -client even after the stream has finished, or by other users, etc. +Its signature is: + +```ts +useStream( + getPersistentBody, + streamUrl, + driven, + streamId, + opts?, +): StreamBody +``` + +- `getPersistentBody` returns the whole body in one query. It is the read path + when `opts.readStream` is absent. +- `driven` tells this browser to try the producer path. The atomic server claim, + not this boolean, decides which request may generate text. +- `opts.readStream` is the app-owned query described above. Provide it to read + bounded pages instead of the whole body on every append. +- `opts.authToken` sends `Authorization: Bearer ` on the drive request. +- `opts.headers` sends additional app headers on the drive request. + +A token that rotates mid-stream does not restart the transport or clear the text +already on screen. + +## Security + +Both entry points return or generate assistant text, and both need your own +authorization: + +- The HTTP action can invoke an LLM or another costly producer. +- `readStream` returns persisted assistant text. + +Authenticate the request, resolve the app record that stores `streamId`, and +verify the caller may access it. Never treat `driven` or possession of a +`StreamId` as authorization. The atomic claim prevents duplicate producers; it +is not an authorization check. + +`readStream` is an ordinary Convex query, so the check belongs in its handler +alongside the one you already have in `getStreamBody`. + +## Performance and recovery + +The driving response sends raw text immediately. Persistence runs through a +single ordered queue and flushes at sentence punctuation, after about 100 ms, or +at 16 KiB. Producer failures flush pending text before recording `error`. + +Followers subscribe to append-only pages of at most 16 events rather than to the +complete growing string. A query invalidation therefore reads and sends only the +delta, where the previous release re-read every chunk and re-sent the whole body +on every append. The React client publishes visible text at a fixed cadence of +about 50 ms rather than once per model token. + +If the raw drive connection fails, the driving browser switches to the same +durable read as every other client. The replay restarts at the beginning of the +stream but does not rewind what is already on screen: the raw text is a prefix +of the durable text, so it stays visible until the replay passes it. + +`getStreamBody()` still returns the complete stored text and status in one +Convex query for history, server logic, and compatibility. It is therefore +subject to Convex transaction and value limits and is not suitable for unbounded +output. Prefer `readStream` for live text. + +## Storage upgrades + +Upgrading requires no data migration and no API signature changes. Apps should +add the `readStream` query and pass it to `useStream`; without it the hook keeps +the previous full-body read behavior. + +- New streams store ordered events and lifecycle state through + `@convex-dev/stream`. +- Existing `StreamId` values remain valid. +- Streams created by older releases keep their legacy chunk rows and remain + readable, writable, followable, and deletable. +- In-flight legacy producers can finish while new clients follow their output. + +The component selects the storage path from the stable stream record. It does +not rewrite or discard existing assistant text. + +## Public API + +Every existing signature is unchanged; `readStream` and the matching `opts` +field are additive: + +```ts +new PersistentTextStreaming(component, options?) +createStream(ctx): Promise +getStreamBody(ctx, streamId): Promise +readStream(ctx, streamId, streamArgs): Promise> +stream(ctx, request, streamId, writer): Promise +deleteStream(ctx, streamId): Promise +useStream(getPersistentBody, streamUrl, driven, streamId, opts?): StreamBody +``` + +`StreamBody.status` is `pending`, `streaming`, `done`, `error`, or `timeout`. ## Background -This component is largely based on the Stack post +This component builds on [AI Chat with HTTP Streaming](https://stack.convex.dev/ai-chat-with-http-streaming). diff --git a/example/convex/chat.ts b/example/convex/chat.ts index 190c501..bb37b5b 100644 --- a/example/convex/chat.ts +++ b/example/convex/chat.ts @@ -11,6 +11,10 @@ export const streamChat = httpAction(async (ctx, request) => { streamId: string; }; + // This demo is public. Production apps must authenticate the caller and + // verify access to the app record that owns body.streamId before calling + // stream(), which starts generation and bills the producer. + // Start streaming and persisting at the same time while // we immediately return a streaming response to the client const response = await streamingComponent.stream( diff --git a/example/convex/occIntegration.ts b/example/convex/occIntegration.ts new file mode 100644 index 0000000..ce7b7ca --- /dev/null +++ b/example/convex/occIntegration.ts @@ -0,0 +1,190 @@ +import { ConvexError, v } from "convex/values"; + +import { components } from "./_generated/api"; +import { internalAction } from "./_generated/server"; + +const ROUNDS = 10; +const CONTENDERS = 8; + +type Outcome = { ok: true; value: T } | { ok: false; code: string }; + +function errorCode(error: unknown): string { + if ( + error instanceof ConvexError && + typeof error.data === "object" && + error.data !== null + ) { + const code = (error.data as { code?: unknown }).code; + if (typeof code === "string") return code; + } + if (typeof error === "object" && error !== null && "data" in error) { + const data = (error as { data: unknown }).data; + if (typeof data === "object" && data !== null && "code" in data) { + const code = (data as { code: unknown }).code; + if (typeof code === "string") return code; + } + } + const match = /["']code["']\s*:\s*["']([^"']+)["']/.exec(String(error)); + return match?.[1] ?? "unknown"; +} + +async function settle(operation: () => Promise): Promise> { + try { + return { ok: true, value: await operation() }; + } catch (error) { + return { ok: false, code: errorCode(error) }; + } +} + +const countersValidator = v.object({ won: v.number(), rejected: v.number() }); + +// This action is invoked only by scripts/run-occ-integration.mjs. It exercises +// real backend transactions; convex-test cannot model concurrent OCC retries. +export const run = internalAction({ + args: { runId: v.string() }, + returns: v.object({ + passed: v.boolean(), + rounds: v.number(), + contenders: v.number(), + claims: countersValidator, + writers: countersValidator, + deleteAppend: countersValidator, + failures: v.array(v.string()), + }), + handler: async (ctx, args) => { + if (!args.runId.startsWith("pts-occ:")) { + throw new ConvexError({ + code: "invalidArgument", + message: "runId must start with pts-occ:.", + }); + } + + let claimsWon = 0; + let claimsRejected = 0; + let writersWon = 0; + let writersRejected = 0; + let deleteAppendWon = 0; + let deleteAppendRejected = 0; + const failures: string[] = []; + + for (let round = 0; round < ROUNDS; round += 1) { + const streamId: string = await ctx.runMutation( + components.persistentTextStreaming.lib.createStream, + {}, + ); + const attempts = await Promise.all( + Array.from({ length: CONTENDERS }, async (_, contender) => { + const claim = await ctx.runMutation( + components.persistentTextStreaming.lib.claim, + { streamId }, + ); + if (!claim.claimed) { + return { claimed: false, wrote: false, error: null }; + } + try { + await ctx.runMutation( + components.persistentTextStreaming.lib.addChunk, + { + streamId, + text: `${args.runId}:${round}:${contender}`, + final: true, + }, + ); + return { claimed: true, wrote: true, error: null }; + } catch (error) { + return { claimed: true, wrote: false, error: errorCode(error) }; + } + }), + ); + const roundClaims = attempts.filter((attempt) => attempt.claimed).length; + const roundWriters = attempts.filter((attempt) => attempt.wrote).length; + claimsWon += roundClaims; + claimsRejected += CONTENDERS - roundClaims; + writersWon += roundWriters; + writersRejected += CONTENDERS - roundWriters; + + if (roundClaims !== 1 || roundWriters !== 1) { + failures.push( + `claim round ${round}: claims=${roundClaims}, writers=${roundWriters}`, + ); + } + for (const attempt of attempts) { + if (attempt.error !== null) { + failures.push( + `claim round ${round}: winning writer=${attempt.error}`, + ); + } + } + + const body: { text: string; status: string } = await ctx.runQuery( + components.persistentTextStreaming.lib.getStreamText, + { streamId }, + ); + if (body.status !== "done" || body.text.length === 0) { + failures.push( + `claim round ${round}: durable status=${body.status}, textLength=${body.text.length}`, + ); + } + await ctx.runMutation( + components.persistentTextStreaming.lib.deleteStream, + { + streamId, + }, + ); + + const deleteStreamId: string = await ctx.runMutation( + components.persistentTextStreaming.lib.createStream, + {}, + ); + const initialClaim = await ctx.runMutation( + components.persistentTextStreaming.lib.claim, + { streamId: deleteStreamId }, + ); + if (!initialClaim.claimed) { + failures.push(`delete/append round ${round}: setup claim lost`); + } + await ctx.runMutation(components.persistentTextStreaming.lib.addChunk, { + streamId: deleteStreamId, + text: "seed", + final: false, + }); + + const [append, deletion] = await Promise.all([ + settle(() => + ctx.runMutation(components.persistentTextStreaming.lib.addChunk, { + streamId: deleteStreamId, + text: "tail", + final: true, + }), + ), + settle(() => + ctx.runMutation(components.persistentTextStreaming.lib.deleteStream, { + streamId: deleteStreamId, + }), + ), + ]); + if (append.ok) deleteAppendWon += 1; + else if ( + append.code === "streamNotFound" || + append.code === "streamDeleting" + ) { + deleteAppendRejected += 1; + } else { + failures.push(`delete/append round ${round}: append=${append.code}`); + } + if (!deletion.ok) { + failures.push(`delete/append round ${round}: delete=${deletion.code}`); + } + } + + return { + passed: failures.length === 0, + rounds: ROUNDS, + contenders: CONTENDERS, + claims: { won: claimsWon, rejected: claimsRejected }, + writers: { won: writersWon, rejected: writersRejected }, + deleteAppend: { won: deleteAppendWon, rejected: deleteAppendRejected }, + failures, + }; + }, +}); diff --git a/example/convex/streaming.ts b/example/convex/streaming.ts index 915b4ed..4781c8d 100644 --- a/example/convex/streaming.ts +++ b/example/convex/streaming.ts @@ -3,6 +3,7 @@ import { StreamId, StreamIdValidator, } from "@convex-dev/persistent-text-streaming"; +import { streamQueryArgsValidator } from "@convex-dev/stream"; import { components } from "./_generated/api"; import { query } from "./_generated/server"; @@ -21,3 +22,20 @@ export const getStreamBody = query({ ); }, }); + +// Followers and recovery subscribe here. This demo is public; a production app +// authorizes the caller against the record that owns the stream, exactly as it +// would for getStreamBody. +export const readStream = query({ + args: { + streamId: StreamIdValidator, + streamArgs: streamQueryArgsValidator, + }, + handler: async (ctx, args) => { + return await streamingComponent.readStream( + ctx, + args.streamId as StreamId, + args.streamArgs, + ); + }, +}); diff --git a/example/src/components/ServerMessage.tsx b/example/src/components/ServerMessage.tsx index 3518238..35c1e3d 100644 --- a/example/src/components/ServerMessage.tsx +++ b/example/src/components/ServerMessage.tsx @@ -22,6 +22,7 @@ export function ServerMessage({ new URL(`${getConvexSiteUrl()}/chat-stream`), isDriven, message.responseStreamId as StreamId, + { readStream: api.streaming.readStream }, ); const isCurrentlyStreaming = useMemo(() => { diff --git a/package-lock.json b/package-lock.json index a3db66e..92caa0b 100644 --- a/package-lock.json +++ b/package-lock.json @@ -9,6 +9,7 @@ "version": "0.3.3", "license": "Apache-2.0", "dependencies": { + "@convex-dev/stream": "https://pkg.pr.new/get-convex/stream/@convex-dev/stream@b153faa", "convex-helpers": "^0.1.114" }, "devDependencies": { @@ -20,11 +21,12 @@ "@types/node": "20.19.39", "@types/react": "19.2.14", "@types/react-dom": "19.2.3", + "@types/react-test-renderer": "19.1.0", "@vitejs/plugin-react": "6.0.1", "chokidar-cli": "3.0.0", "clsx": "2.1.1", - "convex": "1.35.1", - "convex-test": "0.0.48", + "convex": "1.42.2", + "convex-test": "0.0.54", "eslint": "9.39.4", "eslint-plugin-react-hooks": "7.0.1", "eslint-plugin-react-refresh": "0.5.2", @@ -35,6 +37,7 @@ "react": "19.2.5", "react-dom": "19.2.5", "react-markdown": "10.1.0", + "react-test-renderer": "19.2.5", "tailwind-merge": "3.5.0", "tailwindcss": "4.2.2", "typescript": "5.9.3", @@ -43,7 +46,7 @@ "vitest": "4.1.4" }, "peerDependencies": { - "convex": "^1.32.0", + "convex": "^1.39.0", "react": "~18.3.1 || ^19.0.0", "react-dom": "~18.3.1 || ^19.0.0" } @@ -341,6 +344,24 @@ "convex": "^1.34.1" } }, + "node_modules/@convex-dev/stream": { + "version": "0.0.1", + "resolved": "https://pkg.pr.new/get-convex/stream/@convex-dev/stream@b153faa", + "integrity": "sha512-EEkdvHUpefPOEssl3dlfLfGOMsbNG7P583BHuTDxRxSpGvNvXau2ewg9aT4rV7TMkokHuFrDv/9MEt33uj31TA==", + "license": "Apache-2.0", + "dependencies": { + "convex-helpers": "0.1.119" + }, + "peerDependencies": { + "convex": "^1.39.0", + "react": "^18.3.1 || ^19.0.0" + }, + "peerDependenciesMeta": { + "react": { + "optional": true + } + } + }, "node_modules/@edge-runtime/primitives": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/@edge-runtime/primitives/-/primitives-6.0.0.tgz", @@ -1731,6 +1752,16 @@ "@types/react": "^19.2.0" } }, + "node_modules/@types/react-test-renderer": { + "version": "19.1.0", + "resolved": "https://registry.npmjs.org/@types/react-test-renderer/-/react-test-renderer-19.1.0.tgz", + "integrity": "sha512-XD0WZrHqjNrxA/MaR9O22w/RNidWR9YZmBdRGI7wcnWGrv/3dA8wKCJ8m63Sn+tLJhcjmuhOi629N66W6kgWzQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/react": "*" + } + }, "node_modules/@types/unist": { "version": "3.0.3", "resolved": "https://registry.npmjs.org/@types/unist/-/unist-3.0.3.tgz", @@ -2660,14 +2691,14 @@ "license": "MIT" }, "node_modules/convex": { - "version": "1.35.1", - "resolved": "https://registry.npmjs.org/convex/-/convex-1.35.1.tgz", - "integrity": "sha512-g23KrTjBiXqRHzWIN0PVFagKjrmFxWUaOSiBsAWPTpXX2rXl0L1F4PR0YpAcMJEzMgfZR9AGymJvLTM+KA6lsQ==", + "version": "1.42.2", + "resolved": "https://registry.npmjs.org/convex/-/convex-1.42.2.tgz", + "integrity": "sha512-5Mo9EZQ45bG2nbhO7NhTTw5GF9g4RA/kgRQviUQrOI+wFCyfdeNw3PjgOZc2eOM4ruum+0QmUcvVCX5mT2v4rg==", "license": "Apache-2.0", "dependencies": { "esbuild": "0.27.0", "prettier": "^3.0.0", - "ws": "8.18.0" + "ws": "8.21.0" }, "bin": { "convex": "bin/main.js" @@ -2679,7 +2710,7 @@ "peerDependencies": { "@auth0/auth0-react": "^2.0.1", "@clerk/clerk-react": "^4.12.8 || ^5.0.0", - "@clerk/react": "^6.0.0", + "@clerk/react": "^6.4.3", "react": "^18.0.0 || ^19.0.0-0 || ^19.0.0" }, "peerDependenciesMeta": { @@ -2698,9 +2729,9 @@ } }, "node_modules/convex-helpers": { - "version": "0.1.114", - "resolved": "https://registry.npmjs.org/convex-helpers/-/convex-helpers-0.1.114.tgz", - "integrity": "sha512-elEdh+gG6BDv2dWIWVvBeJPbHnDQS5+WexUuwlGVJXz1EbMkXz/UIQwFIfLMZIXUwW6ot4JYf/1JJKNStrE6lg==", + "version": "0.1.119", + "resolved": "https://registry.npmjs.org/convex-helpers/-/convex-helpers-0.1.119.tgz", + "integrity": "sha512-fGNK9KAlBLk8Un729ZXBqD9S5jy910nxSrH6PloIL/Gl6TalFJvjN8+xCCwahlox8Ft+bb54SSg9MbcrsQb98w==", "license": "Apache-2.0", "bin": { "convex-helpers": "bin.cjs" @@ -2710,7 +2741,7 @@ "convex": "^1.32.0", "hono": "^4.0.5", "react": "^17.0.2 || ^18.0.0 || ^19.0.0", - "typescript": "^5.5", + "typescript": "^5.5 || ^6.0.0", "zod": "^3.25.0 || ^4.0.0" }, "peerDependenciesMeta": { @@ -2732,9 +2763,9 @@ } }, "node_modules/convex-test": { - "version": "0.0.48", - "resolved": "https://registry.npmjs.org/convex-test/-/convex-test-0.0.48.tgz", - "integrity": "sha512-ewAkXwNJE0TpHAfHJt38NR6ZsArqdzd4HUy9BxegKENvupYwWCVJezFfkIJoYaZThqPABEfmiChvG8KCqCTm5w==", + "version": "0.0.54", + "resolved": "https://registry.npmjs.org/convex-test/-/convex-test-0.0.54.tgz", + "integrity": "sha512-C0v2SQcuxrELAJRNzE6fQ686XDPor7UFoC22jcoJXeCRqhLo8ZIMZ4jyUHoo1UdAL2enCb0AbiprCVpLDN8u9Q==", "dev": true, "license": "Apache-2.0", "peerDependencies": { @@ -5536,6 +5567,13 @@ "react": "^19.2.5" } }, + "node_modules/react-is": { + "version": "19.2.7", + "resolved": "https://registry.npmjs.org/react-is/-/react-is-19.2.7.tgz", + "integrity": "sha512-kZFnouyVv7eP/Phmrlo9FK+zcAdriZJvzxXHF1Sl1P377WSGe2G/JxVolhTrB/jeV47lKImhNUsijjHAAbcl/A==", + "dev": true, + "license": "MIT" + }, "node_modules/react-markdown": { "version": "10.1.0", "resolved": "https://registry.npmjs.org/react-markdown/-/react-markdown-10.1.0.tgz", @@ -5564,6 +5602,20 @@ "react": ">=18" } }, + "node_modules/react-test-renderer": { + "version": "19.2.5", + "resolved": "https://registry.npmjs.org/react-test-renderer/-/react-test-renderer-19.2.5.tgz", + "integrity": "sha512-kwViRpdISMTpcpy5B6TSewfJzRjnajihRaj57ZmOWKD+SPN6k9LUM13O0pfOuW8ir6B6OOiAXwCRqOoVxRNykA==", + "dev": true, + "license": "MIT", + "dependencies": { + "react-is": "^19.2.5", + "scheduler": "^0.27.0" + }, + "peerDependencies": { + "react": "^19.2.5" + } + }, "node_modules/readdirp": { "version": "3.6.0", "resolved": "https://registry.npmjs.org/readdirp/-/readdirp-3.6.0.tgz", @@ -6582,9 +6634,9 @@ "license": "ISC" }, "node_modules/ws": { - "version": "8.18.0", - "resolved": "https://registry.npmjs.org/ws/-/ws-8.18.0.tgz", - "integrity": "sha512-8VbfWfHLbbwu3+N6OKsOMpBdT4kXPDDB9cJk2bJ6mh9ucxdlnNvH1e+roYkKmN9Nxw2yjz7VzeO9oOz2zJ04Pw==", + "version": "8.21.0", + "resolved": "https://registry.npmjs.org/ws/-/ws-8.21.0.tgz", + "integrity": "sha512-Vsp28b7DRcimFQvrqu2Wek3z1iYxDCWqHYB8Qsnk/S4RfaCQzPGPyBNuVjJV3cd6UiKtUtp6sNM77gWvzcCH+g==", "license": "MIT", "engines": { "node": ">=10.0.0" diff --git a/package.json b/package.json index c74d3c4..01b8eeb 100644 --- a/package.json +++ b/package.json @@ -25,6 +25,8 @@ "typecheck": "tsc --noEmit && tsc -p example && tsc -p example/convex", "lint": "eslint .", "test": "vitest run --typecheck", + "pretest:occ:real": "npm run build", + "test:occ:real": "node scripts/run-occ-integration.mjs", "test:watch": "vitest --typecheck --clearScreen false", "test:debug": "vitest --inspect-brk --no-file-parallelism", "test:coverage": "vitest run --coverage --coverage.reporter=text", @@ -61,7 +63,7 @@ } }, "peerDependencies": { - "convex": "^1.32.0", + "convex": "^1.39.0", "react": "~18.3.1 || ^19.0.0", "react-dom": "~18.3.1 || ^19.0.0" }, @@ -74,11 +76,12 @@ "@types/node": "20.19.39", "@types/react": "19.2.14", "@types/react-dom": "19.2.3", + "@types/react-test-renderer": "19.1.0", "@vitejs/plugin-react": "6.0.1", "chokidar-cli": "3.0.0", "clsx": "2.1.1", - "convex": "1.35.1", - "convex-test": "0.0.48", + "convex": "1.42.2", + "convex-test": "0.0.54", "eslint": "9.39.4", "eslint-plugin-react-hooks": "7.0.1", "eslint-plugin-react-refresh": "0.5.2", @@ -89,6 +92,7 @@ "react": "19.2.5", "react-dom": "19.2.5", "react-markdown": "10.1.0", + "react-test-renderer": "19.2.5", "tailwind-merge": "3.5.0", "tailwindcss": "4.2.2", "typescript": "5.9.3", @@ -99,6 +103,7 @@ "types": "./dist/client/index.d.ts", "module": "./dist/client/index.js", "dependencies": { + "@convex-dev/stream": "https://pkg.pr.new/get-convex/stream/@convex-dev/stream@b153faa", "convex-helpers": "^0.1.114" } } diff --git a/scripts/run-occ-integration.mjs b/scripts/run-occ-integration.mjs new file mode 100644 index 0000000..c6dd4a8 --- /dev/null +++ b/scripts/run-occ-integration.mjs @@ -0,0 +1,59 @@ +import { randomUUID } from "node:crypto"; +import { spawnSync } from "node:child_process"; +import process from "node:process"; +import { URL } from "node:url"; + +const runId = `pts-occ:${Date.now()}:${randomUUID()}`; +const result = spawnSync( + "npx", + [ + "convex", + "run", + "occIntegration:run", + JSON.stringify({ runId }), + "--deployment", + "dev", + "--push", + "--codegen", + "disable", + "--typecheck", + "enable", + "--typecheck-components", + ], + { + cwd: new URL("..", import.meta.url), + encoding: "utf8", + env: process.env, + timeout: 15 * 60 * 1000, + }, +); + +if (result.error) { + process.stderr.write( + `Unable to complete the real-backend OCC test: ${result.error.message}\n`, + ); + if (result.stderr) process.stderr.write(result.stderr); + if (result.stdout) process.stderr.write(result.stdout); + process.exit(1); +} + +if (result.status !== 0) { + if (result.stderr) process.stderr.write(result.stderr); + if (result.stdout) process.stderr.write(result.stdout); + process.stderr.write( + "Real-backend OCC tests require a configured personal Convex dev deployment or a CONVEX_DEPLOY_KEY for one.\n", + ); + process.exit(result.status ?? 1); +} + +let summary; +try { + summary = JSON.parse(result.stdout.trim()); +} catch { + process.stderr.write("Convex returned an unreadable OCC test summary.\n"); + process.stderr.write(result.stdout); + process.exit(1); +} + +process.stdout.write(`${JSON.stringify(summary, null, 2)}\n`); +if (summary.passed !== true) process.exit(1); diff --git a/src/client/index.test.ts b/src/client/index.test.ts new file mode 100644 index 0000000..dea8138 --- /dev/null +++ b/src/client/index.test.ts @@ -0,0 +1,332 @@ +import type { StreamReadResult } from "@convex-dev/stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { + PersistentTextStreaming, + type StreamId, + type StreamWriter, +} from "./index.js"; + +const streamId = "stream" as StreamId; + +type Call = + | { kind: "claim" } + | { kind: "append"; text: string; final: boolean } + | { kind: "status"; status: string } + | { kind: "read"; cursor: string | null; numItems: number }; + +function setup(claims: boolean[] = [true]) { + const refs = { + lib: { + addChunk: {}, + claim: {}, + createStream: {}, + deleteStream: {}, + getStreamStatus: {}, + getStreamText: {}, + read: {}, + setStreamStatus: {}, + }, + }; + const calls: Call[] = []; + const terminal: StreamReadResult = { + streamId: "core" as never, + attempt: 0, + startIndex: 0, + nextIndex: 0, + page: [], + continueCursor: "done", + caughtUp: true, + status: "done", + }; + const ctx = { + runMutation: vi.fn(async (ref: object, args: Record) => { + if (ref === refs.lib.claim) { + calls.push({ kind: "claim" }); + return { claimed: claims.shift() ?? false }; + } + if (ref === refs.lib.addChunk) { + calls.push({ + kind: "append", + text: args.text as string, + final: args.final as boolean, + }); + return null; + } + if (ref === refs.lib.setStreamStatus) { + calls.push({ kind: "status", status: args.status as string }); + return null; + } + throw new Error("Unexpected mutation."); + }), + runQuery: vi.fn(async (ref: object, args: Record) => { + if (ref !== refs.lib.read) throw new Error("Unexpected query."); + calls.push({ + kind: "read", + cursor: args.cursor as string | null, + numItems: args.numItems as number, + }); + return terminal; + }), + }; + return { + calls, + ctx, + streaming: new PersistentTextStreaming(refs as never), + }; +} + +function request(): Request { + const url = new URL("https://example.com/stream"); + return new Request(url, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ streamId }), + }); +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("PersistentTextStreaming transport", () => { + it("invokes only the winning writer", async () => { + const { calls, ctx, streaming } = setup([true, false]); + const writer = vi.fn>( + async (_ctx, _request, _id, append) => { + await append("winner"); + }, + ); + + const winner = await streaming.stream( + ctx as never, + request(), + streamId, + writer, + ); + const loser = await streaming.stream( + ctx as never, + request(), + streamId, + writer, + ); + + expect(loser.status).toBe(205); + expect(await winner.text()).toBe("winner"); + expect(writer).toHaveBeenCalledTimes(1); + expect(calls.filter((call) => call.kind === "claim")).toHaveLength(2); + expect(calls).toContainEqual({ + kind: "append", + text: "winner", + final: true, + }); + }); + + it("reads one bounded page through an app-owned query", async () => { + const { calls, ctx, streaming } = setup(); + + const page = await streaming.readStream(ctx as never, streamId, { + cursor: null, + numItems: 16, + }); + + expect(page.status).toBe("done"); + expect(calls).toEqual([{ kind: "read", cursor: null, numItems: 16 }]); + }); + + it("flushes on cadence and completes durably before closing", async () => { + vi.useFakeTimers(); + const { calls, ctx, streaming } = setup(); + let release: (() => void) | undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const writer: StreamWriter = async (_ctx, _request, _id, append) => { + await append("timed"); + await gate; + }; + + const response = await streaming.stream( + ctx as never, + request(), + streamId, + writer, + ); + await vi.advanceTimersByTimeAsync(99); + expect(calls.some((call) => call.kind === "append")).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect(calls).toContainEqual({ + kind: "append", + text: "timed", + final: false, + }); + + release?.(); + expect(await response.text()).toBe("timed"); + expect(calls.at(-1)).toEqual({ kind: "status", status: "done" }); + }); + + it("splits oversized input and serializes concurrent appends", async () => { + const { calls, ctx, streaming } = setup(); + const first = "a".repeat(12 * 1024); + const second = "🙂".repeat(2 * 1024); + const writer: StreamWriter = async (_ctx, _request, _id, append) => { + await Promise.all([append(first), append(second)]); + }; + + const response = await streaming.stream( + ctx as never, + request(), + streamId, + writer, + ); + expect(await response.text()).toBe(first + second); + + const appends = calls.filter( + (call): call is Extract => + call.kind === "append", + ); + expect(appends.map((call) => call.text).join("")).toBe(first + second); + expect( + appends.every( + (call) => new TextEncoder().encode(call.text).byteLength <= 16 * 1024, + ), + ).toBe(true); + expect(appends.at(-1)?.final).toBe(true); + }); + + it("flushes pending text before recording producer failure", async () => { + const { calls, ctx, streaming } = setup(); + const failure = new Error("producer failed"); + const writer: StreamWriter = async (_ctx, _request, _id, append) => { + await append("durable prefix"); + throw failure; + }; + + const response = await streaming.stream( + ctx as never, + request(), + streamId, + writer, + ); + await expect(response.text()).rejects.toThrow("producer failed"); + expect(calls.slice(-2)).toEqual([ + { kind: "append", text: "durable prefix", final: false }, + { kind: "status", status: "error" }, + ]); + }); + + it("bounds an unread raw response without stopping durable completion", async () => { + const { calls, ctx, streaming } = setup(); + let durableDone: (() => void) | undefined; + const done = new Promise((resolve) => { + durableDone = resolve; + }); + ctx.runMutation.mockImplementation( + async (ref: object, args: Record) => { + if ( + ref === + (streaming.component as never as { lib: { claim: object } }).lib.claim + ) { + calls.push({ kind: "claim" }); + return { claimed: true }; + } + if ("text" in args) { + calls.push({ + kind: "append", + text: args.text as string, + final: args.final as boolean, + }); + return null; + } + calls.push({ kind: "status", status: args.status as string }); + if (args.status === "done") durableDone?.(); + return null; + }, + ); + const text = "🙂".repeat(20 * 1024); + const writer: StreamWriter = async (_ctx, _request, _id, append) => { + await append(text); + }; + + const response = await streaming.stream( + ctx as never, + request(), + streamId, + writer, + ); + + // Deliberately leave the raw body unread until durable persistence finishes. + await done; + const appends = calls.filter( + (call): call is Extract => + call.kind === "append", + ); + expect(appends.map((call) => call.text).join("")).toBe(text); + expect( + appends.every( + (call) => new TextEncoder().encode(call.text).byteLength <= 16 * 1024, + ), + ).toBe(true); + expect(calls.at(-1)).toEqual({ kind: "status", status: "done" }); + await expect(response.text()).rejects.toThrow( + "Raw stream consumer fell behind durable replay.", + ); + }); + + it("continues persistence after the raw consumer disconnects", async () => { + const { calls, ctx, streaming } = setup(); + let release: (() => void) | undefined; + let persisted: (() => void) | undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + const done = new Promise((resolve) => { + persisted = resolve; + }); + ctx.runMutation.mockImplementation( + async (ref: object, args: Record) => { + if ( + ref === + (streaming.component as never as { lib: { claim: object } }).lib.claim + ) { + calls.push({ kind: "claim" }); + return { claimed: true }; + } + if ("text" in args) { + calls.push({ + kind: "append", + text: args.text as string, + final: args.final as boolean, + }); + if (args.final === true) persisted?.(); + return null; + } + calls.push({ kind: "status", status: args.status as string }); + return null; + }, + ); + const writer: StreamWriter = async (_ctx, _request, _id, append) => { + await append("first."); + await gate; + await append("second"); + }; + + const response = await streaming.stream( + ctx as never, + request(), + streamId, + writer, + ); + const reader = response.body!.getReader(); + await reader.read(); + await reader.cancel(); + release?.(); + await done; + + expect(calls.filter((call) => call.kind === "append")).toEqual([ + { kind: "append", text: "first.", final: false }, + { kind: "append", text: "second", final: true }, + ]); + }); +}); diff --git a/src/client/index.ts b/src/client/index.ts index fea9d22..e17aa3f 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -6,6 +6,7 @@ import type { GenericMutationCtx, GenericQueryCtx, } from "convex/server"; +import type { StreamQueryArgs, StreamReadResult } from "@convex-dev/stream"; import { v, type GenericId } from "convex/values"; import { api } from "../component/_generated/api.js"; import type { StreamStatus } from "../component/schema.js"; @@ -25,11 +26,146 @@ export type StreamWriter> = ( chunkAppender: ChunkAppender, ) => Promise; -// TODO -- make more flexible. # of bytes, etc? -const hasDelimeter = (text: string) => { +const FLUSH_MS = 100; +const MAX_BATCH_BYTES = 16 * 1024; +const MAX_RAW_BUFFER_BYTES = 64 * 1024; +const encoder = new TextEncoder(); + +function shouldFlush(text: string): boolean { return text.includes(".") || text.includes("!") || text.includes("?"); +} + +function splitText(text: string): string[] { + const chunks: string[] = []; + let start = 0; + let index = 0; + let bytes = 0; + + for (const character of text) { + const size = encoder.encode(character).byteLength; + if (bytes > 0 && bytes + size > MAX_BATCH_BYTES) { + chunks.push(text.slice(start, index)); + start = index; + bytes = 0; + } + bytes += size; + index += character.length; + } + if (start < text.length) chunks.push(text.slice(start)); + return chunks; +} + +type Sink = { + write(text: string): void; + close(): void; + error(error: unknown): void; }; +class Batch { + private parts: string[] = []; + private bytes = 0; + private timer: ReturnType | null = null; + private queue = Promise.resolve(); + private ended = false; + + constructor( + private readonly appendChunk: ( + text: string, + final: boolean, + ) => Promise, + private readonly setStatus: (status: StreamStatus) => Promise, + private readonly sink: Sink, + ) {} + + add(text: string): Promise { + if (this.ended) + return Promise.reject(new Error("Stream is already finalized.")); + const next = this.queue.then(() => this.accept(text)); + this.queue = next; + return next; + } + + finish(): Promise { + if (this.ended) return this.queue; + this.ended = true; + this.stopTimer(); + const next = this.queue.then(async () => { + this.stopTimer(); + if (this.parts.length === 0) { + await this.setStatus("done"); + } else { + await this.flush(true); + } + }); + this.queue = next; + return next; + } + + fail(): Promise { + this.ended = true; + this.stopTimer(); + const prior = this.queue; + const next = (async () => { + try { + await prior; + this.stopTimer(); + await this.flush(false); + } finally { + await this.setStatus("error"); + } + })(); + this.queue = next; + return next; + } + + private async accept(text: string): Promise { + if (text.length === 0) return; + const chunks = splitText(text); + for (const chunk of chunks) this.sink.write(chunk); + + for (const chunk of chunks) { + const size = encoder.encode(chunk).byteLength; + if (this.bytes > 0 && this.bytes + size > MAX_BATCH_BYTES) { + await this.flush(false); + } + this.parts.push(chunk); + this.bytes += size; + if (this.bytes >= MAX_BATCH_BYTES) await this.flush(false); + } + + if (shouldFlush(text)) await this.flush(false); + else this.startTimer(); + } + + private async flush(final: boolean): Promise { + this.stopTimer(); + if (this.parts.length === 0) { + if (final) await this.setStatus("done"); + return; + } + const text = this.parts.join(""); + this.parts = []; + this.bytes = 0; + await this.appendChunk(text, final); + } + + private startTimer(): void { + if (this.parts.length === 0 || this.timer !== null) return; + this.timer = setTimeout(() => { + this.timer = null; + const next = this.queue.then(() => this.flush(false)); + this.queue = next; + void next.catch(() => undefined); + }, FLUSH_MS); + } + + private stopTimer(): void { + if (this.timer === null) return; + clearTimeout(this.timer); + this.timer = null; + } +} + // TODO -- some sort of wrapper with easy ergonomics for working with LLMs? export class PersistentTextStreaming { constructor( @@ -110,67 +246,139 @@ export class PersistentTextStreaming { request: Request, streamId: StreamId, streamWriter: StreamWriter, - ) { - const streamState = await ctx.runQuery(this.component.lib.getStreamStatus, { + ): Promise { + const { claimed } = await ctx.runMutation(this.component.lib.claim, { streamId, }); - if (streamState !== "pending") { - console.log("Stream was already started"); - return new Response("", { - status: 205, - }); - } - // Create a TransformStream to handle streaming data - const { readable, writable } = new TransformStream(); - let writer = - writable.getWriter() as WritableStreamDefaultWriter | null; - const textEncoder = new TextEncoder(); - let pending = ""; - - const doStream = async () => { - const chunkAppender: ChunkAppender = async (text) => { - // write to this handler's response stream on every update - if (writer) { + if (!claimed) return new Response(null, { status: 205 }); + + let connected = true; + const readable = new ReadableStream( + { + start: async (controller) => { + const sink: Sink = { + write(text) { + if (!connected) return; + const chunk = encoder.encode(text); + const capacity = controller.desiredSize; + if (capacity === null || capacity < chunk.byteLength) { + connected = false; + try { + controller.error( + new Error( + "Raw stream consumer fell behind durable replay.", + ), + ); + } catch { + // The response consumer has already disconnected. + } + return; + } + try { + controller.enqueue(chunk); + if ((controller.desiredSize ?? 0) <= 0) { + connected = false; + controller.error( + new Error( + "Raw stream consumer fell behind durable replay.", + ), + ); + } + } catch { + connected = false; + } + }, + close() { + if (!connected) return; + connected = false; + try { + controller.close(); + } catch { + // The response consumer has already disconnected. + } + }, + error(error) { + if (!connected) return; + connected = false; + try { + controller.error(error); + } catch { + // The response consumer has already disconnected. + } + }, + }; + const batch = new Batch( + (text, final) => this.addChunk(ctx, streamId, text, final), + (status) => this.setStreamStatus(ctx, streamId, status), + sink, + ); + try { - await writer.write(textEncoder.encode(text)); - } catch (e) { - console.error("Error writing to stream", e); - console.error( - "Will skip writing to stream but continue database updates", + await streamWriter(ctx, request, streamId, (text) => + batch.add(text), ); - writer = null; + await batch.finish(); + sink.close(); + } catch (error) { + let failure = error; + try { + await batch.fail(); + } catch (persistenceError) { + failure = persistenceError; + } + sink.error(failure); } - } - pending += text; - // write to the database periodically, like at the end of sentences - if (hasDelimeter(text)) { - await this.addChunk(ctx, streamId, pending, false); - pending = ""; - } - }; - try { - await streamWriter(ctx, request, streamId, chunkAppender); - } catch (e) { - await this.setStreamStatus(ctx, streamId, "error"); - if (writer) { - await writer.close(); - } - throw e; - } - - // Success? Flush any last updates - await this.addChunk(ctx, streamId, pending, true); - - if (writer) { - await writer.close(); - } - }; + }, + cancel() { + connected = false; + }, + }, + { + highWaterMark: MAX_RAW_BUFFER_BYTES, + size: (chunk) => chunk.byteLength, + }, + ); - // Kick off the streaming, but don't await it. - void doStream(); + return new Response(readable, { + headers: { + "Cache-Control": "no-cache, no-transform", + "Content-Type": "text/plain; charset=utf-8", + "X-Accel-Buffering": "no", + }, + }); + } - // Send the readable back to the browser - return new Response(readable); + /** + * Read one bounded page of a stream's ordered text events. + * + * Expose this from an app-owned query so followers can subscribe over the + * normal Convex websocket instead of re-reading the whole body on every + * append. Authorize the caller in that query exactly as you would for + * `getStreamBody`; this method performs no access control of its own. + * + * @param ctx - A convex context capable of running queries. + * @param streamId - The ID of the stream to read. + * @param streamArgs - The cursor and page size supplied by `useStream`. + * @returns One forward-only page plus the stream's lifecycle. + * @example + * ```ts + * export const readStream = query({ + * args: { streamId: StreamIdValidator, streamArgs: streamQueryArgsValidator }, + * handler: (ctx, { streamId, streamArgs }) => + * streaming.readStream(ctx, streamId as StreamId, streamArgs), + * }); + * ``` + */ + async readStream( + ctx: QueryCtx | MutationCtx | ActionCtx, + streamId: StreamId, + streamArgs: StreamQueryArgs, + ): Promise> { + return (await ctx.runQuery(this.component.lib.read, { + streamId, + cursor: streamArgs.cursor, + numItems: streamArgs.numItems, + })) as StreamReadResult; } /** diff --git a/src/component/_generated/component.ts b/src/component/_generated/component.ts index 97320a2..b6ef437 100644 --- a/src/component/_generated/component.ts +++ b/src/component/_generated/component.ts @@ -9,6 +9,7 @@ */ import type { FunctionReference } from "convex/server"; +import type { StreamReadResult } from "@convex-dev/stream"; /** * A utility for referencing a Convex component's exposed API. @@ -32,6 +33,13 @@ export type ComponentApi = Name >; createStream: FunctionReference<"mutation", "internal", {}, any, Name>; + claim: FunctionReference< + "mutation", + "internal", + { streamId: string }, + { claimed: boolean }, + Name + >; deleteStream: FunctionReference< "mutation", "internal", @@ -56,6 +64,13 @@ export type ComponentApi = }, Name >; + read: FunctionReference< + "query", + "internal", + { cursor: string | null; numItems: number; streamId: string }, + StreamReadResult, + Name + >; setStreamStatus: FunctionReference< "mutation", "internal", diff --git a/src/component/lib.test.ts b/src/component/lib.test.ts index e56c55f..e929196 100644 --- a/src/component/lib.test.ts +++ b/src/component/lib.test.ts @@ -1,9 +1,602 @@ /// -import { describe, it } from "vitest"; +import { convexTest } from "convex-test"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { StreamReadResult } from "@convex-dev/stream"; -describe("persistent-text-streaming", () => { - it("should be implemented", () => { - // TODO: Add tests for persistent text streaming functionality +import { api, internal } from "./_generated/api.js"; +import schema from "./schema.js"; + +const modules = import.meta.glob("./**/*.ts"); + +afterEach(() => vi.useRealTimers()); + +describe("persistent text streaming engines", () => { + it("claims one producer atomically while passive reads remain side-effect free", async () => { + const t = convexTest(schema, modules); + const streamId = await t.mutation(api.lib.createStream, {}); + const coreId = await t.run(async (ctx) => { + const stream = await ctx.db.get("streams", streamId); + if (stream?.coreId === undefined) throw new Error("Missing Stream core."); + return stream.coreId; + }); + + const passive = await t.query(api.lib.read, { + streamId, + cursor: null, + numItems: 16, + }); + expect(passive).toMatchObject({ + streamId: coreId, + status: "pending", + page: [], + caughtUp: true, + }); + await expect(t.query(api.lib.getStreamStatus, { streamId })).resolves.toBe( + "pending", + ); + + const claims = await Promise.all([ + t.mutation(api.lib.claim, { streamId }), + t.mutation(api.lib.claim, { streamId }), + ]); + expect(claims).toEqual( + expect.arrayContaining([{ claimed: true }, { claimed: false }]), + ); + expect(claims.filter(({ claimed }) => claimed)).toHaveLength(1); + const claimedFacade = await t.run((ctx) => ctx.db.get("streams", streamId)); + expect(claimedFacade).toMatchObject({ status: "pending" }); + expect(claimedFacade?.claimedAt).toEqual(expect.any(Number)); + + const claimed = await t.query(api.lib.read, { + streamId, + cursor: passive.continueCursor, + numItems: 16, + }); + expect(claimed).toMatchObject({ status: "pending", page: [] }); + + await t.mutation(api.lib.addChunk, { + streamId, + text: "durable", + final: false, + }); + const appended = await t.query(api.lib.read, { + streamId, + cursor: claimed.continueCursor, + numItems: 16, + }); + expect(appended).toMatchObject({ + status: "streaming", + page: [{ attempt: 0, seq: 0, event: "durable" }], + }); + }); + + it("claims legacy streams with the same one-winner semantics", async () => { + const t = convexTest(schema, modules); + const streamId = await t.run((ctx) => + ctx.db.insert("streams", { status: "pending" }), + ); + + await expect(t.mutation(api.lib.claim, { streamId })).resolves.toEqual({ + claimed: true, + }); + await expect(t.mutation(api.lib.claim, { streamId })).resolves.toEqual({ + claimed: false, + }); + await expect(t.query(api.lib.getStreamStatus, { streamId })).resolves.toBe( + "pending", + ); + const claimed = await t.run((ctx) => ctx.db.get("streams", streamId)); + expect(claimed?.claimedAt).toEqual(expect.any(Number)); + }); + + it("creates a stable public handle backed by an ordered Stream core", async () => { + const t = convexTest(schema, modules); + const streamId = await t.mutation(api.lib.createStream, {}); + + const created = await t.run((ctx) => ctx.db.get("streams", streamId)); + expect(created).toMatchObject({ status: "pending" }); + expect(created?.coreId).toBeDefined(); + + await t.mutation(api.lib.addChunk, { + streamId, + text: "Hello ", + final: false, + }); + await t.mutation(api.lib.addChunk, { + streamId, + text: "world!", + final: true, + }); + + await expect(t.query(api.lib.getStreamText, { streamId })).resolves.toEqual( + { + text: "Hello world!", + status: "done", + }, + ); + + const rows = await t.run(async (ctx) => ({ + facade: await ctx.db.get("streams", streamId), + events: await ctx.db.query("textStreamsEvents").collect(), + legacyChunks: await ctx.db.query("chunks").collect(), + })); + expect(rows.facade).toMatchObject({ status: "done" }); + expect(rows.events.map(({ seq, event }) => ({ seq, event }))).toEqual([ + { seq: 0, event: "Hello " }, + { seq: 1, event: "world!" }, + ]); + expect(rows.legacyChunks).toEqual([]); + }); + + it("reads current streams through bounded canonical Stream pages", async () => { + const t = convexTest(schema, modules); + const streamId = await t.mutation(api.lib.createStream, {}); + const coreId = await t.run(async (ctx) => { + const stream = await ctx.db.get("streams", streamId); + if (stream?.coreId === undefined) throw new Error("Missing Stream core."); + return stream.coreId; + }); + await t.mutation(api.lib.claim, { streamId }); + for (const [text, final] of [ + ["one", false], + ["two", false], + ["three", true], + ] as const) { + await t.mutation(api.lib.addChunk, { streamId, text, final }); + } + + const first = await t.query(api.lib.read, { + streamId, + cursor: null, + numItems: 2, + }); + expect(first).toEqual({ + streamId: coreId, + attempt: 0, + startIndex: 0, + nextIndex: 2, + page: [ + { attempt: 0, seq: 0, event: "one" }, + { attempt: 0, seq: 1, event: "two" }, + ], + continueCursor: expect.stringMatching(/^s1:/), + caughtUp: false, + status: "done", + }); + + const second = await t.query(api.lib.read, { + streamId, + cursor: first.continueCursor, + numItems: 2, + }); + expect(second).toEqual({ + streamId: coreId, + attempt: 0, + startIndex: 2, + nextIndex: 3, + page: [{ attempt: 0, seq: 2, event: "three" }], + continueCursor: expect.stringMatching(/^s1:/), + caughtUp: true, + status: "done", + }); + + const otherId = await t.mutation(api.lib.createStream, {}); + await expect( + t.query(api.lib.read, { + streamId: otherId, + cursor: first.continueCursor, + numItems: 2, + }), + ).rejects.toThrow(/different stream/); + await expect( + t.query(api.lib.read, { + streamId, + cursor: "not-a-cursor", + numItems: 2, + }), + ).rejects.toThrow(/Malformed stream cursor/); + }); + + it("keeps existing rows readable and writable through the legacy chunks path", async () => { + const t = convexTest(schema, modules); + const streamId = await t.run(async (ctx) => { + const id = await ctx.db.insert("streams", { status: "pending" }); + await ctx.db.insert("chunks", { streamId: id, text: "Existing " }); + return id; + }); + + await t.mutation(api.lib.addChunk, { + streamId, + text: "response", + final: true, + }); + + await expect(t.query(api.lib.getStreamText, { streamId })).resolves.toEqual( + { + text: "Existing response", + status: "done", + }, + ); + const cores = await t.run((ctx) => ctx.db.query("textStreams").collect()); + expect(cores).toEqual([]); + }); + + it("paginates legacy chunks canonically and includes appends between pages", async () => { + const t = convexTest(schema, modules); + const streamId = await t.run(async (ctx) => { + const id = await ctx.db.insert("streams", { status: "streaming" }); + for (const text of ["one", "two", "three"]) { + await ctx.db.insert("chunks", { streamId: id, text }); + } + return id; + }); + + const first = await t.query(api.lib.read, { + streamId, + cursor: null, + numItems: 128, + }); + expect(first).toEqual({ + streamId, + attempt: 0, + startIndex: 0, + nextIndex: 1, + page: [{ attempt: 0, seq: 0, event: "one" }], + continueCursor: expect.stringMatching(/^p1:/), + caughtUp: false, + status: "streaming", + }); + + await t.mutation(api.lib.addChunk, { + streamId, + text: "four", + final: true, + }); + const second = await t.query(api.lib.read, { + streamId, + cursor: first.continueCursor, + numItems: 128, + }); + expect(second).toEqual({ + streamId, + attempt: 0, + startIndex: 1, + nextIndex: 2, + page: [{ attempt: 0, seq: 1, event: "two" }], + continueCursor: expect.stringMatching(/^p1:/), + caughtUp: false, + status: "done", + }); + + const third = await t.query(api.lib.read, { + streamId, + cursor: second.continueCursor, + numItems: 128, + }); + expect(third.page).toEqual([{ attempt: 0, seq: 2, event: "three" }]); + expect(third.caughtUp).toBe(false); + + const fourth = await t.query(api.lib.read, { + streamId, + cursor: third.continueCursor, + numItems: 128, + }); + expect(fourth.page).toEqual([{ attempt: 0, seq: 3, event: "four" }]); + expect(fourth.caughtUp).toBe(false); + + const tail = await t.query(api.lib.read, { + streamId, + cursor: fourth.continueCursor, + numItems: 128, + }); + expect(tail).toMatchObject({ + startIndex: 4, + nextIndex: 4, + page: [], + caughtUp: true, + status: "done", + }); + + const otherId = await t.run((ctx) => + ctx.db.insert("streams", { status: "pending" }), + ); + await expect( + t.query(api.lib.read, { + streamId: otherId, + cursor: first.continueCursor, + numItems: 2, + }), + ).rejects.toThrow(/Malformed stream cursor/); + await expect( + t.query(api.lib.read, { + streamId, + cursor: "p1:%7Bbad", + numItems: 2, + }), + ).rejects.toThrow(/Malformed stream cursor/); + }); + + it("resumes a caught-up legacy streaming cursor after a later append", async () => { + const t = convexTest(schema, modules); + const streamId = await t.run(async (ctx) => { + const id = await ctx.db.insert("streams", { status: "streaming" }); + await ctx.db.insert("chunks", { streamId: id, text: "before" }); + return id; + }); + + const first = await t.query(api.lib.read, { + streamId, + cursor: null, + numItems: 1024, + }); + expect(first.page).toEqual([{ attempt: 0, seq: 0, event: "before" }]); + expect(first.caughtUp).toBe(false); + + const tail = await t.query(api.lib.read, { + streamId, + cursor: first.continueCursor, + numItems: 1024, + }); + expect(tail).toMatchObject({ + startIndex: 1, + nextIndex: 1, + page: [], + caughtUp: true, + status: "streaming", + }); + + await t.mutation(api.lib.addChunk, { + streamId, + text: "after", + final: false, + }); + const resumed = await t.query(api.lib.read, { + streamId, + cursor: tail.continueCursor, + numItems: 1024, + }); + expect(resumed).toMatchObject({ + startIndex: 1, + nextIndex: 2, + page: [{ attempt: 0, seq: 1, event: "after" }], + caughtUp: false, + status: "streaming", + }); + }); + + it("virtually splits JSON-expanding legacy chunks below the SSE frame limit", async () => { + const t = convexTest(schema, modules); + const text = `${"\u0000".repeat(400_000)}🙂tail`; + expect(new TextEncoder().encode(text).byteLength).toBeLessThan(1024 * 1024); + expect( + new TextEncoder().encode(JSON.stringify(text)).byteLength, + ).toBeGreaterThan(2 * 1024 * 1024); + + const streamId = await t.run(async (ctx) => { + const id = await ctx.db.insert("streams", { status: "done" }); + await ctx.db.insert("chunks", { streamId: id, text }); + return id; + }); + + const pieces: string[] = []; + let cursor: string | null = null; + let caughtUp = false; + for (let reads = 0; reads < 32 && !caughtUp; reads += 1) { + const result: StreamReadResult = await t.query( + api.lib.read, + { + streamId, + cursor, + numItems: 1024, + }, + ); + expect(result.page.length).toBeLessThanOrEqual(1); + expect( + new TextEncoder().encode(JSON.stringify(result)).byteLength, + ).toBeLessThan(2 * 1024 * 1024); + if (result.page[0]) { + expect( + new TextEncoder().encode(JSON.stringify(result.page[0])).byteLength, + ).toBeLessThan(300 * 1024); + pieces.push(result.page[0].event); + } + cursor = result.continueCursor; + caughtUp = result.caughtUp; + } + + expect(caughtUp).toBe(true); + expect(pieces.length).toBeGreaterThan(1); + expect(pieces.join("")).toBe(text); + }); + + it("maps legacy terminal statuses into canonical lifecycle results", async () => { + const t = convexTest(schema, modules); + for (const status of ["error", "timeout"] as const) { + const streamId = await t.run((ctx) => + ctx.db.insert("streams", { status }), + ); + const result = await t.query(api.lib.read, { + streamId, + cursor: null, + numItems: 1, + }); + expect(result).toMatchObject({ + streamId, + status: "failed", + error: { code: status }, + caughtUp: true, + }); + } + }); + + it("rejects invalid read bounds on a live stream", async () => { + const t = convexTest(schema, modules); + const streamId = await t.mutation(api.lib.createStream, {}); + for (const numItems of [0, 1.5, 1025]) { + await expect( + t.query(api.lib.read, { + streamId, + cursor: null, + numItems, + }), + ).rejects.toThrow(/numItems must be an integer/); + } + }); + + it("rejects missing or deleting streams", async () => { + const t = convexTest(schema, modules); + const deletedId = await t.mutation(api.lib.createStream, {}); + await t.mutation(api.lib.deleteStream, { streamId: deletedId }); + await expect( + t.query(api.lib.read, { + streamId: deletedId, + cursor: null, + numItems: 1, + }), + ).rejects.toThrow(/Stream not found/); + await expect( + t.mutation(api.lib.claim, { streamId: deletedId }), + ).rejects.toThrow(/Stream not found/); + + const deletingId = await t.mutation(api.lib.createStream, {}); + await t.run(async (ctx) => { + const facade = await ctx.db.get("streams", deletingId); + if (facade?.coreId === undefined) throw new Error("Missing Stream core."); + const core = await ctx.db.get("textStreams", facade.coreId); + if (core === null) throw new Error("Missing Stream core."); + await ctx.db.replace("textStreams", core._id, { + attempt: core.attempt, + nextSeq: core.nextSeq, + restarts: core.restarts, + status: "deleting", + deletingAt: Date.now(), + }); + }); + await expect( + t.query(api.lib.read, { + streamId: deletingId, + cursor: null, + numItems: 1, + }), + ).rejects.toThrow(); + await expect( + t.mutation(api.lib.claim, { streamId: deletingId }), + ).rejects.toThrow(/Stream not found/); + }); + + it("maps producer failures and timeouts onto the Stream lifecycle", async () => { + vi.useFakeTimers(); + vi.setSystemTime(0); + const t = convexTest(schema, modules); + + const failedId = await t.mutation(api.lib.createStream, {}); + await t.mutation(api.lib.setStreamStatus, { + streamId: failedId, + status: "error", + }); + await expect( + t.query(api.lib.getStreamText, { streamId: failedId }), + ).resolves.toEqual({ + text: "", + status: "error", + }); + await expect( + t.query(api.lib.read, { + streamId: failedId, + cursor: null, + numItems: 8, + }), + ).resolves.toMatchObject({ + status: "failed", + error: { code: "error" }, + caughtUp: true, + }); + await expect( + t.mutation(api.lib.claim, { streamId: failedId }), + ).resolves.toEqual({ claimed: false }); + + const timedOutId = await t.mutation(api.lib.createStream, {}); + vi.setSystemTime(20 * 60 * 1000 + 1); + await t.mutation(internal.lib.cleanupExpiredStreams, {}); + await expect( + t.query(api.lib.getStreamText, { streamId: timedOutId }), + ).resolves.toEqual({ + text: "", + status: "timeout", + }); + await expect( + t.query(api.lib.read, { + streamId: timedOutId, + cursor: null, + numItems: 8, + }), + ).resolves.toMatchObject({ + status: "failed", + error: { code: "timeout" }, + caughtUp: true, + }); + + const errors = await t.run(async (ctx) => { + const failed = await ctx.db.get("streams", failedId); + const timedOut = await ctx.db.get("streams", timedOutId); + return Promise.all([ + failed?.coreId ? ctx.db.get("textStreams", failed.coreId) : null, + timedOut?.coreId ? ctx.db.get("textStreams", timedOut.coreId) : null, + ]); + }); + expect(errors[0]).toMatchObject({ + status: "failed", + error: { code: "error" }, + }); + expect(errors[1]).toMatchObject({ + status: "failed", + error: { code: "timeout" }, + }); + }); + + it("deletes current and legacy storage without changing the public API", async () => { + vi.useFakeTimers(); + const t = convexTest(schema, modules); + + const currentId = await t.mutation(api.lib.createStream, {}); + await t.run(async (ctx) => { + const facade = await ctx.db.get("streams", currentId); + if (facade?.coreId === undefined) throw new Error("Missing Stream core."); + for (let seq = 0; seq < 513; seq += 1) { + await ctx.db.insert("textStreamsEvents", { + streamId: facade.coreId, + attempt: 0, + seq, + event: `current-${seq}`, + }); + } + await ctx.db.patch("textStreams", facade.coreId, { + status: "streaming", + nextSeq: 513, + }); + }); + await t.mutation(api.lib.deleteStream, { streamId: currentId }); + + const legacyId = await t.run(async (ctx) => { + const id = await ctx.db.insert("streams", { status: "done" }); + await ctx.db.insert("chunks", { streamId: id, text: "legacy" }); + return id; + }); + await t.mutation(api.lib.deleteStream, { streamId: legacyId }); + await t.finishAllScheduledFunctions(vi.runAllTimers); + + const remaining = await t.run(async (ctx) => ({ + facades: await ctx.db.query("streams").collect(), + chunks: await ctx.db.query("chunks").collect(), + cores: await ctx.db.query("textStreams").collect(), + events: await ctx.db.query("textStreamsEvents").collect(), + })); + expect(remaining).toEqual({ + facades: [], + chunks: [], + cores: [], + events: [], + }); }); }); diff --git a/src/component/lib.ts b/src/component/lib.ts index 89931d0..94a385d 100644 --- a/src/component/lib.ts +++ b/src/component/lib.ts @@ -1,38 +1,422 @@ +import { + streamReadResultValidator, + type StreamReadResult, +} from "@convex-dev/stream"; import { paginator } from "convex-helpers/server/pagination"; -import { v } from "convex/values"; +import { + ConvexError, + convexToJson, + jsonToConvex, + v, + type GenericId, +} from "convex/values"; + import { internal } from "./_generated/api.js"; -import { internalMutation, mutation, query } from "./_generated/server.js"; -import schema, { streamStatusValidator } from "./schema.js"; +import type { Doc, Id } from "./_generated/dataModel.js"; +import { + internalMutation, + mutation, + query, + type MutationCtx, + type QueryCtx, +} from "./_generated/server.js"; +import schema, { streamStatusValidator, type StreamStatus } from "./schema.js"; +import { textStreams } from "./streams.js"; + +const EXPIRATION_TIME = 20 * 60 * 1000; +const BATCH_SIZE = 100; +const DELETE_BATCH_SIZE = 64; +const MAX_READ_ITEMS = 1024; +const MAX_READ_BYTES = 8 * 1024 * 1024; +const MAX_LEGACY_SEGMENT_JSON_BYTES = 256 * 1024; +const LEGACY_CURSOR_PREFIX = "p1:"; +const encoder = new TextEncoder(); + +type ReadCtx = QueryCtx | MutationCtx; + +type TextReadResult = StreamReadResult; +type LegacyCursor = { + streamId: Id<"streams">; + nextIndex: number; + lastIndexKey: string | null; + offset: number; +}; +type TextReadBase = { + streamId: GenericId; + attempt: number; + startIndex: number; + nextIndex: number; + page: Array<{ attempt: number; seq: number; event: string }>; + continueCursor: string; + caughtUp: boolean; +}; + +function fail(code: string, message: string): never { + throw new ConvexError({ code, message }); +} + +function readCount(value: number): number { + if (!Number.isSafeInteger(value) || value < 1 || value > MAX_READ_ITEMS) { + fail( + "limitExceeded", + `numItems must be an integer from 1 through ${MAX_READ_ITEMS}.`, + ); + } + return value; +} + +function encodeLegacyCursor(cursor: LegacyCursor): string { + return `${LEGACY_CURSOR_PREFIX}${encodeURIComponent( + JSON.stringify(convexToJson(cursor)), + )}`; +} + +function validLegacyIndexKey(cursor: string, streamId: Id<"streams">): boolean { + try { + const key = jsonToConvex(JSON.parse(cursor)); + return ( + Array.isArray(key) && + key.length === 3 && + key[0] === streamId && + typeof key[1] === "number" && + Number.isFinite(key[1]) && + typeof key[2] === "string" + ); + } catch { + return false; + } +} + +function decodeLegacyCursor( + cursor: string | null, + streamId: Id<"streams">, +): LegacyCursor { + if (cursor === null) { + return { streamId, nextIndex: 0, lastIndexKey: null, offset: 0 }; + } + if (!cursor.startsWith(LEGACY_CURSOR_PREFIX)) { + fail("invalidCursor", "Malformed stream cursor."); + } + + try { + const decoded = jsonToConvex( + JSON.parse(decodeURIComponent(cursor.slice(LEGACY_CURSOR_PREFIX.length))), + ); + if ( + typeof decoded !== "object" || + decoded === null || + Array.isArray(decoded) || + Object.keys(decoded).length !== 4 || + !("streamId" in decoded) || + !("nextIndex" in decoded) || + !("lastIndexKey" in decoded) || + !("offset" in decoded) || + decoded.streamId !== streamId || + !Number.isSafeInteger(decoded.nextIndex) || + (decoded.nextIndex as number) < 0 || + !Number.isSafeInteger(decoded.offset) || + (decoded.offset as number) < 0 || + (decoded.lastIndexKey !== null && + (typeof decoded.lastIndexKey !== "string" || + !validLegacyIndexKey(decoded.lastIndexKey, streamId))) + ) { + fail("invalidCursor", "Malformed stream cursor."); + } + return decoded as LegacyCursor; + } catch (error) { + if (error instanceof ConvexError) throw error; + fail("invalidCursor", "Malformed stream cursor."); + } +} + +function legacyIndexKey(streamId: Id<"streams">, chunk: Doc<"chunks">): string { + return JSON.stringify( + convexToJson([streamId, chunk._creationTime, chunk._id]), + ); +} + +function isCodePointBoundary(text: string, offset: number): boolean { + if (offset <= 0 || offset >= text.length) return true; + const previous = text.charCodeAt(offset - 1); + const current = text.charCodeAt(offset); + return !( + previous >= 0xd800 && + previous <= 0xdbff && + current >= 0xdc00 && + current <= 0xdfff + ); +} + +function floorCodePointBoundary(text: string, offset: number): number { + return isCodePointBoundary(text, offset) ? offset : offset - 1; +} + +function nextCodePointBoundary(text: string, offset: number): number { + const first = text.charCodeAt(offset); + return first >= 0xd800 && + first <= 0xdbff && + offset + 1 < text.length && + text.charCodeAt(offset + 1) >= 0xdc00 && + text.charCodeAt(offset + 1) <= 0xdfff + ? offset + 2 + : offset + 1; +} + +function serializedStringBytes(value: string): number { + return encoder.encode(JSON.stringify(value)).byteLength; +} + +function legacySegment( + text: string, + offset: number, +): { event: string; nextOffset: number } { + if (offset > text.length || !isCodePointBoundary(text, offset)) { + fail("invalidCursor", "Malformed stream cursor."); + } + if (offset === text.length) return { event: "", nextOffset: offset }; + + let low = offset; + let high = text.length; + let best = offset; + while (low <= high) { + const middle = Math.floor((low + high) / 2); + const end = floorCodePointBoundary(text, middle); + if (end <= offset) { + low = middle + 1; + continue; + } + if ( + serializedStringBytes(text.slice(offset, end)) <= + MAX_LEGACY_SEGMENT_JSON_BYTES + ) { + best = end; + low = middle + 1; + } else { + high = middle - 1; + } + } + + if (best === offset) best = nextCodePointBoundary(text, offset); + return { event: text.slice(offset, best), nextOffset: best }; +} + +function withLifecycle( + base: TextReadBase, + stream: Doc<"streams">, +): TextReadResult { + if (stream.status === "error" || stream.status === "timeout") { + return { + ...base, + status: "failed", + error: { + code: stream.status, + message: + stream.status === "timeout" + ? "Stream generation timed out." + : "Stream generation failed.", + }, + }; + } + return { ...base, status: stream.status }; +} + +async function readLegacy( + ctx: QueryCtx, + stream: Doc<"streams">, + cursor: string | null, +): Promise { + const decoded = decodeLegacyCursor(cursor, stream._id); + const result = await paginator(ctx.db, schema) + .query("chunks") + .withIndex("byStream", (q) => q.eq("streamId", stream._id)) + .paginate({ + cursor: decoded.lastIndexKey, + // Legacy chunks do not carry Stream-native sequence cursors. Returning + // one conservatively sized virtual event keeps every SSE frame + // indivisible, so serveHttpStream never synthesizes an incompatible s1 + // cursor for this adapter. + numItems: 1, + maximumRowsRead: 1, + maximumBytesRead: MAX_READ_BYTES, + }); + + const chunk = result.page.at(0); + if (chunk === undefined && decoded.offset !== 0) { + fail("invalidCursor", "Malformed stream cursor."); + } + const segment = chunk ? legacySegment(chunk.text, decoded.offset) : null; + const fullyConsumed = + chunk !== undefined && segment?.nextOffset === chunk.text.length; + const nextIndex = decoded.nextIndex + (segment === null ? 0 : 1); + const lastIndexKey = fullyConsumed + ? legacyIndexKey(stream._id, chunk) + : decoded.lastIndexKey; + const offset = fullyConsumed ? 0 : (segment?.nextOffset ?? decoded.offset); + + return withLifecycle( + { + streamId: stream._id, + attempt: 0, + startIndex: decoded.nextIndex, + nextIndex, + page: segment + ? [{ attempt: 0, seq: decoded.nextIndex, event: segment.event }] + : [], + continueCursor: encodeLegacyCursor({ + streamId: stream._id, + nextIndex, + lastIndexKey, + offset, + }), + caughtUp: + segment === null ? result.isDone : fullyConsumed && result.isDone, + }, + stream, + ); +} + +async function requireStream( + ctx: ReadCtx, + streamId: Id<"streams">, +): Promise> { + const stream = await ctx.db.get("streams", streamId); + if (stream === null) fail("streamNotFound", "Stream not found."); + return stream; +} + +function publicStatus( + core: NonNullable>>, +): StreamStatus { + if (core.status === "failed") { + return core.error.code === "timeout" ? "timeout" : "error"; + } + if (core.status === "canceled" || core.status === "deleting") return "error"; + return core.status; +} -// Create a new stream with zero chunks. +async function legacyText( + ctx: QueryCtx, + streamId: Id<"streams">, +): Promise { + const chunks = await ctx.db + .query("chunks") + .withIndex("byStream", (q) => q.eq("streamId", streamId)) + .collect(); + return chunks.map((chunk) => chunk.text).join(""); +} + +async function currentText(ctx: QueryCtx, coreId: Id<"textStreams">) { + const core = await textStreams.get(ctx, { streamId: coreId }); + if (core === null || core.status === "deleting") { + fail("streamNotFound", "Stream not found."); + } + + const events = await ctx.db + .query("textStreamsEvents") + .withIndex("by_streamId_and_attempt_and_seq", (q) => + q.eq("streamId", coreId).eq("attempt", core.attempt), + ) + .collect(); + + return { + text: events.map((event) => event.event).join(""), + status: publicStatus(core), + }; +} + +// Create the Stream-backed core and stable public handle atomically. Existing +// applications continue storing an Id<"streams"> regardless of the engine. export const createStream = mutation({ args: {}, + returns: v.id("streams"), handler: async (ctx) => { - const streamId = await ctx.db.insert("streams", { + const core = await textStreams.create(ctx); + return ctx.db.insert("streams", { status: "pending", + coreId: core.streamId, }); - return streamId; }, }); -// Add a chunk to a stream. -// If final is true, set the stream to done. -// Can only be done on streams which are pending or streaming. +// Atomically elect exactly one producer without changing the stable stream ID. +// Followers only call `read`; they never claim production as a side effect. +export const claim = mutation({ + args: { streamId: v.id("streams") }, + returns: v.object({ claimed: v.boolean() }), + handler: async (ctx, args) => { + const stream = await requireStream(ctx, args.streamId); + if (stream.status !== "pending" || stream.claimedAt !== undefined) { + return { claimed: false }; + } + + if (stream.coreId !== undefined) { + const core = await textStreams.get(ctx, { streamId: stream.coreId }); + if (core === null || core.status === "deleting") { + fail("streamNotFound", "Stream not found."); + } + if (core.status !== "pending") return { claimed: false }; + } + + await ctx.db.patch("streams", args.streamId, { claimedAt: Date.now() }); + return { claimed: true }; + }, +}); + +// Canonical bounded follower read. Current rows delegate ordering and cursor +// validation to Stream; legacy chunks use a facade-ID-bound compatibility +// cursor and expose the same append-only envelope. +export const read = query({ + args: { + streamId: v.id("streams"), + cursor: v.union(v.string(), v.null()), + numItems: v.number(), + }, + returns: streamReadResultValidator(v.string()), + handler: async (ctx, args): Promise => { + const stream = await requireStream(ctx, args.streamId); + const numItems = readCount(args.numItems); + + if (stream.coreId === undefined) { + return readLegacy(ctx, stream, args.cursor); + } + + // Stream is the lifecycle authority for current rows. The facade status is + // only a compatibility/discovery mirror; notably, claim may mark it + // streaming before the first durable append moves the core from pending. + return textStreams.read(ctx, { + streamId: stream.coreId, + cursor: args.cursor, + numItems, + }); + }, +}); + +// Add a persisted text chunk. Rows created before the Stream-backed engine keep +// their original behavior; every newly created row appends through Stream. export const addChunk = mutation({ args: { streamId: v.id("streams"), text: v.string(), final: v.boolean(), }, + returns: v.null(), handler: async (ctx, args) => { - const stream = await ctx.db.get("streams", args.streamId); - if (!stream) { - throw new Error("Stream not found"); + const stream = await requireStream(ctx, args.streamId); + + if (stream.coreId !== undefined) { + const result = await textStreams.append(ctx, { + streamId: stream.coreId, + event: args.text, + ...(args.final ? { complete: true as const } : {}), + }); + if (stream.status !== result.status) { + await ctx.db.patch("streams", args.streamId, { status: result.status }); + } + return null; } + if (stream.status === "pending") { - await ctx.db.patch("streams", args.streamId, { - status: "streaming", - }); + await ctx.db.patch("streams", args.streamId, { status: "streaming" }); } else if (stream.status !== "streaming") { throw new Error("Stream is not streaming; did it timeout?"); } @@ -41,49 +425,61 @@ export const addChunk = mutation({ text: args.text, }); if (args.final) { - await ctx.db.patch("streams", args.streamId, { - status: "done", - }); + await ctx.db.patch("streams", args.streamId, { status: "done" }); } + return null; }, }); -// Set the status of a stream. -// Can only be done on streams which are pending or streaming. +// Set a terminal stream status. The client wrapper uses this to record producer +// errors, while the timeout job uses the same path for Stream-backed rows. export const setStreamStatus = mutation({ args: { streamId: v.id("streams"), - status: v.union( - v.literal("pending"), - v.literal("streaming"), - v.literal("done"), - v.literal("error"), - v.literal("timeout"), - ), + status: streamStatusValidator, }, + returns: v.null(), handler: async (ctx, args) => { - const stream = await ctx.db.get("streams", args.streamId); - if (!stream) { - throw new Error("Stream not found"); - } + const stream = await requireStream(ctx, args.streamId); if (stream.status !== "pending" && stream.status !== "streaming") { console.log( "Stream is already finalized; ignoring status change", stream, ); - return; + return null; } - await ctx.db.patch("streams", args.streamId, { - status: args.status, - }); + + if (stream.coreId !== undefined) { + if (args.status === "done") { + await textStreams.complete(ctx, { streamId: stream.coreId }); + } else if (args.status === "error" || args.status === "timeout") { + await textStreams.fail(ctx, { + streamId: stream.coreId, + error: { + code: args.status, + message: + args.status === "timeout" + ? "Stream generation timed out." + : "Stream generation failed.", + }, + }); + } else { + const core = await textStreams.get(ctx, { streamId: stream.coreId }); + if (core === null || core.status !== args.status) { + throw new Error( + `Cannot set a Stream-backed stream to ${args.status}.`, + ); + } + } + } + + await ctx.db.patch("streams", args.streamId, { status: args.status }); + return null; }, }); -// Get the status of a stream. export const getStreamStatus = query({ - args: { - streamId: v.id("streams"), - }, + args: { streamId: v.id("streams") }, returns: streamStatusValidator, handler: async (ctx, args) => { const stream = await ctx.db.get("streams", args.streamId); @@ -91,61 +487,50 @@ export const getStreamStatus = query({ }, }); -// Get the full text of a stream. -// Involves concatenating all the chunks. +// Preserve the existing full-body API. Legacy rows join `chunks`; current rows +// join the active attempt's ordered Stream events. export const getStreamText = query({ - args: { - streamId: v.id("streams"), - }, + args: { streamId: v.id("streams") }, returns: v.object({ text: v.string(), status: streamStatusValidator, }), handler: async (ctx, args) => { - const stream = await ctx.db.get("streams", args.streamId); - if (!stream) { - throw new Error("Stream not found"); - } - let text = ""; - if (stream.status !== "pending") { - const chunks = await ctx.db - .query("chunks") - .withIndex("byStream", (q) => q.eq("streamId", args.streamId)) - .collect(); - text = chunks.map((chunk) => chunk.text).join(""); - } + const stream = await requireStream(ctx, args.streamId); + if (stream.coreId !== undefined) return currentText(ctx, stream.coreId); return { - text, + text: + stream.status === "pending" ? "" : await legacyText(ctx, args.streamId), status: stream.status, }; }, }); -const EXPIRATION_TIME = 20 * 60 * 1000; // 20 minutes in milliseconds -const BATCH_SIZE = 100; -const DELETE_BATCH_SIZE = 64; - -// Delete a stream and all its chunks. -// The stream is deleted immediately; chunks are cleaned up asynchronously. +// Delete the stable handle immediately, matching the legacy contract. Stream +// drains its event rows in bounded, self-scheduling transactions. export const deleteStream = mutation({ - args: { - streamId: v.id("streams"), - }, + args: { streamId: v.id("streams") }, returns: v.null(), handler: async (ctx, args) => { - const stream = await ctx.db.get("streams", args.streamId); - if (!stream) { - throw new Error(`Stream ${args.streamId} not found`); + const stream = await requireStream(ctx, args.streamId); + if (stream.coreId !== undefined) { + await textStreams.delete( + ctx, + { streamId: stream.coreId }, + { run: internal.lib.run }, + ); + } else { + await ctx.scheduler.runAfter(0, internal.lib._deleteChunksPage, { + streamId: args.streamId, + cursor: null, + }); } await ctx.db.delete("streams", args.streamId); - await ctx.scheduler.runAfter(0, internal.lib._deleteChunksPage, { - streamId: args.streamId, - cursor: null, - }); + return null; }, }); -// Internal: delete a page of chunks for a deleted stream, re-scheduling if more remain. +// Legacy deletion continuation. New streams use the Stream runner below. export const _deleteChunksPage = internalMutation({ args: { streamId: v.id("streams"), @@ -158,7 +543,9 @@ export const _deleteChunksPage = internalMutation({ .withIndex("byStream", (q) => q.eq("streamId", args.streamId)) .paginate({ cursor: args.cursor, numItems: DELETE_BATCH_SIZE }); - await Promise.all(result.page.map((chunk) => ctx.db.delete("chunks", chunk._id))); + await Promise.all( + result.page.map((chunk) => ctx.db.delete("chunks", chunk._id)), + ); if (!result.isDone) { await ctx.scheduler.runAfter(0, internal.lib._deleteChunksPage, { @@ -166,13 +553,23 @@ export const _deleteChunksPage = internalMutation({ cursor: result.continueCursor, }); } + return null; }, }); -// If the last chunk of a stream was added more than 20 minutes ago, -// set the stream to timeout. The action feeding it has to be dead. +// Self-scheduling maintenance boundary for Stream-backed deletion. +export const run = internalMutation({ + args: textStreams.args.run, + returns: textStreams.returns.run, + handler: async (ctx, args): Promise<{ isDone: boolean }> => + textStreams.run(ctx, args, { run: internal.lib.run }), +}); + +// Preserve the existing 20-minute timeout behavior for both engines. Timeout +// retains text; it is a failed lifecycle, not Stream expiration/deletion. export const cleanupExpiredStreams = internalMutation({ args: {}, + returns: v.null(), handler: async (ctx) => { const now = Date.now(); const pendingStreams = await ctx.db @@ -185,12 +582,16 @@ export const cleanupExpiredStreams = internalMutation({ .take(BATCH_SIZE); for (const stream of [...pendingStreams, ...streamingStreams]) { - if (now - stream._creationTime > EXPIRATION_TIME) { - console.log("Cleaning up expired stream", stream._id); - await ctx.db.patch("streams", stream._id, { - status: "timeout", + if (now - stream._creationTime <= EXPIRATION_TIME) continue; + console.log("Timing out expired stream", stream._id); + if (stream.coreId !== undefined) { + await textStreams.fail(ctx, { + streamId: stream.coreId, + error: { code: "timeout", message: "Stream generation timed out." }, }); } + await ctx.db.patch("streams", stream._id, { status: "timeout" }); } + return null; }, }); diff --git a/src/component/schema.ts b/src/component/schema.ts index d7435fc..9f1b0d5 100644 --- a/src/component/schema.ts +++ b/src/component/schema.ts @@ -1,6 +1,8 @@ import { defineSchema, defineTable } from "convex/server"; import { v, type Infer } from "convex/values"; +import { textStreams } from "./streams.js"; + export const streamStatusValidator = v.union( v.literal("pending"), v.literal("streaming"), @@ -13,9 +15,16 @@ export type StreamStatus = Infer; export default defineSchema({ streams: defineTable({ status: streamStatusValidator, + // Optional producer-election marker. It is separate from lifecycle so a + // claimed stream remains pending until its first durable append. + claimedAt: v.optional(v.number()), + // Absent on legacy rows. New rows retain the public ID while delegating + // ordered persistence and lifecycle coordination to `textStreams`. + coreId: v.optional(v.id("textStreams")), }).index("byStatus", ["status"]), chunks: defineTable({ streamId: v.id("streams"), text: v.string(), }).index("byStream", ["streamId"]), + ...textStreams.tables(), }); diff --git a/src/component/streams.ts b/src/component/streams.ts new file mode 100644 index 0000000..e3f38e0 --- /dev/null +++ b/src/component/streams.ts @@ -0,0 +1,16 @@ +import { defineStream, type StreamHandle } from "@convex-dev/stream/server"; +import { v } from "convex/values"; + +// New streams use the shared durable stream engine. The existing `streams` +// table remains the stable public handle, and its optional `coreId` points at +// one of these coordination rows. +const textEvent = v.string(); + +export const textStreams: StreamHandle< + "textStreams", + typeof textEvent, + Record +> = defineStream("textStreams", { + event: textEvent, + eventFields: {}, +}); diff --git a/src/react/index.test.tsx b/src/react/index.test.tsx new file mode 100644 index 0000000..e13f593 --- /dev/null +++ b/src/react/index.test.tsx @@ -0,0 +1,253 @@ +import { createElement, StrictMode } from "react"; +import { act, create, type ReactTestRenderer } from "react-test-renderer"; +import type { FunctionReference } from "convex/server"; +import { afterEach, beforeAll, describe, expect, it, vi } from "vitest"; + +import type { StreamBody, StreamId } from "../client/index.js"; +import type { StreamTextQuery } from "./index.js"; +import type { TextTransportSink } from "./transport.js"; + +const mocks = vi.hoisted(() => ({ + startTextTransport: vi.fn(), + useQuery: vi.fn(), + useStreamQuery: vi.fn(), +})); + +vi.mock("convex/react", () => ({ useQuery: mocks.useQuery })); +vi.mock("@convex-dev/stream/react", () => ({ + useStream: mocks.useStreamQuery, +})); +vi.mock("./transport.js", () => ({ + startTextTransport: mocks.startTextTransport, +})); + +import { useStream } from "./index.js"; + +const getPersistentBody = {} as FunctionReference< + "query", + "public", + { streamId: string }, + StreamBody +>; +const readStream = {} as StreamTextQuery; +const streamId = "stream-1" as StreamId; +const streamUrl = new URL("https://example.com/chat"); + +type Session = { + close: ReturnType; + sink: TextTransportSink; +}; + +function snapshot( + text: string, + overrides: Record = {}, +): Record { + return { + events: text.length === 0 ? [] : [{ attempt: 0, seq: 0, event: text }], + status: "streaming", + isDone: false, + caughtUp: false, + ...overrides, + }; +} + +const emptySnapshot = snapshot("", { status: null }); + +let mounted: ReactTestRenderer | null = null; + +beforeAll(() => { + Object.assign(globalThis, { IS_REACT_ACT_ENVIRONMENT: true }); +}); + +afterEach(async () => { + if (mounted !== null) { + await act(async () => mounted?.unmount()); + mounted = null; + } + mocks.startTextTransport.mockReset(); + mocks.useQuery.mockReset(); + mocks.useStreamQuery.mockReset(); +}); + +function trackTransports(sessions: Session[]) { + mocks.startTextTransport.mockImplementation( + (_config: unknown, sink: TextTransportSink) => { + const close = vi.fn(); + sessions.push({ close, sink }); + return { close, closed: new Promise(() => undefined) }; + }, + ); +} + +function probe(driven: boolean, opts?: Record) { + return function Probe() { + const body = useStream( + getPersistentBody, + streamUrl, + driven, + streamId, + opts as never, + ); + return createElement("output", { "data-status": body.status }, body.text); + }; +} + +describe("useStream mounted lifecycle", () => { + it("fences StrictMode cleanup and keeps both queries skipped while driving", async () => { + const sessions: Session[] = []; + mocks.useQuery.mockReturnValue(undefined); + mocks.useStreamQuery.mockReturnValue(emptySnapshot); + trackTransports(sessions); + + await act(async () => { + mounted = create( + createElement(StrictMode, null, createElement(probe(true))), + ); + }); + const renderer = mounted as ReactTestRenderer; + + expect(sessions).toHaveLength(2); + const stale = sessions[0]!; + const active = sessions[1]!; + expect(stale.close).toHaveBeenCalledOnce(); + expect(active.close).not.toHaveBeenCalled(); + expect(mocks.useQuery.mock.calls.every(([, args]) => args === "skip")).toBe( + true, + ); + expect( + mocks.useStreamQuery.mock.calls.every(([, args]) => args === "skip"), + ).toBe(true); + + await act(async () => { + stale.sink.publish({ text: "stale", status: "done" }); + stale.sink.handoff(); + }); + expect(renderer.toJSON()).toMatchObject({ + children: null, + props: { "data-status": "pending" }, + }); + + await act(async () => { + active.sink.publish({ text: "live", status: "streaming" }); + }); + expect(renderer.toJSON()).toMatchObject({ + children: ["live"], + props: { "data-status": "streaming" }, + }); + + await act(async () => { + stale.sink.publish({ text: "late stale", status: "done" }); + }); + expect(renderer.toJSON()).toMatchObject({ + children: ["live"], + props: { "data-status": "streaming" }, + }); + + await act(async () => renderer.unmount()); + mounted = null; + expect(active.close).toHaveBeenCalledOnce(); + }); + + it("reads a passive follower durably without any drive request", async () => { + mocks.useQuery.mockReturnValue(undefined); + mocks.useStreamQuery.mockReturnValue( + snapshot("persisted", { status: "done", isDone: true }), + ); + trackTransports([]); + + await act(async () => { + mounted = create(createElement(probe(false, { readStream }))); + }); + + expect(mocks.startTextTransport).not.toHaveBeenCalled(); + expect(mocks.useStreamQuery.mock.calls.at(-1)?.[1]).toEqual({ streamId }); + expect(mocks.useQuery.mock.calls.every(([, args]) => args === "skip")).toBe( + true, + ); + expect(mounted?.toJSON()).toMatchObject({ + children: ["persisted"], + props: { "data-status": "done" }, + }); + }); + + it("maps a failed durable lifecycle onto the published status", async () => { + mocks.useQuery.mockReturnValue(undefined); + mocks.useStreamQuery.mockReturnValue( + snapshot("partial", { + status: "failed", + error: { code: "timeout", message: "Stream generation timed out." }, + isDone: true, + }), + ); + trackTransports([]); + + await act(async () => { + mounted = create(createElement(probe(false, { readStream }))); + }); + + expect(mounted?.toJSON()).toMatchObject({ + props: { "data-status": "timeout" }, + }); + }); + + it("holds the raw prefix after handoff until the replay catches up", async () => { + const sessions: Session[] = []; + mocks.useQuery.mockReturnValue(undefined); + mocks.useStreamQuery.mockReturnValue(emptySnapshot); + trackTransports(sessions); + + const Probe = probe(true, { readStream }); + await act(async () => { + mounted = create(createElement(Probe)); + }); + const renderer = mounted as ReactTestRenderer; + const session = sessions[0]!; + + await act(async () => { + session.sink.publish({ text: "raw-prefix", status: "streaming" }); + session.sink.handoff(); + }); + expect(renderer.toJSON()).toMatchObject({ children: ["raw-prefix"] }); + + // A short replay page must not rewind the reader behind the raw prefix. + mocks.useStreamQuery.mockReturnValue(snapshot("raw-")); + await act(async () => renderer.update(createElement(Probe))); + expect(renderer.toJSON()).toMatchObject({ children: ["raw-prefix"] }); + + mocks.useStreamQuery.mockReturnValue( + snapshot("raw-prefix and the durable tail", { + status: "done", + isDone: true, + }), + ); + await act(async () => renderer.update(createElement(Probe))); + expect(renderer.toJSON()).toMatchObject({ + children: ["raw-prefix and the durable tail"], + props: { "data-status": "done" }, + }); + }); + + it("falls back to the full-body query when readStream is not provided", async () => { + const sessions: Session[] = []; + mocks.useQuery.mockReturnValue({ text: "whole body", status: "done" }); + mocks.useStreamQuery.mockReturnValue(emptySnapshot); + trackTransports(sessions); + + const Probe = probe(true); + await act(async () => { + mounted = create(createElement(Probe)); + }); + const renderer = mounted as ReactTestRenderer; + + await act(async () => sessions[0]!.sink.handoff()); + + expect(mocks.useQuery.mock.calls.at(-1)?.[1]).toEqual({ streamId }); + expect( + mocks.useStreamQuery.mock.calls.every(([, args]) => args === "skip"), + ).toBe(true); + expect(renderer.toJSON()).toMatchObject({ + children: ["whole body"], + props: { "data-status": "done" }, + }); + }); +}); diff --git a/src/react/index.ts b/src/react/index.ts index eee750a..254cf33 100644 --- a/src/react/index.ts +++ b/src/react/index.ts @@ -1,11 +1,39 @@ "use client"; /// React helpers for persistent text streaming. -import type { StreamStatus } from "../component/schema.js"; +import type { StreamQueryArgs, StreamReadResult } from "@convex-dev/stream"; +import { useStream as useStreamQuery } from "@convex-dev/stream/react"; import { useQuery } from "convex/react"; -import type { StreamBody, StreamId } from "../client/index.js"; -import { useEffect, useMemo, useRef, useState } from "react"; import type { FunctionReference } from "convex/server"; +import { useEffect, useMemo, useRef, useState } from "react"; + +import type { StreamBody, StreamId } from "../client/index.js"; +import { publicStatus } from "./status.js"; +import { startTextTransport } from "./transport.js"; + +const EMPTY_BODY: StreamBody = { text: "", status: "pending" }; +// Page size for one durable read. Each page is a bounded delta, never the +// whole body, so this caps the work of a single query execution. +const READ_ITEMS = 16; + +/** + * An app-owned query that exposes `PersistentTextStreaming.readStream`. + */ +export type StreamTextQuery = FunctionReference< + "query", + "public", + { streamId: string; streamArgs: StreamQueryArgs }, + StreamReadResult +>; + +function stableHeaders( + authToken: string | null | undefined, + headers: Record | undefined, +): Record { + const value = new Headers(headers); + if (authToken) value.set("Authorization", `Bearer ${authToken}`); + return Object.fromEntries(value.entries()); +} /** * React hook for persistent text streaming. @@ -39,124 +67,115 @@ export function useStream( authToken?: string | null; // If provided, these will be passed as additional headers. headers?: Record; + // An app-owned query exposing `readStream`. When provided, followers and + // recovery read bounded append-only pages over the normal Convex + // subscription instead of re-reading the full body on every append. + readStream?: StreamTextQuery; }, ) { - const [streamBody, setStreamBody] = useState(""); - const [streamEnded, setStreamEnded] = useState(null); - - // Track the active streamId to handle multiple streams and serve as a - // Strict Mode guard (prevents double-firing when the same streamId is seen). - const activeStreamRef = useRef(undefined); - - const usePersistence = useMemo(() => { - // Something is wrong with the stream, so we need to use the database value. - if (streamEnded === false) { - return true; - } - // If we're not driving the stream, we must use the database value. - if (!driven) { - return true; - } - // Otherwise, we'll try to drive the stream and use the HTTP response. - return false; - }, [driven, streamEnded]); + const url = streamUrl.toString(); + const transportKey = JSON.stringify({ + driven, + streamId: streamId ?? null, + url, + }); + + // Headers ride along with each request rather than keying the transport. An + // auth token that rotates mid-stream must not tear down a live connection or + // blank the text the reader is watching. + const headersKey = JSON.stringify( + stableHeaders(opts?.authToken, opts?.headers), + ); + const headers = useMemo( + () => JSON.parse(headersKey) as Record, + [headersKey], + ); + const headersRef = useRef>({}); + useEffect(() => { + headersRef.current = headers; + }, [headers]); + + const [view, setView] = useState<{ body: StreamBody; key: string }>(() => ({ + body: EMPTY_BODY, + key: transportKey, + })); + const [durableKey, setDurableKey] = useState(null); + const generationRef = useRef(0); + + // A passive client never drives, so it reads durably from the first render. + const readDurably = !driven || durableKey === transportKey; + const readStream = opts?.readStream; + const canRead = readDurably && streamId !== undefined; + + // Rules of hooks require an unconditional call. When the app has not adopted + // `readStream` the args are always "skip", so the placeholder is never run. + const snapshot = useStreamQuery( + (readStream ?? getPersistentBody) as unknown as StreamTextQuery, + canRead && readStream !== undefined ? { streamId } : "skip", + { numItems: READ_ITEMS, maxEvents: null, maxBytes: null }, + ); + // Compatibility path for apps that have not exposed `readStream`, and the + // last resort when the durable read itself cannot resolve. const persistentBody = useQuery( getPersistentBody, - usePersistence && streamId ? { streamId } : "skip", + canRead && readStream === undefined ? { streamId } : "skip", ); useEffect(() => { - if (!driven || !streamId) { - return; - } - - // Strict Mode guard: don't restart streaming for the same streamId - if (streamId === activeStreamRef.current) { - return; - } - - // New stream: reset state and track the new streamId - activeStreamRef.current = streamId; - setStreamBody(""); - setStreamEnded(null); - - const controller = new AbortController(); - - void (async () => { - try { - const response = await fetch(streamUrl, { - method: "POST", - body: JSON.stringify({ streamId }), - headers: { - "Content-Type": "application/json", - ...opts?.headers, - ...(opts?.authToken - ? { Authorization: `Bearer ${opts.authToken}` } - : {}), - }, - signal: controller.signal, - }); - - if (response.status === 205) { - console.error("Stream already finished", response); - setStreamEnded(false); - return; - } - if (!response.ok) { - console.error("Failed to reach streaming endpoint", response); - setStreamEnded(false); - return; - } - if (!response.body) { - console.error("No body in response", response); - setStreamEnded(false); - return; - } - - const reader = response.body.getReader(); - const decoder = new TextDecoder(); - - for (;;) { - const { done, value } = await reader.read(); - const text = decoder.decode(value, { stream: !done }); - if (text) { - setStreamBody((prev) => prev + text); - } - if (done) { - setStreamEnded(true); - return; - } - } - } catch (e) { - if (!controller.signal.aborted) { - console.error("Error reading stream", e); - setStreamEnded(false); - } - } - })(); + const generation = generationRef.current + 1; + generationRef.current = generation; + if (!driven || !streamId) return; + + const isCurrent = () => generationRef.current === generation; + const transport = startTextTransport( + { headers: headersRef.current, streamId, url }, + { + publish(body) { + if (isCurrent()) setView({ body, key: transportKey }); + }, + handoff() { + if (isCurrent()) setDurableKey(transportKey); + }, + report(error) { + if (isCurrent()) + console.error("Persistent text stream transport failed", error); + }, + }, + ); return () => { - controller.abort(); + generationRef.current += 1; + transport.close(); }; - }, [driven, streamId, streamUrl, opts?.authToken, opts?.headers]); - - const body = useMemo(() => { - if (persistentBody) { - return persistentBody; - } - let status: StreamStatus; - if (streamEnded === null) { - status = streamBody.length > 0 ? "streaming" : "pending"; - } else { - status = streamEnded ? "done" : "error"; - } - return { - text: streamBody, - status: status as StreamStatus, - }; - }, [persistentBody, streamBody, streamEnded]); + }, [driven, streamId, transportKey, url]); - return body; -} + const durableText = useMemo( + () => snapshot.events.map((event) => event.event).join(""), + [snapshot.events], + ); + return useMemo(() => { + const raw = view.key === transportKey ? view.body : EMPTY_BODY; + if (!canRead) return raw; + if (readStream === undefined) return persistentBody ?? raw; + if (snapshot.status === null) return raw; + + // Durable replay restarts at zero and the raw text is a prefix of it, so + // hold the raw prefix until the replay has caught up rather than rewinding + // the reader to an empty message. + const status = publicStatus(snapshot.status, snapshot.error); + if (!snapshot.isDone && durableText.length < raw.text.length) return raw; + return { text: durableText, status }; + }, [ + canRead, + durableText, + persistentBody, + readStream, + snapshot.error, + snapshot.isDone, + snapshot.status, + transportKey, + view, + ]); +} diff --git a/src/react/status.ts b/src/react/status.ts new file mode 100644 index 0000000..693a9e9 --- /dev/null +++ b/src/react/status.ts @@ -0,0 +1,24 @@ +import type { + StreamError, + StreamStatus as CoreStreamStatus, +} from "@convex-dev/stream"; + +import type { StreamStatus } from "../component/schema.js"; + +/** + * Project a Stream lifecycle onto this component's published status union. + * + * `failed` carries the distinction between a producer error and the expiration + * job's timeout; `canceled` has no public counterpart and reads as `error`. + */ +export function publicStatus( + status: CoreStreamStatus | null, + error?: StreamError, +): StreamStatus { + if (status === null) return "pending"; + if (status === "failed") { + return error?.code === "timeout" ? "timeout" : "error"; + } + if (status === "canceled") return "error"; + return status; +} diff --git a/src/react/transport.test.ts b/src/react/transport.test.ts new file mode 100644 index 0000000..13712f6 --- /dev/null +++ b/src/react/transport.test.ts @@ -0,0 +1,328 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import type { StreamBody, StreamId } from "../client/index.js"; +import { + CadencedText, + MAX_DRIVE_ATTEMPTS, + startTextTransport, + VISIBLE_COMMIT_MS, +} from "./transport.js"; + +const streamId = "stream-1" as StreamId; + +function config(url = "https://example.com/chat") { + return { headers: {}, streamId, url }; +} + +async function settle(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("text transport", () => { + it("posts to the action URL unchanged", async () => { + const fetchMock = vi.fn(async () => new Response(null, { status: 205 })); + const transport = startTextTransport( + config("https://example.com/chat?room=one"), + { publish: vi.fn(), handoff: vi.fn() }, + { fetch: fetchMock as typeof fetch }, + ); + + await transport.closed; + expect(String((fetchMock.mock.calls[0] as unknown[])?.[0])).toBe( + "https://example.com/chat?room=one", + ); + }); + + it("hands off a 205 drive response without retrying", async () => { + const fetchMock = vi.fn(async () => new Response(null, { status: 205 })); + const handoff = vi.fn(); + const sleep = vi.fn(async () => undefined); + const transport = startTextTransport( + config(), + { publish: vi.fn(), handoff }, + { fetch: fetchMock as typeof fetch, sleep }, + ); + + await transport.closed; + expect(fetchMock).toHaveBeenCalledOnce(); + expect(handoff).toHaveBeenCalledOnce(); + expect(sleep).not.toHaveBeenCalled(); + }); + + it("retries ambiguous drive failures with capped backoff", async () => { + const fetchMock = vi + .fn() + .mockRejectedValueOnce(new TypeError("network unavailable")) + .mockResolvedValueOnce(new Response(null, { status: 500 })) + .mockRejectedValueOnce(new TypeError("network unavailable again")) + .mockResolvedValueOnce(new Response(null, { status: 503 })) + .mockResolvedValueOnce(new Response(null, { status: 502 })) + .mockResolvedValueOnce(new Response(null, { status: 205 })); + const sleep = vi.fn( + async (_milliseconds: number, _signal: AbortSignal) => undefined, + ); + const handoff = vi.fn(); + const transport = startTextTransport( + config(), + { publish: vi.fn(), handoff, report: vi.fn() }, + { fetch: fetchMock as typeof fetch, sleep }, + ); + + await transport.closed; + expect(fetchMock).toHaveBeenCalledTimes(6); + expect(sleep.mock.calls.map(([milliseconds]) => milliseconds)).toEqual([ + 100, 200, 400, 800, 1_600, + ]); + expect(handoff).toHaveBeenCalledOnce(); + }); + + it("retries throttled and timed-out drive responses", async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(null, { status: 408 })) + .mockResolvedValueOnce(new Response(null, { status: 429 })) + .mockResolvedValueOnce(new Response(null, { status: 205 })); + const sleep = vi.fn( + async (_milliseconds: number, _signal: AbortSignal) => undefined, + ); + const transport = startTextTransport( + config(), + { publish: vi.fn(), handoff: vi.fn() }, + { fetch: fetchMock as typeof fetch, sleep }, + ); + + await transport.closed; + expect(fetchMock).toHaveBeenCalledTimes(3); + expect(sleep.mock.calls.map(([milliseconds]) => milliseconds)).toEqual([ + 100, 200, + ]); + }); + + it("stops retrying at the attempt cap and hands off", async () => { + const fetchMock = vi.fn(async () => new Response(null, { status: 500 })); + const sleep = vi.fn(async () => undefined); + const handoff = vi.fn(); + const transport = startTextTransport( + config(), + { publish: vi.fn(), handoff, report: vi.fn() }, + { fetch: fetchMock as typeof fetch, sleep }, + ); + + await transport.closed; + expect(fetchMock).toHaveBeenCalledTimes(MAX_DRIVE_ATTEMPTS); + expect(handoff).toHaveBeenCalledOnce(); + }); + + it("hands off an ok drive response that is missing its body", async () => { + const fetchMock = vi.fn(async () => new Response(null, { status: 200 })); + const handoff = vi.fn(); + const transport = startTextTransport( + config(), + { publish: vi.fn(), handoff }, + { fetch: fetchMock as typeof fetch }, + ); + + await transport.closed; + expect(fetchMock).toHaveBeenCalledOnce(); + expect(handoff).toHaveBeenCalledOnce(); + }); + + it("hands off a nonretryable drive rejection immediately", async () => { + const handoff = vi.fn(); + const sleep = vi.fn(async () => undefined); + const transport = startTextTransport( + config(), + { publish: vi.fn(), handoff }, + { + fetch: vi.fn(async () => new Response(null, { status: 401 })), + sleep, + }, + ); + + await transport.closed; + expect(handoff).toHaveBeenCalledOnce(); + expect(sleep).not.toHaveBeenCalled(); + }); + + it("stops retrying when closed", async () => { + const fetchMock = vi.fn(async () => { + throw new TypeError("offline"); + }); + let retrySignal: AbortSignal | undefined; + const sleep = vi.fn( + (_milliseconds: number, signal: AbortSignal) => + new Promise((resolve) => { + retrySignal = signal; + signal.addEventListener("abort", () => resolve(), { once: true }); + }), + ); + const handoff = vi.fn(); + const transport = startTextTransport( + config(), + { publish: vi.fn(), handoff, report: vi.fn() }, + { fetch: fetchMock as typeof fetch, sleep }, + ); + + await vi.waitFor(() => expect(sleep).toHaveBeenCalledOnce()); + transport.close(); + await transport.closed; + await settle(); + expect(retrySignal?.aborted).toBe(true); + expect(fetchMock).toHaveBeenCalledOnce(); + expect(handoff).not.toHaveBeenCalled(); + }); + + it("hands off after a mid-body disconnect instead of retrying", async () => { + vi.useFakeTimers(); + let rejectRead!: (error: unknown) => void; + let reads = 0; + const rawResponse = { + status: 200, + ok: true, + headers: new Headers(), + body: { + getReader() { + return { + read() { + reads += 1; + if (reads === 1) { + return Promise.resolve({ + done: false, + value: new TextEncoder().encode("raw-prefix"), + }); + } + return new Promise((_, reject) => { + rejectRead = reject; + }); + }, + releaseLock() {}, + }; + }, + }, + } as unknown as Response; + const fetchMock = vi.fn(async () => rawResponse); + const published: StreamBody[] = []; + const handoff = vi.fn(); + startTextTransport( + config(), + { + publish: (body) => published.push(body), + handoff, + report: vi.fn(), + }, + { + fetch: fetchMock as typeof fetch, + setTimer: setTimeout, + clearTimer: clearTimeout, + }, + ); + + await settle(); + await vi.advanceTimersByTimeAsync(0); + await settle(); + await vi.advanceTimersByTimeAsync(VISIBLE_COMMIT_MS); + expect(published.at(-1)).toEqual({ + text: "raw-prefix", + status: "streaming", + }); + + rejectRead(new Error("disconnected")); + await settle(); + expect(handoff).toHaveBeenCalledOnce(); + expect(fetchMock).toHaveBeenCalledOnce(); + }); + + it("publishes a completed raw body without handing off", async () => { + vi.useFakeTimers(); + const fetchMock = vi.fn( + async () => new Response("hello there", { status: 200 }), + ); + const published: StreamBody[] = []; + const handoff = vi.fn(); + const transport = startTextTransport( + config(), + { publish: (body) => published.push(body), handoff }, + { + fetch: fetchMock as typeof fetch, + setTimer: setTimeout, + clearTimer: clearTimeout, + }, + ); + + await vi.advanceTimersByTimeAsync(VISIBLE_COMMIT_MS); + await transport.closed; + expect(published.at(-1)).toEqual({ text: "hello there", status: "done" }); + expect(handoff).not.toHaveBeenCalled(); + }); + + it("sends auth, custom headers, and the POST body", async () => { + const fetchMock = vi.fn(async () => new Response(null, { status: 205 })); + const transport = startTextTransport( + { + headers: { + Authorization: "Bearer secret", + "X-Workspace": "workspace", + }, + streamId, + url: "https://example.com/chat", + }, + { publish: vi.fn(), handoff: vi.fn() }, + { fetch: fetchMock as typeof fetch }, + ); + + await transport.closed; + const init = (fetchMock.mock.calls[0] as unknown[])?.[1] as RequestInit; + expect(init.method).toBe("POST"); + expect(init.body).toBe(JSON.stringify({ streamId })); + const headers = new Headers(init.headers); + expect(headers.get("Authorization")).toBe("Bearer secret"); + expect(headers.get("X-Workspace")).toBe("workspace"); + expect(headers.get("Content-Type")).toBe("application/json"); + }); + + it("coalesces token-frequency appends to the fixed visible cadence", async () => { + vi.useFakeTimers(); + const published: StreamBody[] = []; + const text = new CadencedText((body) => published.push(body)); + + for (let index = 0; index < 100; index += 1) text.append("x"); + expect(published).toEqual([]); + await vi.advanceTimersByTimeAsync(VISIBLE_COMMIT_MS - 1); + expect(published).toEqual([]); + await vi.advanceTimersByTimeAsync(1); + expect(published).toEqual([{ text: "x".repeat(100), status: "streaming" }]); + + for (let index = 0; index < 50; index += 1) text.append("y"); + await vi.advanceTimersByTimeAsync(VISIBLE_COMMIT_MS); + expect(published).toHaveLength(2); + expect(published.at(-1)?.text).toBe("x".repeat(100) + "y".repeat(50)); + }); + + it("fences publish and handoff after close", async () => { + vi.useFakeTimers(); + const fetchMock = vi.fn(async () => new Response("late", { status: 200 })); + const publish = vi.fn(); + const handoff = vi.fn(); + const transport = startTextTransport( + config(), + { publish, handoff }, + { + fetch: fetchMock as typeof fetch, + setTimer: setTimeout, + clearTimer: clearTimeout, + }, + ); + transport.close(); + + await vi.advanceTimersByTimeAsync(VISIBLE_COMMIT_MS * 2); + await settle(); + expect(publish).not.toHaveBeenCalled(); + expect(handoff).not.toHaveBeenCalled(); + }); +}); diff --git a/src/react/transport.ts b/src/react/transport.ts new file mode 100644 index 0000000..8da8c7d --- /dev/null +++ b/src/react/transport.ts @@ -0,0 +1,254 @@ +import type { StreamBody, StreamId } from "../client/index.js"; +import type { StreamStatus } from "../component/schema.js"; + +export const VISIBLE_COMMIT_MS = 50; +const DRIVE_RETRY_BASE_MS = 100; +const DRIVE_RETRY_MAX_MS = 5_000; +// A drive request that keeps failing must eventually yield to the durable read +// path rather than polling the app's endpoint for the component's lifetime. +export const MAX_DRIVE_ATTEMPTS = 6; + +type Timer = ReturnType; + +export type TextTransportSink = { + publish: (body: StreamBody) => void; + /** The raw path cannot finish; the caller should read durably instead. */ + handoff: () => void; + report?: (error: unknown) => void; +}; + +export type TextTransportConfig = { + headers: HeadersInit; + streamId: StreamId; + url: string; +}; + +export type TextTransportDependencies = { + fetch: typeof fetch; + setTimer: typeof setTimeout; + clearTimer: typeof clearTimeout; + sleep: (milliseconds: number, signal: AbortSignal) => Promise; +}; + +export type TextTransport = { + close: () => void; + closed: Promise; +}; + +const defaultDependencies: TextTransportDependencies = { + fetch: globalThis.fetch.bind(globalThis), + setTimer: globalThis.setTimeout.bind(globalThis), + clearTimer: globalThis.clearTimeout.bind(globalThis), + sleep: (milliseconds, signal) => { + if (signal.aborted) return Promise.resolve(); + return new Promise((resolve) => { + const timer = setTimeout(done, milliseconds); + function done() { + clearTimeout(timer); + signal.removeEventListener("abort", done); + resolve(); + } + signal.addEventListener("abort", done, { once: true }); + }); + }, +}; + +export class CadencedText { + private committed = ""; + private pending: string[] = []; + private status: StreamStatus = "pending"; + private timer: Timer | null = null; + private active = true; + + constructor( + private readonly publish: (body: StreamBody) => void, + private readonly setTimer: typeof setTimeout = setTimeout, + private readonly clearTimer: typeof clearTimeout = clearTimeout, + ) {} + + append(text: string): void { + if (!this.active || text.length === 0) return; + this.pending.push(text); + if (this.status === "pending") this.status = "streaming"; + this.schedule(); + } + + setStatus(status: StreamStatus): void { + if (!this.active) return; + this.status = status; + this.schedule(); + } + + flush(): void { + if (!this.active) return; + if (this.timer !== null) { + this.clearTimer(this.timer); + this.timer = null; + } + if (this.pending.length > 0) { + this.committed += this.pending.join(""); + this.pending = []; + } + this.publish({ text: this.committed, status: this.status }); + } + + close(): void { + if (!this.active) return; + this.active = false; + if (this.timer !== null) this.clearTimer(this.timer); + this.timer = null; + this.pending = []; + } + + private schedule(): void { + if (this.timer !== null) return; + this.timer = this.setTimer(() => { + this.timer = null; + this.flush(); + }, VISIBLE_COMMIT_MS); + } +} + +function requestHeaders(headers: HeadersInit): Headers { + const result = new Headers(headers); + result.set("Content-Type", "application/json"); + return result; +} + +function driveRetryDelay(attempt: number): number { + return Math.min( + DRIVE_RETRY_BASE_MS * 2 ** Math.min(attempt, 6), + DRIVE_RETRY_MAX_MS, + ); +} + +function retryableDriveStatus(status: number): boolean { + return status === 408 || status === 429 || status >= 500; +} + +/** + * Drive one stream over a plain POST and publish its raw body at a fixed + * cadence. + * + * This is the low-latency path for the single browser that generates the text. + * Every other outcome -- a lost claim, a mid-flight disconnect, or exhausted + * retries -- hands off to the caller, which reads the same stream durably + * through an app-owned Convex query. + */ +export function startTextTransport( + config: TextTransportConfig, + sink: TextTransportSink, + dependencies: Partial = {}, +): TextTransport { + const deps = { ...defaultDependencies, ...dependencies }; + const controller = new AbortController(); + const body = JSON.stringify({ streamId: config.streamId }); + const headers = requestHeaders(config.headers); + let active = true; + const raw = new CadencedText(sink.publish, deps.setTimer, deps.clearTimer); + let settleClosed!: () => void; + const closed = new Promise((resolve) => { + settleClosed = resolve; + }); + let settled = false; + + const settle = () => { + if (settled) return; + settled = true; + settleClosed(); + }; + const report = (error: unknown) => { + if (!controller.signal.aborted) sink.report?.(error); + }; + const handoff = () => { + if (!active || controller.signal.aborted) return; + raw.close(); + sink.handoff(); + settle(); + }; + const waitForRetry = async (attempt: number): Promise => { + await deps.sleep(driveRetryDelay(attempt), controller.signal); + return active && !controller.signal.aborted; + }; + + const drive = async () => { + for ( + let attempt = 0; + active && !controller.signal.aborted && attempt < MAX_DRIVE_ATTEMPTS; + attempt += 1 + ) { + let receivedRawBytes = false; + try { + const response = await deps.fetch(config.url, { + method: "POST", + body, + headers, + signal: controller.signal, + }); + // Another request already owns production, so nothing will arrive here. + if (response.status === 205) { + void response.body?.cancel().catch(() => undefined); + handoff(); + return; + } + if (retryableDriveStatus(response.status)) { + void response.body?.cancel().catch(() => undefined); + if (!(await waitForRetry(attempt))) return; + continue; + } + if (!response.ok || response.body === null) { + void response.body?.cancel().catch(() => undefined); + handoff(); + return; + } + + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + try { + for (;;) { + const chunk = await reader.read(); + if (!chunk.done && chunk.value.byteLength > 0) { + receivedRawBytes = true; + } + const text = chunk.done + ? decoder.decode() + : decoder.decode(chunk.value, { stream: true }); + if (text.length > 0) raw.append(text); + if (chunk.done) { + raw.setStatus("done"); + raw.flush(); + settle(); + return; + } + } + } finally { + reader.releaseLock(); + } + } catch (error) { + if (!active || controller.signal.aborted) return; + report(error); + // Partial raw text is already on screen but its tail is unknown, so the + // durable replay must supersede it rather than resume behind it. + if (receivedRawBytes) { + handoff(); + return; + } + if (!(await waitForRetry(attempt))) return; + } + } + handoff(); + }; + + void drive(); + + return { + close() { + if (!active) return; + active = false; + controller.abort(); + raw.close(); + settle(); + }, + closed, + }; +}