Uh oh!
There was an error while loading. Please reload this page.
[world] Add stream pagination and metadata endpoints - #1470
Conversation
Negative startIndex values (e.g. -3) resolve to n chunks before the known end of the stream. All world implementations (local, postgres, vercel) support this. Includes unit tests, e2e test, and doc updates. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Peter Wielander <mittgfu@gmail.com>
Replaces three separate tests (no startIndex, positive, negative) with a single describe block that iterates over startIndex cases. Fixes the negative startIndex test by waiting for workflow completion before connecting the reader — the backend resolves negative indices at connection time using knownChunkCount, which is 0 if the stream hasn't been fully written yet. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Replaces three separate tests (no startIndex, positive, negative) with a single describe block that iterates over startIndex cases. Fixes the negative startIndex test by waiting for workflow completion before connecting the reader — the backend resolves negative indices at connection time using knownChunkCount, which is 0 if the stream hasn't been fully written yet. Signed-off-by: Peter Wielander <mittgfu@gmail.com> Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Clarify that negative startIndex is for custom consumers, not WorkflowChatTransport - Add live-stream caveat for negative startIndex resolution timing - Add pagination limitation callout for live streams - Use explicit typeof guards for negative startIndex checks (world-local, world-postgres) - Add cost comment for EOF marker disk read in world-local Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
🦋 Changeset detectedLatest commit: a82fbe7 The changes in this PR will be included in the next version bump. This PR includes changesets to release 20 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
📊 Benchmark Results
workflow with no steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 1 step💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 10 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) | Nitro workflow with 25 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 50 sequential steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) Promise.all with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) Promise.all with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) Promise.all with 50 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) Promise.race with 10 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) | Nitro Promise.race with 25 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) Promise.race with 50 concurrent steps💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 10 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 25 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 50 sequential data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 10 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 25 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) workflow with 50 concurrent data payload steps (10KB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) | Nitro Stream Benchmarks(includes TTFB metrics)workflow with stream💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) stream pipeline with 5 transform steps (1MB)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) | Nitro 10 parallel streams (1MB each)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Next.js (Turbopack) | Nitro fan-out fan-in 10 streams (1MB each)💻 Local Development
▲ Production (Vercel)
🔍 Observability: Nitro | Next.js (Turbopack) SummaryFastest Framework by WorldWinner determined by most benchmark wins
Fastest World by FrameworkWinner determined by most benchmark wins
Column Definitions
Worlds:
❌ Some benchmark jobs failed:
Check the workflow run for details. |
🧪 E2E Test Results❌ Some tests failed Summary
❌ Failed Tests🌍 Community Worlds (56 failed)mongodb (3 failed):
redis (2 failed):
turso (51 failed):
Details by Category✅ ▲ Vercel Production
✅ 💻 Local Development
✅ 📦 Local Production
✅ 🐘 Local Postgres
✅ 🪟 Windows
❌ 🌍 Community Worlds
✅ 📋 Other
|
Uh oh!
There was an error while loading. Please reload this page.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
karthikscale3
left a comment
There was a problem hiding this comment.
Here's my review of PR #1470:
Summary
This is a feature PR by @VaguelySerious that adds two new Streamer interface methods (getStreamChunks and getStreamInfo) across all world implementations, extends Run.getReadable() with a getTailIndex() helper, and adds initialStartIndex support to WorkflowChatTransport for resuming streams from the end.
No Blockers
CI is in excellent shape. The code changes are well-structured, well-documented, and backward-compatible.
CI Status
Passing:
- All Vercel Production E2E tests: 11/11 frameworks pass (802 passed, 0 failed)
- All Local Dev E2E tests: 12/12 pass (782 passed)
- All Local Prod E2E tests: 12/12 pass
- All Local Postgres E2E tests: 12/12 pass
- Windows E2E tests pass
- Unit tests pass (ubuntu + windows)
- Vitest plugin tests pass
- All Vercel deployments succeed (except
workbench-nestjs-workflow- discussed below) - Benchmarks pass for local/postgres; no performance regressions
- Docs code samples and links pass
Failing (non-blocking / pre-existing):
workbench-nestjs-workflowVercel deployment fails - this is likely a pre-existing issue unrelated to streaming changes (the nest workbench has no streaming code changes in this PR)- Benchmark Vercel (nitro-v3) fails at ~30min - likely infra timeout, not PR-related
- Community World (Turso) - 51 failures, pre-existing community world issue (same pattern as other PRs)
- Community World (MongoDB/Redis) - 3/2 failures on hook-related tests, pre-existing
Code Review
1. Streamer interface additions (packages/world/src/interfaces.ts) - Clean
Two new methods added to the Streamer interface:
getStreamChunks(name, runId, options?)- cursor-based pagination for stream chunksgetStreamInfo(name, runId)- lightweight metadata (tailIndex, done flag)
New types (StreamChunk, StreamChunksResponse, GetChunksOptions, StreamInfoResponse) are well-documented in packages/world/src/shared.ts. The interface is additive and non-breaking.
2. world-vercel implementation (packages/world-vercel/src/streamer.ts) - Clean
Uses makeRequest + Zod schemas to call two new server endpoints:
GET /v2/runs/:runId/streams/:name/chunks?limit=&cursor=GET /v2/runs/:runId/streams/:name/info
The StreamChunksResponseSchema correctly expects z.instanceof(Uint8Array) since CBOR handles binary natively. Clean and minimal.
3. world-local implementation (packages/world-local/src/streamer.ts) - Works but verbose
The implementation reads chunk files from disk, builds a file extension map, sorts by ULID, and paginates. One observation:
getStreamInforeads every chunk file to count data chunks and detect EOF. For large streams this could be expensive. ThegetStreamChunksimplementation has the same pattern. Both could potentially be optimized with a count-based approach, but for local dev this is fine.- The cursor encoding (base64-encoded JSON
{ i: index }) is a reasonable pattern.
4. world-postgres implementation (packages/world-postgres/src/streamer.ts) - Solid
Uses Drizzle ORM with proper query patterns:
- Fetches
limit + 1rows to detecthasMoreefficiently - Correctly handles the EOF row being mixed into results
- Has a follow-up EOF check query when
!streamDone && !hasMoreto correctly reportdonestatus - Cursor encodes both chunk ID and running index (
{ c: chunkId, i: baseIndex })
One minor cleanup: void fn() on line 407 (changing cleanups.forEach((fn) => fn()) to cleanups.forEach((fn) => void fn())) is a good lint fix to suppress floating promise warnings.
5. WorkflowReadableStream type and getTailIndex() (packages/core/src/runtime/run.ts) - Clean
Uses Object.assign to add getTailIndex() to the ReadableStream. This is a clean pattern that avoids subclassing. The type is properly exported through packages/workflow/src/api.ts.
6. WorkflowChatTransport changes (packages/ai/src/workflow-chat-transport.ts) - Well thought out
The negative initialStartIndex feature is carefully designed:
- First request uses the explicit (potentially negative)
startIndex - On retry, resolves to an absolute position using
x-workflow-stream-tail-indexheader - Falls back to
startIndex=0with a warning if the header is missing useExplicitStartIndexflag ensures the negative value is only used once, then subsequent retries use the runningchunkIndex
The replayFromStart flag handles the edge case where the header is absent gracefully.
7. Tests - Good coverage
- Two new e2e tests for
getTailIndexandgetStreamChunkspagination (skipped on local deployments since they require server-side endpoints) - Two new unit tests for
WorkflowChatTransportcovering negativeinitialStartIndexwith and without the tail-index header
8. Docs - Thorough
Updated resumable-streams.mdx with:
- New "Resuming from the end of the stream" section
- Updated stream reconnection endpoint to include
getTailIndex()and thex-workflow-stream-tail-indexheader - Callout about header requirement for negative
initialStartIndex - API reference docs updated
9. Changeset - Present and correct
Includes @workflow/world, world-local, world-vercel, world-postgres, core, and ai as patch changes.
Minor Notes
- The PR description is empty ("No description provided"). Worth adding a summary for posterity.
- The
getStreamChunksmethod uses_runId(unused) in local/postgres implementations since the stream name is already unique there. The vercel implementation passes it to the server endpoint. This is fine - the interface is designed for the vercel world's needs.
Verdict
No blockers or regressions. This is a well-implemented, additive feature with good test coverage and documentation. The only CI failures are pre-existing (nest deployment, community world Turso/MongoDB/Redis, nitro benchmark timeout). Ready to merge once it gets a review approval.
Signed-off-by: Peter Wielander <mittgfu@gmail.com>
pranaygp
left a comment
There was a problem hiding this comment.
Overall this is a well-designed, additive feature. The negative initialStartIndex logic in the transport is carefully thought through, the interface additions are clean, and the docs/tests cover the new behavior well. A few issues worth addressing below.
| GetChunksOptions, | ||
| StreamChunk, | ||
| StreamChunksResponse, | ||
| } from './shared.js'; |
There was a problem hiding this comment.
The explicit export type { GetChunksOptions, StreamChunk, StreamChunksResponse } is redundant with the export type * from './shared.js' on line 34 — and it's also missing StreamInfoResponse. Since line 34 already re-exports all types from shared.js, you can just remove this block entirely (lines 39-43).
| .select({ chunkId: streams.chunkId }) | ||
| .from(streams) | ||
| .where(and(eq(streams.streamId, name), eq(streams.eof, false))) | ||
| .orderBy(asc(streams.chunkId)); |
There was a problem hiding this comment.
getStreamInfo fetches all data rows into memory just to count them. For streams with thousands of chunks this is wasteful — a SELECT COUNT(*) FROM streams WHERE streamId = $1 AND eof = false would be much cheaper. Consider using Drizzle's count() aggregate or a raw SQL count instead of selecting all chunk IDs.
| hasMore, | ||
| done: streamDone, | ||
| }; | ||
| }, |
There was a problem hiding this comment.
The file-listing + extension-mapping logic (~30 lines) is duplicated verbatim between getStreamChunks and getStreamInfo. Consider extracting a helper like listStreamDataFiles(name) that returns { dataFiles, fileExtMap, streamDone } to keep this DRY.
| break; | ||
| } | ||
| dataFiles.push(file); | ||
| } |
There was a problem hiding this comment.
getStreamChunks reads and deserializes every chunk file from disk on every call (to separate data files from the EOF marker), regardless of the cursor position. For a stream with 10k chunks, page 2 still reads all 10k files.
Two possible improvements:
- Only read files starting from the cursor position (the ULID sort gives chronological order, so you can skip earlier files)
- At minimum, stop reading once you have enough files past the cursor — you only need
startIndex + limitdata files plus potentially the EOF marker
pranaygp
left a comment
There was a problem hiding this comment.
Deeper review: testing, docs, observability, edge cases, and performance
Testing gaps
world-local and world-postgres have zero unit tests for the new getStreamChunks and getStreamInfo methods. The only coverage comes from two e2e tests that are skipIf(isLocalDeployment()), so local dev never exercises these paths. Cursor pagination logic (invalid cursors, out-of-range cursors, single-item pages), empty stream behavior (tailIndex = -1), and mid-flight streams (calling before workflow completes) are all untested.
The transport unit tests cover the happy path for negative initialStartIndex but miss: NaN/malformed x-workflow-stream-tail-index header values (the !Number.isNaN guard is untested), positive explicit startIndex override, boundary clamping when magnitude exceeds stream length, and multi-retry chains where the header varies across attempts.
Documentation gaps
building-a-world.mdxstill shows the oldStreamerinterface withoutgetStreamChunksorgetStreamInfo. Custom world implementers won't know these methods are required.- No API reference for
WorkflowReadableStreamorgetTailIndex()— theget-run.mdxpage documentsWorkflowReadableStreamOptionsbut not the new return type. - Changeset mentions
getStreamInfobut omitsgetStreamChunks.
Observability
Existing storage methods are wrapped with instrumentObject() for automatic OTEL spans (e.g. world.runs.get, world.events.create). Streamer methods — both existing and new — are not instrumented at all. No spans, no semantic conventions for stream names/indices, and no structured logging in any world implementation's streaming code. The only observability signal is the console.warn in WorkflowChatTransport for missing headers. This is consistent with the pre-existing pattern (streaming was never instrumented), but it means production debugging of pagination issues, tail-index mismatches, or reconnection failures will be blind.
Edge cases and performance (see inline comments below)
Several specific issues are called out inline.
| break; | ||
| } | ||
| dataRows.push(row); | ||
| } |
There was a problem hiding this comment.
Edge case: EOF row appearing mid-batch. The query fetches limit + 1 rows ordered by chunkId without filtering out EOF rows. If the EOF row sorts between data rows (e.g. due to concurrent writes or clock skew), the loop breaks early on row.eof, and subsequent data rows in the result set are silently dropped.
Then hasMore = !streamDone && dataRows.length > limit evaluates to false (because streamDone = true), so the client sees done: true, hasMore: false even though there were more data rows in the fetch.
This is probably fine in practice since the EOF ULID is generated after all data chunks, but it's worth noting the implicit ordering assumption. Adding eq(streams.eof, false) to the main query and doing a separate EOF check (like the lookahead on line 250) would make this robust.
| .select({ chunkId: streams.chunkId }) | ||
| .from(streams) | ||
| .where(and(eq(streams.streamId, name), eq(streams.eof, false))) | ||
| .orderBy(asc(streams.chunkId)); |
There was a problem hiding this comment.
Performance: O(N) memory to count rows. This SELECT chunkId FROM streams WHERE ... loads every non-EOF row into Node.js memory just to take .length. For a stream with 100k chunks, that's 100k objects allocated and immediately discarded.
Use a COUNT(*) aggregate instead:
const[{ count }]=awaitdrizzle.select({count: sql<number>`count(*)`}).from(streams).where(and(eq(streams.streamId,name),eq(streams.eof,false)));return{tailIndex: (count??0)-1,done: !!eofRow};| break; | ||
| } | ||
| dataFiles.push(file); | ||
| } |
There was a problem hiding this comment.
Performance: every page reads ALL chunk files.getStreamChunks deserializes every chunk file from disk (lines 286-296) to separate data files from the EOF marker, even for chunks that precede the cursor. Page 50 of a 10k-chunk stream still reads all 10k files.
getStreamInfo (line 372) has the exact same pattern — reads all files just to count them.
For local dev this is tolerable, but if these methods get called frequently (e.g. polling for tail index), the cost adds up. A lighter approach: since ULID sort gives chronological order, you could skip files before the cursor index and stop after collecting limit data files + EOF check.
| const readable = run.getReadable(); | ||
| const tailIndex = await readable.getTailIndex(); | ||
| // outputStreamWorkflow writes 2 chunks to the default stream |
There was a problem hiding this comment.
Testing gap: no empty-stream or mid-flight test. These tests only run after await run.returnValue (stream fully complete) and outputStreamWorkflow always writes exactly 2 chunks.
Missing scenarios:
- Empty stream: What does
getTailIndex()return when 0 chunks have been written? (should be-1per the type docs) - Mid-flight stream: What does
getTailIndex()/getStreamChunks()return while the workflow is still running and the stream isn't closed yet? getStreamInfodirectly: Only tested indirectly viagetTailIndex()— no direct assertion on thedoneflag
| warnSpy.mockRestore(); | ||
| }); | ||
| }); |
There was a problem hiding this comment.
Testing gap: malformed header values. The implementation has a !Number.isNaN(tailIndex) guard (transport line 362) but no test covers it. Consider adding a case where the header is "not-a-number" — parseInt returns NaN, the guard should skip resolution, and the code should fall through without setting chunkIndex.
Also missing: a test for initialStartIndex: -1000 on a 5-chunk stream (should clamp to 0 via Math.max), and a test where a positive startIndex is passed via reconnectToStream options to override initialStartIndex.
| const tailIndexHeader = res.headers.get('x-workflow-stream-tail-index'); | ||
| if (tailIndexHeader !== null) { | ||
| const tailIndex = parseInt(tailIndexHeader, 10); | ||
| if (!Number.isNaN(tailIndex)) { |
There was a problem hiding this comment.
Edge case: NaN guard doesn't fall through cleanly. When parseInt returns NaN, the if (!Number.isNaN(tailIndex)) block is skipped, but replayFromStart is NOT set (that only happens in the else branch for tailIndexHeader === null). So on the next iteration, useExplicitStartIndex is false and replayFromStart is false, meaning the code falls through to chunkIndex — which is still 0 from initialization.
This accidentally works (retries from 0), but the intent is unclear. If the header is present but malformed, should it warn and set replayFromStart = true like the missing-header case? As-is, a malformed header silently degrades to replaying from 0 without the console warning.
| GetChunksOptions, | ||
| StreamChunk, | ||
| StreamChunksResponse, | ||
| } from './shared.js'; |
There was a problem hiding this comment.
Redundant + incomplete export. Line 34 already has export type * from './shared.js' which re-exports all types. This explicit block is redundant and also missing StreamInfoResponse. Just delete lines 39-43.
- world-postgres: use COUNT(*) in getStreamInfo instead of fetching all rows; filter EOF rows in getStreamChunks query to prevent mid-batch data loss from clock-skewed EOF rows - world-local: skip files before cursor in getStreamChunks, read only EOF byte in getStreamInfo and pre-cursor files - WorkflowChatTransport: treat unparseable tail-index header same as missing (set replayFromStart); add unit test for NaN header fallback - e2e: add test for getTailIndex returning -1 on nonexistent stream Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- building-a-world.mdx: add getStreamChunks and getStreamInfo to Streamer interface definition - get-run.mdx: document WorkflowReadableStream and getTailIndex() - Changeset: mention getStreamChunks alongside getStreamInfo - world-vercel, world-local: wrap streamer with instrumentObject for OTEL spans (world.streams.* prefix), matching storage instrumentation Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
pranaygp
left a comment
There was a problem hiding this comment.
Fresh review after update
This update addresses all the critical issues from the previous round. Nice work cleaning this up.
Previously flagged — now resolved
- Postgres
getStreamInfo: Now usesCOUNT(*)aggregate instead of fetching all rows into memory - Postgres
getStreamChunksEOF mid-batch: Main query now filterseq(streams.eof, false), with a separate EOF check — no more risk of EOF silently dropping data rows - Transport NaN guard: Missing and malformed headers now both flow through the same
elsebranch —replayFromStart = true+console.warnin both cases - world-local
getStreamChunksperf: No longer deserializes every chunk — pre-cursor files useisEofChunk()(byte check only), and the loop stops early after collecting enough data - world-local
getStreamInfo: Also usesisEofChunk()instead of full deserialization - Changeset: Now mentions both
getStreamChunks()andgetStreamInfo() - Docs:
building-a-world.mdxincludes the new Streamer methods with explanatory text.get-run.mdxhas aWorkflowReadableStreamsection withgetTailIndex()reference. - E2e tests: Added
tailIndex returns -1test using a nonexistent namespace. Added NaN header unit test. - OTEL: Both world-vercel and world-local now wrap the streamer with
instrumentObject('world.streams', ...) - Redundant export:
StreamInfoResponseadded to the explicit export list inworld/src/index.ts
Minor remaining items (non-blocking)
world-local file-listing boilerplate still duplicated — the ~20-line block that lists files by extension, builds the
fileExtMap, and filters by stream name is copy-pasted betweengetStreamChunks(lines 266-286) andgetStreamInfo(lines 367-389). Extracting alistChunkFilesForStream(name)helper would reduce maintenance surface.world-postgres streamer not instrumented —
world-vercelandworld-localboth wrap their streamer withinstrumentObject('world.streams', ...), giving the new methods automatic OTEL spans.world-postgresspreads the streamer directly without instrumentation (packages/world-postgres/src/index.ts:49). Pre-existing inconsistency, but worth noting since this PR adds new methods that would benefit from tracing.No unit tests for world-local or world-postgres implementations of
getStreamChunks/getStreamInfo. The cursor pagination logic (invalid cursors, boundary conditions, EOF detection) is only tested via e2e tests that are skipped in local dev. Consider adding unit tests in a follow-up.
| dataCount++; | ||
| } | ||
| return { tailIndex: dataCount - 1, done: streamDone }; |
There was a problem hiding this comment.
Nit: The file-listing + extension-mapping block (~20 lines) is duplicated between getStreamChunks (lines 266-286) and getStreamInfo (lines 367-389). A shared listChunkFilesForStream(name) helper returning { chunkFiles, fileExtMap } would keep these two methods DRY.
No description provided.