feat(backend): checkpointed, reorg-aware ledger ingestion pipeline - #392
Open
ndii-dev wants to merge 1 commit into
Open
feat(backend): checkpointed, reorg-aware ledger ingestion pipeline#392ndii-dev wants to merge 1 commit into
ndii-dev wants to merge 1 commit into
Conversation
Adds backend/src/ingestion/, a pipeline that pulls ledger data and writes derived corridor/anchor/network rows in a way that survives being re-run over the same range twice, and survives a Horizon/RPC reorg. - watermark.rs: checkpoint keyed on (ledger_sequence, ledger_hash), not sequence alone, so a reorg is detectable directly as a hash mismatch on the next fetch rather than inferred from a gap. - upsert.rs: every derived write is a full replace keyed on (ledger_sequence, entity_id) — re-ingesting a range, or rolling one back after a reorg, converges on the same state instead of accumulating duplicates or leaving orphaned rows from an abandoned fork. - fetch.rs: a real Horizon-backed LedgerSource, plus a deterministic FakeLedgerSource whose fork_from() builds a genuinely diverging hash chain for tests, not a canned response. - mod.rs: IngestionPipeline orchestrates fetch -> reorg-check -> upsert -> checkpoint. On a hash mismatch it rolls the watermark back by a confirmation depth, invalidates every derived row from there forward, and resumes ingestion — self-healing with no manual intervention. run_forever logs and continues past a failed tick rather than dying permanently on the first transient error. - reconcile.rs: a scheduled check that recomputes a checksum straight from raw ledger data and compares it against what the derived store actually has, independent of the ingestion path itself. Verified against a synthetic reorg fixture (tests/ingestion_reorg_test.rs) and 27 unit tests covering idempotent upsert, checkpoint rollback depth, the deeper-than-start-sequence reset path, and checksum drift detection. Closes Stellar-Analysis#319
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.
Summary
Closes #319.
This repo had no ingestion pipeline for Stellar ledger data yet. Adds
backend/src/ingestion/, which pulls ledger data and writes derived corridor/anchor/network rows (RawSnapshotRow, already used bybackend/src/snapshot) in a way that:Design
watermark.rs— the checkpoint is(ledger_sequence, ledger_hash), not sequence alone, so a reorg is detectable directly: the next fetched ledger'sprev_hashstops matching the checkpoint's hash.WatermarkStoreis a trait with anInMemoryWatermarkStorereference implementation, matching this codebase's existing pattern (AnalyticsSink/InMemoryAnalyticsSink, the reconciliation module's stores) rather than introducing a new database dependency this crate doesn't otherwise have.upsert.rs— every derived write is a full replace keyed on(ledger_sequence, entity_id)whereentity_id = (source, corridor). Re-ingesting a range, or rolling one back after a reorg and re-ingesting it, converges on the same stored state instead of accumulating duplicates or leaving orphaned rows behind from an abandoned fork.fetch.rs—LedgerSourcetrait with two implementations:HorizonLedgerSource: a real Horizon-backed wrapper (viareqwest), reusingnetwork::NetworkClientfor its URL config. It derives one coarse, ledger-levelRawSnapshotRowfrom each ledger's transaction success/failure counts — intentionally a proxy, not the full per-corridor asset-pair breakdown a real analytics engine would compute from payment operations (that computation doesn't exist in this codebase yet — see Core payment-reliability/latency-percentile computation engine does not exist #383). This wrapper's job is to prove the checkpointing/reorg machinery works end-to-end with a real row flowing through it.FakeLedgerSource: deterministic, in-memory, for tests.fork_from()builds a chain whose hash linkage genuinely diverges from the original at the fork point (not a canned response), so the pipeline's reorg detection is actually exercised, not simulated.mod.rs—IngestionPipelineorchestrates fetch → reorg-check → upsert → checkpoint. On a hash mismatch it rolls the watermark back by a confirmation depth (default N=3), invalidates every derived row from there forward, and resumes — no manual intervention needed. If the reorg is deeper than the pipeline's configured start ledger, it resets ingestion entirely rather than getting stuck.run_foreverlogs and retries past a failed tick instead of dying permanently on the first transient error (worth calling out since ReconciliationJob::run_forever dies permanently on the first transient error #380, open in this same repo, is exactly that anti-pattern in the reconciliation job — this pipeline deliberately doesn't repeat it).reconcile.rs—ReconciliationCheckruns on a schedule, recomputing a checksum straight from raw ledger data (refetched fromLedgerSource) and comparing it against what's actually stored, so a bug in the ingestion write path itself — not just an outage — gets caught. The checksum combines rows with wrapping addition rather than XOR, deliberately: XOR would let two occurrences of the same row cancel out, silently hiding exactly the double-counting bug this check exists to catch.Testing
cargo fmt/cargo clippy(including the strict--libflags this repo's CI uses:-D clippy::unwrap_used -D clippy::expect_used -D clippy::panic -D warnings) are clean for every file this PR touches. (Note:cargo clippy --libcurrently fails on unmodifiedmainwith ~51 pre-existing errors in unrelated files — verified viagit stashbefore this diff existed — and two existing integration tests,connected_trace_test.rsandmixed_version_replay_test.rs, already fail to compile onmaindue to a pre-existingsoroban-sdk/stellar-xdrversion conflict. None of that is from this PR; out of scope to fix here.)backend/tests/ingestion_reorg_test.rs— the acceptance test: replays a 7-ledger fixture where a reorg at ledger 5 replaces two ledgers with a 3-ledger competing fork, and asserts the final state matches only the canonical chain with zero orphaned or duplicated rows, using nothing but the pipeline's normalrun_once()poll loop.mainstill pass unmodified.Test plan
cargo build --libcargo clippy(strict flags) clean on every file this PR touchescargo fmt --checkclean on every file this PR touchesingestion_reorg_test.rs(the reorg fixture + byte-identical re-run acceptance tests) passcargo test --lib(54 tests) and other working integration test files still pass — no regressions