vercel / vercel/workflow

[world-postgres] streams.get(name, startIndex) buffers the entire stream before serving the first byte

Open
#3,254 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
TypeScript
Stars
2.4k
Forks
365
Avg merge
2d 11h
Merged PRs (30d)
169

Description

Summary

@workflow/world-postgres streams.get(name, startIndex) issues a single unbounded query and discards the first startIndex rows in JavaScript (dist/streamer.js, the historical read inside get()):

const chunks = await drizzle
  .select({ id: streams.chunkId, eof: streams.eof, data: streams.chunkData })
  .from(streams)
  .where(and(eq(streams.streamId, name)))
  .orderBy(streams.chunkId); // no LIMIT/OFFSET — the whole stream

node-pg materializes the full result set before the ReadableStream serves its first byte, so a consumer that only needs the tail of a large stream pays the entire stream in latency and resident memory on every fresh GET.

Real-world impact

Observed twice in one day on our deployment (@workflow/world-postgres@5.0.0-beta.30, eve 0.29.2): agent turns produced streams of 11,597 chunks (~350MB raw) and later 43,279 chunks. A catch-up reader that was ~1,900 chunks behind could never receive its first byte within its read deadlines, because each reconnect re-materialized the whole stream first. The downstream consumer stalled permanently on an otherwise healthy run, and the stall was silent (an empty read is a legal idle shape).

Suggested fix

Page the historical read the same way getChunks() already does in the same file — keyset pagination on chunk_id, with the initial startIndex skip pushed down as a count-bounded OFFSET, driven by ReadableStream pull() so at most one page is resident and the first byte is immediate. Sketch of what we currently run as a local patch (semantics byte-for-byte preserved, including the uniform skip of the EOF marker row, negative-startIndex resolution via count(*) where eof = false, the live NOTIFY buffer, and ULID-order dedup):

// state hoisted: lastChunkId='', offset=startIndex??0, buffer=[],
// historyDone=false, sqlCursor=null, PAGE_SIZE=64
async pull(controller) {
  if (historyDone) return; // live events flow via the NOTIFY handler
  if (!negativeResolved) { /* offset = max(0, count(eof=false) + offset) */ }
  let rows;
  if (sqlCursor === null) {
    let skip = 0;
    if (offset > 0) {
      const total = /* count(*) for the stream */;
      skip = Math.min(offset, total);
      offset -= skip; // remainder spills to the live buffer (reads past tail)
    }
    rows = await q.orderBy(asc(streams.chunkId)).limit(PAGE_SIZE).offset(skip);
  } else {
    rows = await q
      .where(and(eq(streams.streamId, name), gt(streams.chunkId, sqlCursor)))
      .orderBy(asc(streams.chunkId))
      .limit(PAGE_SIZE);
  }
  for (const row of rows) { enqueue(row); sqlCursor = row.id; }
  if (rows.length < PAGE_SIZE) { historyDone = true; flushLiveBuffer(); }
}

Verified properties of this shape on a live Postgres (we run it in production-like dogfood):

  • byte-exact reads across page boundaries (213-chunk probe, page size 64);
  • startIndex on a page edge / mid-page / at tail / past tail / negative — all byte-identical to the current semantics;
  • duplicate-free, gap-free history→live handoff (30 live appends interleaved into a 70-chunk historical read);
  • getInfo untouched.

Happy to open a PR with the full patch if you'd take it.

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start in dist/streamer.js, comparing the historical read inside get() with the paginated approach already used by getChunks(); keep getInfo untouched. Verify the stated 213-chunk and 70-chunk history-to-live probes, including startIndex edge cases, and consider the work done when reads preserve byte-exact, gap-free semantics while serving the first byte without materializing the full stream.

Written by the indexing model from the issue text.

Assessment

Tech stack
postgresql, typescript
Domain
backend-api-design, databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Clearly specified
Newbie friendliness
58/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.