Problem
The worker drains stream_data with OFFSET-based pagination against a live table, so events inserted while a poll is in progress are skipped or duplicated. getPendingEvents() (api/src/streams/repository/streams-db.repository.ts) is:
SELECT stream_id, data, timestamp
FROM stream_data
ORDER BY timestamp ASC
LIMIT $1 OFFSET $2
and the worker advances cursor by batch size until nextCursor is null (xstreamroll-processing/src/worker.ts pollOnce). Two failure modes follow directly:
- Skip: events are ordered by
timestamp, which is DEFAULT NOW() — identical timestamps for a burst, with no tiebreaker. As the worker pages OFFSET 0 → 100 → 200, a concurrent insert at the front of the result set shifts every row's offset, so rows that were between the pages are never seen in this pass.
- Duplicate/never-drained: nothing marks a row as claimed or processed. Even after the missing
POST /streams/processed endpoint lands (the drain contract), the poll itself gives no stable identity: PendingStreamEvent has only streamId, data, timestamp — no row id — so the processed-event callback cannot reference which stream_data row it drained. Two workers polling with the memory lock backend fetch overlapping batches, and the same row is delivered to the worker repeatedly until drained.
The instability compounds with latency: the poll returns rows ordered by insertion time, but a slow batch means new arrivals can be older-timestamped than the current cursor position (clock skew across the app and the DB, or retried inserts), permanently hiding them behind the cursor.
Root cause
// api/src/streams/repository/streams-db.repository.ts — getPendingEvents()
`SELECT stream_id, data, timestamp
FROM stream_data
ORDER BY timestamp ASC
LIMIT $1 OFFSET $2` // ← OFFSET into a table with concurrent inserts
// xstreamroll-processing/src/worker.ts
let cursor = 0
...
const response = await axiosInstance.get(`${API_URL}/streams/pending?limit=...&cursor=${cursor}`)Why this is architecturally hard
- The cursor is an offset, not a position. The correct replacement is keyset pagination over a unique, stable key —
(timestamp, id) with a composite index on stream_data(timestamp, id) — and the nextCursor semantics the worker already understands (null when the batch is short) can be preserved. But the wire contract of GET /streams/pending ({ data, nextCursor }) is consumed by the worker, and the SDK's pagination helper (xstreamroll-sdk/src/pagination.ts) is page-based; changing the cursor shape is an API/worker contract change that needs coordination with the POST /streams/processed work, which defines the drain.
- Exposing the row
id in PendingStreamEvent is the enabling step for exact drain (DELETE FROM stream_data WHERE id = $1 in the processed-event transaction). Without it, "processed" can only be matched heuristically (e.g. by (stream_id, timestamp, data)), which breaks under duplicate payloads.
- Ordering by
timestamp alone is non-deterministic for equal timestamps; the keyset must include the id tiebreaker, and the existing index idx_stream_data_timestamp must be extended to (timestamp, id) or the query plan regresses to a sort.
- This interacts with the at-least-once delivery model: the distributed lock (
xstreamroll-processing/src/leader-election.ts) prevents concurrent processing of the same stream, but it does not make polling consistent. The pagination fix and the drain semantics (single transaction, idempotency key) must land together or the pipeline regresses to duplicates.
Acceptance criteria
Contract
Behaviour
Tests
Out of scope
The POST /streams/processed drain endpoint itself (separate issue — but this issue's contract change should be coordinated with it), and per-claim leases to prevent two workers processing the same stream.
Getting started
Real files in scope: api/src/streams/repository/streams-db.repository.ts (getPendingEvents), api/src/streams/streams.controller.ts (GET /streams/pending), api/src/streams/repository/streams.repository.ts (in-memory getPendingEvents), xstreamroll-processing/src/worker.ts (pollOnce), database/schema.sql (stream_data indexes), api/src/database.integration.spec.ts.
Verify with:
cd api && npm run typecheck && npm test
cd ../xstreamroll-processing && npm run typecheck && npm test
Good first files to read: api/src/streams/repository/streams-db.repository.ts (getPendingEvents), xstreamroll-processing/src/worker.ts (pollOnce), database/schema.sql (the idx_stream_data_timestamp index).
Problem
The worker drains
stream_datawith OFFSET-based pagination against a live table, so events inserted while a poll is in progress are skipped or duplicated.getPendingEvents()(api/src/streams/repository/streams-db.repository.ts) is:and the worker advances
cursorby batch size untilnextCursoris null (xstreamroll-processing/src/worker.tspollOnce). Two failure modes follow directly:timestamp, which isDEFAULT NOW()— identical timestamps for a burst, with no tiebreaker. As the worker pagesOFFSET 0 → 100 → 200, a concurrent insert at the front of the result set shifts every row's offset, so rows that were between the pages are never seen in this pass.POST /streams/processedendpoint lands (the drain contract), the poll itself gives no stable identity:PendingStreamEventhas onlystreamId,data,timestamp— no row id — so the processed-event callback cannot reference whichstream_datarow it drained. Two workers polling with the memory lock backend fetch overlapping batches, and the same row is delivered to the worker repeatedly until drained.The instability compounds with latency: the poll returns rows ordered by insertion time, but a slow batch means new arrivals can be older-timestamped than the current cursor position (clock skew across the app and the DB, or retried inserts), permanently hiding them behind the cursor.
Root cause
Why this is architecturally hard
(timestamp, id)with a composite index onstream_data(timestamp, id)— and thenextCursorsemantics the worker already understands (nullwhen the batch is short) can be preserved. But the wire contract ofGET /streams/pending({ data, nextCursor }) is consumed by the worker, and the SDK's pagination helper (xstreamroll-sdk/src/pagination.ts) is page-based; changing the cursor shape is an API/worker contract change that needs coordination with thePOST /streams/processedwork, which defines the drain.idinPendingStreamEventis the enabling step for exact drain (DELETE FROM stream_data WHERE id = $1in the processed-event transaction). Without it, "processed" can only be matched heuristically (e.g. by(stream_id, timestamp, data)), which breaks under duplicate payloads.timestampalone is non-deterministic for equal timestamps; the keyset must include theidtiebreaker, and the existing indexidx_stream_data_timestampmust be extended to(timestamp, id)or the query plan regresses to a sort.xstreamroll-processing/src/leader-election.ts) prevents concurrent processing of the same stream, but it does not make polling consistent. The pagination fix and the drain semantics (single transaction, idempotency key) must land together or the pipeline regresses to duplicates.Acceptance criteria
Contract
PendingStreamEventincludes a stable row identifier (e.g.id), andGET /streams/pendingpagination is keyset-based (ORDER BY (timestamp, id)), withnextCursorcarrying the last-seen position andnullat the end.pollOnce(xstreamroll-processing/src/worker.ts) drives the new cursor without re-fetching or skipping rows, and the worker integration tests (xstreamroll-processing/__tests__/integration/worker.integration.test.ts) exercise the new response shape.Behaviour
timestamprows (a burst insert) are delivered deterministically — no skips and no reordering between pages.Tests
api/src/database.integration.spec.ts) inserts N rows in a burst and asserts a full poll walk (repeatnextCursoruntil null) yields every row exactly once, including rows inserted between page fetches.PendingStreamEventshape change.Out of scope
The
POST /streams/processeddrain endpoint itself (separate issue — but this issue's contract change should be coordinated with it), and per-claim leases to prevent two workers processing the same stream.Getting started
Real files in scope:
api/src/streams/repository/streams-db.repository.ts(getPendingEvents),api/src/streams/streams.controller.ts(GET /streams/pending),api/src/streams/repository/streams.repository.ts(in-memorygetPendingEvents),xstreamroll-processing/src/worker.ts(pollOnce),database/schema.sql(stream_dataindexes),api/src/database.integration.spec.ts.Verify with:
Good first files to read:
api/src/streams/repository/streams-db.repository.ts(getPendingEvents),xstreamroll-processing/src/worker.ts(pollOnce),database/schema.sql(theidx_stream_data_timestampindex).