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

[lunareed720]

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.