[world-postgres] streams.get(name, startIndex) buffers the entire stream before serving the first byte
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);
startIndexon 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);
getInfountouched.
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
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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