Repository navigation
feat(indexer): reconnect dropped WebSockets and resume from a persisted cursor - #14
Merged
Merged
Conversation
…ed cursor alloy's PubSubService reconnects and re-subscribes transparently without surfacing the gap, and SubscriptionStream discards broadcast-lagged notifications with only a debug log, so a live subscription cannot be relied on for completeness. Resuming on stream end alone would only fire once alloy burned its 10 retries, silently losing events on every shorter blip. Restructure the runner around that: subscribe() is a latency optimisation and a periodic reconcile sweep is the correctness guarantee. Each interval the sweep re-scans eth_getLogs from the persisted cursor to the tip, closing gaps from silent reconnects, broadcast lag and process restarts through one code path. Inserts are already idempotent, so the overlap costs nothing. - supervisor.rs rebuilds the provider and runner on any fault, with 1s->30s capped backoff that resets after a 5 minute stable window - chain_cursors persists the per-chain high-water mark; the upsert uses GREATEST so a restarting runner cannot rewind another's progress - resolve_start_block prefers the cursor, then EVM_START_BLOCK, then the tip, replacing the u64::MAX sentinel that silently skipped backfill - fetch_chunk propagates insert failures, so the cursor cannot advance past an event that was never written - EVM_RECONCILE_INTERVAL_SECS is clamped to a non-zero floor, since tokio::time::interval panics on a zero period Cursor writes land only after a chunk fully completes, so tearing the sweep down mid-flight costs at most one chunk of rework. Verified against Postgres 16: migration 0004 applies incrementally on top of 0001-0003, and shared/tests/chain_cursors.rs covers the cursor round trip, monotonicity and the u64 to BIGINT conversion. Those tests are #[ignore]d so CI stays green without a database; run them with `make test-db`. Note: watched_addresses is still empty pending #10, and the sweep re-pulls every log on every chain on a timer until that lands. Closes #2 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Makes a dropped WebSocket lossless: the indexer now reconnects with exponential backoff and resumes from a persisted per-chain high-water mark instead of going dark until the process restarts.
Closes #2.
Why
The issue's premise turned out not to hold, which changed the design.
#2 assumes a dropped connection surfaces as the log stream ending. It usually doesn't. In alloy 2.1
PubSubServicecatches the disconnect itself, reconnects with its own capped exponential backoff (max_retries: 10,retry_interval: 3s) and re-issues every active subscription (alloy-pubsub-2.1.0/src/service.rs:71-110). The broadcast channel survives, sostream.next()never observes the drop — but the logs produced during the gap are gone with no signal.with_max_retries(0)doesn't opt out either:reconnect()is attempted once before the retry count is checked.Two consequences:
SubscriptionStreamswallows broadcastLaggederrors with adebug!and a comment saying "This is OK" (sub.rs:392-395). Sinceprocess()awaits a DB insert inline per matching log, a slow database drops logs with no error.So a supervisor and a high-water mark are both necessary but neither delivers "no events are missed" on its own.
How
The live subscription is a latency optimisation; a periodic reconcile sweep is the correctness guarantee. Both run under
tokio::try_join!inChainRunner::run. Each interval the sweep re-scanseth_getLogsfrom the persisted cursor to the tip, which closes gaps from silent reconnects, broadcast lag and cold restarts through a single code path. Inserts are already idempotent viaON CONFLICT DO NOTHING, so the overlap between the two costs nothing.supervisor.rsrebuilds the provider and runner on any fault. 1s → 30s capped backoff, reset after a 5 minute stable window so a long-lived process doesn't keep waiting the maximum delay after one blip.backoff.rsholds that policy as pure logic, so it is testable without I/O.chain_cursorspersists the per-chain high-water mark. The upsert usesGREATESTso a runner restarting mid-sweep cannot rewind progress another runner already made.resolve_start_blockprefers the cursor, thenEVM_START_BLOCK, then the tip. This retires theu64::MAXsentinel that made backfill a silent no-op by default.Two bugs found while reviewing the above, both fixed here:
EVM_RECONCILE_INTERVAL_SECS=0panicked every chain task.tokio::time::intervalasserts a non-zero period (interval.rs:74). Now clamped to a 1s floor.process()logged and continued, but the cursor then advanced past the unwritten event.fetch_chunknow propagates, so a failed write holds the cursor back and the next sweep retries. Live-path failures still only warn, since the sweep covers them.Cursor writes land only after a chunk fully completes, so tearing the sweep down mid-flight costs at most one chunk of rework — cancellation-safety falls out of the ordering rather than needing explicit handling.
Testing
make cigreen: fmt, clippy-D warnings, 43 unit tests (33 existing + 10 new covering the backoff curve, the stability reset, and the interval clamp).Verified against Postgres 16 in Docker, not just compiled:
0004applies incrementally on top of an existing0001–0003database — the same situation Supabase is in — and is recordedsuccess = t.shared/tests/chain_cursors.rscovers the absent-cursor case, round-trip, monotonicity, and a block number pastu32::MAXto exercise the u64 →BIGINTconversion.GREATEST(...)with a plainEXCLUDED.last_blockmakes it fail withleft: Some(150), right: Some(200), then reverted.Those tests are
#[ignore]d so CI stays green without a database —make ciwithDATABASE_URLunset gives 43 passed, 5 ignored. Run them withdocker compose up -d postgres && make test-db.watched_addressesis stillvec![], so the filter is unconstrained. The reconcile sweep now re-pulls every log on every chain on a timer, which multiplies the cost of that considerably. There's aFIXME(#10)at the construction site. I deliberately did not wirecontracts::all_addresseshere since that is #10's scope, but the two are now coupled in a way they weren't before.Known trade-off
Failing the chunk on insert error means one permanently-unwritable event (say a constraint violation) stalls that chain's cursor: 5 failed sweeps → runner errors → supervisor restarts → same row. That's a loud backoff loop rather than quiet data loss, which I think is the right default for an explorer, but a dead-letter or skip-after-N would fix it properly. Worth a follow-up issue.
Checklist
Always
cargo clippy --workspace --all-targetspasses with no warningscargo test --workspacepasses.env.exampleupdated if new env vars introduced —EVM_START_BLOCK(semantics changed) andEVM_RECONCILE_INTERVAL_SECS(new)New protocol adapter
Database / schema changes
shared/migrations/—0004_create_chain_cursors.sqlsqlx-data.jsonregenerated — N/A. The project uses runtimesqlx::query/query_as, not the compile-time-verified macros, so there is nosqlx-data.json/.sqlxto regenerate. Worth noting separately that CI setsSQLX_OFFLINE: truedespite this, which implies a guarantee that isn't there.CREATE TABLE IF NOT EXISTS. Rollback isDROP TABLE chain_cursors;, after which every chain falls back toEVM_START_BLOCKor the tip.Note:
Cargo.lockmoves becausetokiois added as a dev-dependency ofsharedfor the integration tests — it must stay in the commit or CI's--lockedwill fail.