Worker poll

[Qoder-Undefined]

Worker poll pagination uses OFFSET into a live table: concurrent inserts skip or duplicate events

closes #524
Description
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.

[Xhristin3]

LGTM