feat: evmStream facade with multi-source RPC fallback and a pluggable switching strategy

[abernatskiy]

Summary

• One facade for everything: evmStream({source}) now also accepts an ordered source list — portal URLs/options, plain {type: 'rpc', url} JSON-RPC endpoints, or custom clients — and drives them with health-based failover and switch-back. A single source compiles to exactly the previous code path, and evmPortalStream remains a deprecated alias, so nothing breaks.
• portal option renamed to source: the option takes RPC endpoints and fallback lists, not just portals, so the old name no longer described it. The portal spelling is still accepted (deprecated), and the repo — CLI init templates, docs, examples, tests — now uses evmStream + source throughout.
• Fallback under the stream, not beside it: sources plug in through a new BlockStreamClient interface (the exact client surface PortalStream consumes). FallbackClient multiplexes any such clients (chain-agnostic), and EvmRpcBlockClient serves portal-wire-shaped blocks off a JSON-RPC endpoint — so outputs/decoders, multi-range plans, cache guards, progress, metrics, the finalized watermark and fork handling work identically for every source kind. The RPC stack (@subsquid/evm-rpc + peers) stays an optional, lazily-loaded dependency.
• Mixed-finality source lists: a finalized-only Portal can do the cheap bulk backfill and hand off to a hot RPC at the finality frontier (driven by detection.maxStalenessMs — the Portal's request simply sits outstanding there). A set containing any hot source reports itself as hot, so the target keeps fork handling, and each source's head is polled at its own commitment so an exhausted source never looks "fresher" than one that is genuinely ahead.
• Detection vs strategy, each with its own vocabulary: fallback.detection configures how failure/recovery is sensed — capability probes, head polls, liveness thresholds, cooldowns — and defines the freshness conditions, whose verdicts ride on the strategy events (stall.stale, batch.lagging). fallback.strategy is the deciding half: plain options tune the stock strategy ({ preferPrimary, allDownTimeoutMs }), or a function replaces its decisions per select / batch / stall event — it sees the measurements and the stock decision (ctx.defaultCommand), and returning undefined keeps stock behavior. Safety invariants (fork propagation, cursor-continuous boundary-only switches) are not delegated.

Replaces #109 (same engine — health model, probes, staleness/lag detection, diagnostics, metrics — rebased across the package rename and restructured onto the facade).

const stream = evmStream({
  id: 'swaps',
  source: [
    { url: 'https://portal.example/datasets/ethereum-mainnet', name: 'primary' },
    'https://portal.sqd.dev/datasets/ethereum-mainnet',
    { type: 'rpc', url: RPC_URL, rateLimit: 10 },
  ],
  fallback: {
    strategy: (ctx) => {
      // e.g. veto failover to the paid RPC while still backfilling
      const d = ctx.defaultCommand
      if (d?.action === 'use' && ctx.sources[d.index].name === 'rpc-2' && !ctx.atTip) return { action: 'hold' }
      return undefined // otherwise: stock behavior
    },
  },
  outputs,
})

Breaking changes (type-level)

StartContext.portal (transformer start hooks) and EvmFactory.runPreindex's portal are typed
BlockStreamClient instead of PortalClient, because a pipe's source may now be an RPC-backed or
fallback client rather than a portal. Runtime behaviour is unchanged — a single portal URL still
passes a real PortalClient. A hook needing portal-specific request options should narrow with
instanceof PortalClient.

Coverage

Module	Stmts	Δ	Branch	Δ
All files	88.6%	+0.8	88.4%	+0.4
src/core	90.9%	+3.7	91.9%	+1.4
src/evm	85.4%	-5.1	89.1%	-3.6
src/evm/rpc (new)	97.2%	+97.2	85.0%	+85.0
src/http-client	63.0%	+0.0	69.6%	+0.3
src/internal	93.2%	+0.4	84.9%	+0.6
src/portal-client	94.2%	+1.5	94.8%	-0.4
src/testing	92.6%	+2.9	93.2%	+1.5

evm-rpc-block-client.ts itself sits at ~52% statements: the uncovered remainder is the EvmRpcDataSource stream orchestration + fork translation, which requires a live endpoint and is covered by the network-gated e2e suites below.

Test plan

• 1105 unit tests green (full suite incl. ClickHouse/Postgres targets), among them:
  • 38 FallbackClient supervisor tests (failover, cursor-continuous resume, fork propagation across switches, staleness/lag/chain-stall, probe gating, all-down, gauges)
  • 7 pure defaultFallbackStrategy tests + custom-strategy tests (pinning, voluntary jumps, stall holds, defaultCommand inspect/veto, per-event fallthrough, invalid index)
  • wire-mapper unit tests on a hand-built raw block (filtering via decoded copy, relation inclusion, downstream cast decode+prune parity)
  • facade tests over mock portals (mid-range failover with resume-anchor validation, custom strategy routing, cache/source-list and single-source fallback config errors)
• Network-gated live e2e (RPC_E2E=1): RPC-vs-Portal wire parity at a historical block (equal decode, projection of where-only fields), portal→portal failover, the full facade streaming typed blocks from an RPC-only source list, and head semantics (a finalized-head batch never reports latest)

Review-round fixes (each with a regression test)

• isBlockStreamClient now verifies the whole contract, so a partial object is rejected at classification instead of failing mid-stream
• EvmRpcBlockClient rejects a non-EVM query by name
• head.latest is no longer derived from the finalized head — that reported an unbounded run complete a finality window early
• the pre-filter cast decodes only what filtering reads, so selecting receipt-backed transaction fields alongside a logs-only request no longer throws
• lag/chain-head gauges stay unset until a block has been delivered (an empty first batch previously published a chain-height lag to metrics and ctx.lagBlocks)
• mixed-finality source lists are allowed (see above), with regression tests for the handoff, the conservative finality flag, and the per-source head commitment
• staleness measures a source's unproductive wait, so a source that answers without progressing is caught and a slow consumer never fails a healthy source over
• reclaim is gated on why a source was left, so an exhausted source is not crawled back into and a recovered one is not permanently demoted
• fallback metrics carry the pipe id, so several pipes sharing a metrics server stay individually observable; lag is absent rather than zero when it cannot be computed
• the optional RPC peers stay out of the public type graph (guarded by a structural test), sources in a list get a bounded retry budget, and a lazily loaded RPC source is constructed once under concurrent use

Specification

spec/16-fallback.md specifies this machinery in the style of the existing suite — model, driving rules, safety invariants, liveness, failure model and observability — with four decisions recorded as ADRs (detection/strategy split, conservative finality, staleness as unproductive wait, reclaim gating) and every threshold registered as a parameter. node spec/check-spec.mjs gates it in CI.

🤖 Generated with Claude Code

https://claude.ai/code/session_017Vfwn6cqjZpV4pF4J27sRd

[Copilot]

Pull request overview

This PR introduces a unified evmStream facade that can stream EVM data from a single Portal source, a JSON-RPC source, or an ordered multi-source list driven by a health-based fallback supervisor, while keeping Portal-wire compatibility so the existing decoding/transform pipeline remains unchanged.

Changes:

• Added a generic BlockStreamClient contract, a FallbackClient supervisor (detection + strategy), and Prometheus-style fallback metrics.
• Added an RPC-backed EVM block client plus RPC→Portal-wire mapping/filtering utilities, with optional/lazy loading of the RPC stack.
• Refactored EVM entrypoints/tests to use evmStream and keep evmPortalStream as a deprecated alias.

Reviewed changes

Copilot reviewed 55 out of 56 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
pnpm-lock.yaml	Locks newly added deps for RPC + fallback support.
packages/pipes/package.json	Adds optional peer deps for the RPC stack + internal utilities used by the new modules.
packages/pipes/src/testing/test-metrics-server.ts	Extends the mock Gauge with reset() to support stale-series pruning in metrics collect.
packages/pipes/src/testing/test-block-stream-client.ts	New in-memory BlockStreamClient mock used by fallback/probe tests.
packages/pipes/src/testing/index.ts	Re-exports the new mockBlockStreamClient helper.
packages/pipes/src/testing/evm/evm-portal-mock-stream.test.ts	Updates imports to use evmPortalStream from evm-stream.
packages/pipes/src/targets/pubsub/pubsub-target.test.ts	Updates EVM source import to the new facade module.
packages/pipes/src/targets/pubsub/pubsub-target-fork.test.ts	Updates EVM source import to the new facade module.
packages/pipes/src/targets/pubsub/pubsub-signal.test.ts	Updates EVM source import to the new facade module.
packages/pipes/src/targets/memory/memory-target.test.ts	Updates EVM source import to the new facade module.
packages/pipes/src/targets/bigquery/bigquery-target-fork.integration.test.ts	Updates EVM source import to the new facade module.
packages/pipes/src/portal-client/client.ts	Introduces BlockStreamClient + isBlockStreamClient for plugging non-Portal sources into PortalStream.
packages/pipes/src/evm/rpc/wire.ts	Adds RPC block → Portal-wire mapping with client-side filtering integration.
packages/pipes/src/evm/rpc/wire.test.ts	Unit tests for wire mapping + downstream cast/prune parity expectations.
packages/pipes/src/evm/rpc/shim.ts	Shim for trace-tag/action-field discrepancies between normalization output and Portal schema.
packages/pipes/src/evm/rpc/shim.test.ts	Tests for shim behavior and malformed input tolerance.
packages/pipes/src/evm/rpc/request.ts	Derives coarse RPC fetch toggles (logs vs receipts, traces, stateDiffs).
packages/pipes/src/evm/rpc/project.ts	Field augmentation for where-only keys + empty-block dropping parity logic.
packages/pipes/src/evm/rpc/project.test.ts	Tests for field augmentation and empty-block dropping behavior.
packages/pipes/src/evm/rpc/filter.ts	Client-side filtering + relation expansion to mirror Portal semantics for RPC blocks.
packages/pipes/src/evm/rpc/filter.test.ts	Tests for filter semantics, relations, and required-data derivation scenarios.
packages/pipes/src/evm/rpc/decode.ts	Required-field forcing and Portal decoder reuse for RPC-origin blocks.
packages/pipes/src/evm/rpc/decode.test.ts	Tests for required-field forcing behavior (cursor/filter discriminators).
packages/pipes/src/evm/factory.ts	Broadens preindex input type to BlockStreamClient and switches to new EVM stream import.
packages/pipes/src/evm/factory.test.ts	Updates EVM stream import to evm-stream.
packages/pipes/src/evm/evm-stream.ts	New evmStream facade + fallback-source-list support; keeps deprecated aliases.
packages/pipes/src/evm/evm-stream.test.ts	New unit tests covering facade defaults, fallback list behavior, strategy routing, and config validation.
packages/pipes/src/evm/evm-rpc-block-client.ts	RPC-backed BlockStreamClient implementation that yields Portal-wire-shaped blocks.
packages/pipes/src/evm/evm-rpc-block-client.test.ts	Network-free surface tests for RPC client construction/metadata/errors.
packages/pipes/src/evm/evm-rpc-block-client.parity.e2e.test.ts	Network-gated live parity tests (Portal vs RPC) using the same downstream cast.
packages/pipes/src/evm/evm-portal-source.test.ts	Removes the old portal-only stream test (replaced by evm-stream.test.ts).
packages/pipes/src/evm/evm-fallback.ts	Builds a fallback client from Portal/RPC/custom specs + lazy RPC loading & peer diagnostics.
packages/pipes/src/evm/evm-fallback.test.ts	Tests for barrel reachability and missing-peer translation + spec parsing.
packages/pipes/src/evm/evm-fallback.e2e.test.ts	Network-gated e2e tests for mixed Portal/RPC fallback and RPC-only facade streaming.
packages/pipes/src/evm/evm-decoder.test.ts	Updates EVM stream import to evm-stream.
packages/pipes/src/evm/browser.ts	Updates public EVM barrel exports to include fallback + new stream facade.
packages/pipes/src/evm/abi/define-abi.test.ts	Updates EVM stream import to evm-stream.
packages/pipes/src/core/transformer.ts	Generalizes transformer hook context from PortalClient to BlockStreamClient.
packages/pipes/src/core/portal-source.ts	Allows PortalStream to consume any BlockStreamClient; validates cache compatibility.
packages/pipes/src/core/metrics-server.ts	Extends Gauge interface with optional reset?() for collect-time stale-series pruning.
packages/pipes/src/core/index.ts	Exposes new fallback modules from the core barrel.
packages/pipes/src/core/fallback-strategy.ts	Adds fallback strategy types + default strategy implementation.
packages/pipes/src/core/fallback-strategy.test.ts	Unit tests for default strategy decisions and composition patterns.
packages/pipes/src/core/fallback-metrics.ts	Registers pull-based gauges exporting fallback state.
packages/pipes/src/core/fallback-metrics.test.ts	Tests for fallback metrics collection behavior and label handling.
packages/pipes/src/core/fallback-health.ts	Implements detection options + the per-source trinary health state machine.
packages/pipes/src/core/fallback-health.test.ts	Tests for health transitions, thresholds, cooldowns, and probe gating.
packages/pipes/src/core/fallback-diagnostics.ts	Error classification + URL redaction for logs/metrics-safe reporting.
packages/pipes/src/core/fallback-diagnostics.test.ts	Tests for URL/text redaction and error classification behavior.
packages/pipes/src/core/fallback-client.ts	Implements the multi-source FallbackClient supervisor (switching, probes, freshness ticks).
packages/pipes/src/core/fallback-capability.ts	Generic capability probe builder for BlockStreamClient sources.
packages/pipes/src/core/fallback-capability.test.ts	Tests for probe anchoring, fork tolerance, classification, and timeouts.
packages/pipes/src/core/fallback-async.ts	Async utilities (withTimeout, safeReturn, delay) used by fallback engine.
packages/pipes/src/core/fallback-async.test.ts	Tests promise hygiene (timeouts and iterator return).
packages/pipes/src/core/errors.ts	Updates example error message text to reference evmStream.

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[abernatskiy]

(With Claude): Agreed — tightened in 9302a46. isBlockStreamClient now checks all six members (finalized, getUrl, getMetadata, getHead, resolveTimestamp, getStream), so a partial object is rejected where it is classified rather than failing deep inside a stream. Covered by a new test that deletes each member in turn.

[abernatskiy]

(With Claude): Added in 9302a46 — #stream now throws EvmRpcBlockClient can only serve EVM queries, got type "<type>" before destructuring, with a test. The client implements the chain-generic contract but only speaks EVM, so naming that explicitly beats a downstream mapping error.

[abernatskiy]

(With Claude): Good catch — fixed in 9302a46. The mock now returns BlockRef (hash required) and forwards the { finalized } option, matching the contract; the test helpers were retyped accordingly. BlockCursor.hash being optional was exactly the looseness that could have hidden a missing head hash.

[abernatskiy]

(With Claude): Considered and kept, with the tradeoff now documented in the code. The serialize costs a few ms per batch while the RPC fetch it accompanies takes seconds, so it is noise against the work it measures — and setting it to 0 would make the progress/throughput display read as broken, which is what this field feeds. On the UTF-16 point: the portal wire JSON here is ASCII (hex strings and numbers), so length equals byte length; the comment now says so. If profiling ever shows it mattering, swapping in a cheaper estimate is a one-line change.

[Copilot]

Pull request overview

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

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

Suppressed comments (1)

packages/pipes/src/evm/evm-rpc-block-client.ts:161

• PortalHead.latest is intended to represent the chain “latest” head number (PortalClient populates it from X-Sqd-Head-Number), but here it is derived from finalizedHead (a finalized head). That can make progress/lag calculations treat the finalized head as the latest head for RPC sources, which is semantically incorrect and may under-report head/lag.

If the RPC stream cannot provide the latest head cheaply, it’s better to omit latest than to populate it with a finalized value.

        const head = finalizedHead ? { number: finalizedHead.number, hash: finalizedHead.hash } : undefined

        yield {
          blocks: data,
          head: { finalized: head, latest: head ? { number: head.number } : undefined },
          meta: {

[abernatskiy]

(With Claude): Picking up the suppressed finding from the latest review — it was right, and fixing it surfaced a second bug. Both are in 7809529.

1. head.latest must not be the finalized head. latest is the chain head (the portal fills it from its head header) and downstream it is the progress denominator, so deriving it from finalizedHead reported an unbounded run as complete roughly a finality window early. The RPC stream offers no cheap latest head, so the field is now simply absent — exactly what a portal response without a head header looks like. Freshness and lag are unaffected: they come from the independently polled getHead().

2. (found while fixing 1) A valid query could throw. Receipts are fetched only when transactions are requested, but the pre-filter cast decoded the user's output selection — so fields: { transaction: { gasUsed } } with a logs-only request (a shared field preset plus addLogRequest, entirely realistic) blew up with invalid value at /transactions/0/gasUsed. The internal cast now decodes only what filtering actually reads (structural relation fields plus where-clause fields); the wire block still carries every field, and the downstream normalize/cast prunes to the user's selection as before. Regression test added, and it does strictly less decoding work than before.

Verified: 1069 unit tests green, plus the network-gated e2e suites — including a new assertion that a finalized-head batch never reports latest, and the RPC-vs-Portal parity tests still match exactly.

[Copilot]

Pull request overview

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

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

Suppressed comments (1)

packages/pipes/src/core/fallback-client.ts:393

• #observeBoundary uses cursor?.number ?? -1 as a sentinel when no blocks have been delivered yet. But a BlockStreamClient can legitimately yield empty batches (e.g. PortalClient on HTTP 204 yields blocks: []), so cursor can be undefined at a batch boundary. In that case this code reports chainHead = -1 (and can report a large positive lag when other sources have heads), which is a bogus value for observability and can confuse strategies that read ctx.lagBlocks/metrics before the first real block arrives.

    const lastNumber = cursor?.number ?? -1
    this.chainHead = others != null ? Math.max(others, lastNumber) : lastNumber
    if (others == null) {
      this.lag = 0 // no independent reference ⇒ lag is not computable; don't report a stale value
    } else {

[abernatskiy]

(With Claude): The suppressed finding in the latest review was right too — fixed in 489a656.

#observeBoundary measured from a -1 sentinel when no block had been delivered yet, which is reachable: a source can yield an empty batch (PortalClient answers HTTP 204 exactly that way), and the boundary runs after every batch. With a standby head at N, that published lag = N + 1 on the sqd_fallback_lag_blocks gauge and in a custom strategy's ctx.lagBlocks, and reported chainHead = -1 when no other source had a head.

Scope worth stating precisely: failover was never affected. Arming the tip latch requires lag <= maxLagBlocks, so a chain-height lag can never arm it, and the lagging verdict requires the latch. This was an observability defect only — but a misleading one, and a custom strategy reading ctx.lagBlocks could have acted on it.

Now both gauges stay unset until there is a position to measure from, and chainHead reports only what is actually known. The regression test asserts the discriminating value directly: it reads lag: 1001 against the previous code and 0 against the fix.

1070 unit tests green.

[Copilot]

Pull request overview

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

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

[abernatskiy]

(With Claude): Real race, fixed in 2acd29f.

The lazy loader guarded on the resolved client, so every caller arriving before the first load finished started its own — and the two callers that race here are not exotic: a head poll and a capability probe overlap routinely, since probes are fire-and-forget and head polls run at every batch boundary. The result was several EvmRpcBlockClient instances for one configured endpoint, and therefore several connection pools and rate limiters, which for a metered endpoint is precisely the opposite of what capacity/rateLimit asked for.

Rather than inline the guard, I moved the sharing into a small internal helper so the mechanism itself is testable, and pinned two properties:

• concurrent callers share one run (this fails against the previous value-guard shape — verified by restoring it);
• a failed run is not cached, so one transient failure cannot poison the source for the life of the process (a naive pending ??= fn() fails this — also verified).

Worth noting for the missing-peer path specifically: that failure is deterministic, so not caching it just means the actionable "install these peers" error is raised again on the next attempt rather than a stale rejection being replayed.

1105 unit tests, live e2e and the full build green.

[Copilot]

Pull request overview

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

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

[Copilot]

Pull request overview

Copilot reviewed 112 out of 115 changed files in this pull request and generated 3 comments.

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

[abernatskiy]

(With Claude): Reworded — the blanket claim is now scoped to the removed *PortalSource names, and the next paragraph already spells out the EVM exception (evmPortalStream deprecated alias, portal option spelling deprecated in favor of source). Fixed in 18aab60.

[abernatskiy]

(With Claude): Agreed — the Removals section now states the exception explicitly: evmPortalStream remains as a deprecated alias of evmStream, and the portal option spelling stays accepted (deprecated in favor of source), since that rename landed after the 1.0 betas shipped. Fixed in 18aab60.

[Copilot]

Pull request overview

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

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

Suppressed comments (1)

Previously missed (1) — in code that hasn't changed since the last review.

packages/pipes/src/core/fallback-metrics.ts:127

• The sqd_fallback_staleness_ms help text says it measures how long a request has been outstanding, but FallbackClient.staleness is defined/updated as unproductive wait (accumulated time answering without delivering a block, excluding consumer time). The metric description should match the implemented semantics so dashboards/alerts interpret it correctly.

[abernatskiy]

(With Claude): Addressed the suppressed finding from the latest Copilot round: the sqd_fallback_staleness_ms help text now states the implemented ADR-25 semantics (accumulated unproductive wait, excluding consumer time) instead of a per-request outstanding clock, and the FallbackMetrics.staleness field got a matching doc comment. Fixed in b844d76.

[Copilot]

Pull request overview

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

Files not reviewed (1)

• pnpm-lock.yaml: Generated file

[abernatskiy]

(With Claude): d452dea addresses a beta-tester report: at tip pace a healthy pipe issued one eth_getBlockByNumber per block per standby. Two fixes: the batch-boundary head poll now runs only when something consumes it (lag detection, a stale verdict, a custom strategy, or a reclaimable more-preferred source), and head polls prefer a new optional number-only client method — eth_blockNumber on RPC sources — over the full block-reference lookup. Spec WP-63/WP-66 updated to match. Sibling change for the Squid SDK: subsquid/squid-sdk#561 (that fallback already had the gate; only the call cost needed fixing).