fix(core): keep a bounded stream alive until the finalized head reaches its range end

[abernatskiy]

Problem

A bounded stream (toBlock set) ends as soon as every requested block has been delivered — with no regard for whether the portal's reported finalized head has reached toBlock. Finalized-only targets (parquet) hold rows back until some batch reports them final; after the last block no such batch ever comes (empty batches are filtered out in portal-source), so await pipeTo(...) resolves while the target silently discards the unfinalized tail.

Concretely: a backfill whose END_BLOCK sits above the current finalized head "succeeds" with its output files ending short of END_BLOCK. This is how an Avalanche transactions backfill shipped a range its completion marker claimed but its parquet files did not cover (~1.7k blocks short — the gap between the network data source's reported finalized head and the range end).

MRE

packages/pipes/src/core/bounded-range-finality.test.ts, first test: one response delivers blocks 1..3 but reports finalized = 1 — the shape a backfill sees when END_BLOCK is above the current finalized head. On main the test fails with:

AssertionError: expected 1 to be 3

i.e. pipeTo resolved and the last finalized head the target ever saw was 1 — a finalized-only target must discard blocks 2..3, with no error anywhere.

Fix

The contract is now: a bounded stream ends only after the consumer has been shown a finalized head ≥ the range end — waiting for finality to catch up, not crashing, not ending short.

• portal-client — after the last block of a bounded range, poll GET /finalized-head until it reaches toBlock, then deliver the catch-up as one final empty batch. No-finality datasets (no finalized head ever reported) and already-final ranges end exactly as before, without a single poll. A bounded range the portal answers with no data is retried on the same cadence instead of ending the range short, and a 200 that advanced nothing no longer re-requests in a hot loop.
• portal-source — forward an otherwise-filtered empty batch when it advances the finalized head over the last delivered block, so the consumer can commit its tail. All other empty batches stay filtered.
• targets (clickhouse / parquet / bigquery / drizzle) — skip the user onData handler for batches built from zero source blocks (user handlers never saw blockless batches before; one indexed data[0] and would crash). Finalization, cursor and offset persistence still run — recording the caught-up finalized head is the point of the batch. The postgres sync-table insert becomes an upsert, since the catch-up re-saves the last block's offset row with newer finality.

The wait is bounded by finality itself: for a range ending above the chain's current finalized head it is minutes. For a range the portal can never finalize (a frozen dataset) the stream now visibly waits instead of silently completing with missing data — the failure mode this bug traded that for.

Tests

• New: bounded-range-finality.test.ts — the MRE (fails on main), plus no-poll fast-paths (already-final, no-finality) and the bounded no-data retry.
• The mock portal now serves GET /finalized-head (default: everything it served is final; scriptable per test).
• Updated existing snapshots/expectations that had frozen the old end-state — including a parquet test that literally asserted 3–5 stay buffered and are never written, i.e. the incident behavior. It now asserts nothing is published above the finalized head while finality lags, and that the tail lands once finality arrives.
• Full suite: 734 passed (with local clickhouse + postgres services).

🤖 Generated with Claude Code

[Copilot]

Pull request overview

This PR changes the bounded-range streaming contract so a stream with toBlock set does not complete until the consumer has observed a finalized head >= toBlock, preventing finalized-only targets (e.g. parquet) from silently dropping an unfinalized tail.

Changes:

• Portal client now polls GET /finalized-head after delivering the last block of a bounded range and emits a final empty “finality catch-up” batch once finality reaches toBlock.
• Portal source forwards only the otherwise-filtered empty batch that advances finality past the last delivered block, and targets skip user onData for these blockless batches while still persisting cursor/finality.
• Adds/updates tests across core, portal client/cache, and targets to validate finality catch-up behavior and bounded no-data retries.

Reviewed changes

Copilot reviewed 17 out of 17 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
packages/pipes/src/testing/test-portal.ts	Extends mock portal to serve GET /finalized-head for finality polling tests.
packages/pipes/src/portal-client/client.ts	Adds end-of-bounded-range finality polling + bounded no-data retry pacing and prevents hot loops.
packages/pipes/src/portal-client/client.test.ts	Updates batching/accounting tests to include the final empty finality catch-up batch and retry pacing.
packages/pipes/src/core/portal-source.ts	Forwards the finality catch-up empty batch to consumers while keeping other empty batches filtered.
packages/pipes/src/core/portal-source.test.ts	Updates expectations for the extra catch-up batch and lifecycle span counts.
packages/pipes/src/core/bounded-range-finality.test.ts	New tests covering bounded finality wait, fast paths, and bounded no-data retry.
packages/pipes/src/targets/parquet/parquet-target.ts	Skips user onData for blockless finality catch-up batches while still committing finality/cursor.
packages/pipes/src/targets/parquet/parquet-target.test.ts	Updates finalized-only parquet behavior to assert tail publication once finality catches up.
packages/pipes/src/targets/parquet/custom-engine.test.ts	Updates custom engine expectations to include tail rows after finality catch-up.
packages/pipes/src/targets/memory/memory-target.test.ts	Updates memory target expectations to include tail row once finality catch-up arrives.
packages/pipes/src/targets/drizzle/node-postgres/postgres-state.ts	Changes sync-table insert to an upsert to support catch-up re-saving the last offset row.
packages/pipes/src/targets/drizzle/node-postgres/drizzle-target.ts	Skips user onData for blockless catch-up batches but still persists cursor/finality.
packages/pipes/src/targets/drizzle/node-postgres/drizzle-target.test.ts	Updates snapshot to reflect finalized head catch-up and empty rollback chain.
packages/pipes/src/targets/clickhouse/clickouse-target.ts	Skips user onData for blockless catch-up batches while still saving cursor/finality.
packages/pipes/src/targets/clickhouse/clickhouse-target.test.ts	Updates snapshots to include the additional saved offset row from the catch-up batch.
packages/pipes/src/targets/bigquery/bigquery-target.ts	Skips user onData for blockless catch-up batches but still WAL-commits finality/cursor.
packages/pipes/src/portal-cache/node/portal-cache.test.ts	Updates cache snapshots to include the finality catch-up batch metadata.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 17 out of 17 changed files in this pull request and generated no new comments.

Suppressed comments (1)

packages/pipes/src/portal-client/client.ts:307

• getFinalizedHead() is awaited without any abort signal, so a bounded stream that is stopped/aborted while waiting for finality can hang until the HTTP request times out. The rest of the stream requests are wired to buffer.signal (via abort: buffer.signal), so this new polling path should be cancellable the same way (e.g., pass an AbortSignal into getFinalizedHead, or otherwise ensure the underlying request is aborted when buffer.signal fires).

        const head = await getFinalizedHead()
        if (head != null && head.number > finalizedSeen) {
          finalizedSeen = head.number
          if (head.number >= toBlock) {
            await buffer.put({