ingest/ledgerbackend: add LedgerStream streaming ingestion API - #5944
Conversation
BufferedStorageBackend decompressed every batch into a freshly-allocated buffer: DecodeAll's destination was pre-sized only from the frame's FrameContentSize, which streaming-compressed objects omit, so dst stayed nil and DecodeAll grew the whole output each call. Under parallel multi-chunk ingest this dominated allocation. Pre-size the destination from the previous batch's decompressed size when the frame carries no FCS (batches are similar-sized) so DecodeAll appends into a pooled buffer; return the original buffer to the pool when DecodeAll has to reallocate. Also set WithDecoderConcurrency(1): decompress() is driven serially by the consumer, so a single decode/buffer combo is both correct and the most memory-efficient choice (concurrency 0 = GOMAXPROCS kept GOMAXPROCS decoder combos and allocated per-call decode state). Co-Authored-By: Claude Opus 4.7 <[email protected]>
Introduce a streaming-first, interchangeable ingestion source interface:
type LedgerStream interface {
RawLedgers(ctx context.Context, r Range) iter.Seq2[[]byte, error]
}
Each implementation owns its whole lifecycle: it builds its backend and
prepares the range when iteration begins, and tears everything down when it
ends (completion, break, error, or ctx cancellation), logging close errors.
Each yielded slice is a borrow valid only until the next iteration step.
Constructors for all three backends:
- NewBufferedStorageStream(cfg, dsConfig, log) (GCS/S3 datastore)
- NewCaptiveCoreStream(config, log) (captive stellar-core)
- NewRPCStream(options, log) (RPC server)
This subsumes the PrepareRange + per-ledger read + Close sequence for
streaming consumers and avoids the per-ledger copy: each stream yields the
backend's internal raw frame directly via a lock-free read — a single-owner
stream has no concurrent access to guard, and a blocked read is cancelled via
ctx. Each backend has an injectable factory seam for unit testing.
Removes the unreleased single-item GetLedgerRaw method — streaming replaces
its only use; GetLedger (decoded random access) is unchanged.
Co-Authored-By: Claude Opus 4.7 <[email protected]>
There was a problem hiding this comment.
Pull request overview
This PR introduces a new streaming-first ingestion abstraction (LedgerStream) with implementations for buffered storage (GCS/S3 datastore), captive stellar-core, and RPC, and removes the unreleased single-ledger raw API (GetLedgerRaw) in favor of streaming. It also optimizes BufferedStorageBackend zstd decompression by reusing output buffers and forcing single-threaded decoder state to reduce allocations.
Changes:
- Added
LedgerStream(RawLedgers) API plus buffered-storage / captive-core / RPC stream implementations with shared lifecycle skeleton. - Removed
GetLedgerRawfromLedgerBackendand updated all backends, mocks, tests, and benches accordingly. - Improved BufferedStorageBackend decompression performance by pooling output buffers and setting
zstd.WithDecoderConcurrency(1).
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| ingest/loadtest/ledger_backend.go | Drops loadtest backend’s GetLedgerRaw implementation after interface removal. |
| ingest/loadtest/ledger_backend_test.go | Updates loadtest mock to match the new LedgerBackend interface. |
| ingest/ledgerbackend/ledger_backend.go | Removes GetLedgerRaw from the LedgerBackend interface. |
| ingest/ledgerbackend/ledger_stream.go | Adds LedgerStream interface, shared streamRaw skeleton, and stream impls for buffered storage / captive core / RPC. |
| ingest/ledgerbackend/buffered_storage_backend.go | Removes GetLedgerRaw and updates comments/error messages; adjusts locking narrative. |
| ingest/ledgerbackend/ledger_buffer.go | Adds decompression output buffer reuse via lastDecompressedSize; forces zstd decoder concurrency to 1. |
| ingest/ledgerbackend/buffered_storage_backend_test.go | Replaces GetLedgerRaw test coverage with BufferedStorageStream stream-based coverage. |
| ingest/ledgerbackend/buffered_storage_backend_bench_test.go | Replaces raw benchmark with BenchmarkBufferedStorageStream. |
| ingest/ledgerbackend/captive_core_backend.go | Removes GetLedgerRaw and updates related comments. |
| ingest/ledgerbackend/captive_core_backend_test.go | Replaces GetLedgerRaw tests with TestCaptiveCoreStream. |
| ingest/ledgerbackend/rpc_backend.go | Removes GetLedgerRaw and updates comments/error strings accordingly. |
| ingest/ledgerbackend/rpc_backend_test.go | Replaces GetLedgerRaw tests with TestRPCStream. |
| ingest/ledgerbackend/metrics.go | Removes metrics passthrough for GetLedgerRaw. |
| ingest/ledgerbackend/metrics_test.go | Updates metrics test to only cover GetLedger summary recording. |
| ingest/ledgerbackend/mock_database_backend.go | Removes GetLedgerRaw from the mock backend. |
| ingest/ledgerbackend/buffered_meta_pipe_reader.go | Updates comments to remove GetLedgerRaw references. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Resolves Copilot review comments on the LedgerStream PR: - getLedgerRaw / captive fetchSequence: the "caller must hold lock" docs were false for the stream's legitimate single-owner use. Restate as: concurrent callers must hold the lock (GetLedger does); a single exclusive owner — the LedgerStream — may call lock-free. - Rename RPCLedgerBackend.fetchSequenceLocked -> fetchSequence (the "Locked" suffix is a "lock held" convention, which the stream's lock-free call contradicts) and document the same dual contract. - ledger_buffer: clamp the lastDecompressedSize fallback hint to maxBatchObjectSize so a single outsized/corrupt batch can't permanently inflate the pooled decompress buffer; mirrors the FrameContentSize cap. - buffered_meta_pipe_reader: note raw frames are decoded in GetLedger or directly by LedgerStream consumers. Doc/naming + a two-line clamp; no public API or behavior change. Co-Authored-By: Claude Opus 4.7 <[email protected]>
Shaptic
left a comment
There was a problem hiding this comment.
Awesome 👏 can't wait for this to get used downstream. Just missing a changelog for the release?
LedgerStream is the streaming-first ingestion API, but it reads via the internal lock-free, zero-copy path (getLedgerRaw/fetchSequence) instead of GetLedger, so the metricsLedgerBackend decorator never sees its reads — a consumer adopting the streaming API emitted no metrics at all. Add WithStreamMetrics(registry, namespace) as a per-call StreamOption on RawLedgers. fetch_duration is timed once in the shared streamRaw skeleton (the single point all three backends funnel reads through), and the captive-core suite is registered on the core that call builds via the backend's own registerMetrics — the same call WithMetrics makes — so both ingestion APIs emit the identical metric names. The summary definition is factored into newLedgerFetchDurationSummary, shared with WithMetrics. Metrics register on the call, so a registry instruments a single RawLedgers invocation; a second instrumented call panics on duplicate registration. Uninstrumented calls stay reusable, and the clock is only sampled when instrumented to keep the per-ledger hot path free of wasted work. Make CaptiveStellarCore.closed atomic: the captive gauges read the live core from the Prometheus scrape goroutine (GetLatestLedgerSequence) while the lock-free reader advances it and Close writes the flag under the read lock — no single mutex covers all three callers. Caught by a concurrent-scrape test under -race. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
What
Introduce a streaming-first, interchangeable ingestion source interface:
with constructors for all three backends:
Each implementation owns its whole lifecycle: it builds its backend and prepares
the range when iteration begins, and tears everything down when iteration ends
(completion, early break, error, or ctx cancellation), logging close errors. The
three share a single streamRaw skeleton + rawReader so they don't diverge.
This also removes the unreleased GetLedgerRaw (the LedgerBackend interface
method, its metricsLedgerBackend passthrough, and the mock) — streaming replaces
its only use. GetLedger (decoded random access) is unchanged.
A second commit pools the zstd decompression output buffer in
BufferedStorageBackend: streaming-compressed objects omit FrameContentSize, so
DecodeAll's destination stayed nil and it reallocated the full output on every
batch. It now pre-sizes from the previous batch's decompressed size and reuses the
buffer, and sets WithDecoderConcurrency(1) (the consumer drives decompress
serially). Under parallel multi-chunk ingest this allocation dominated.
Why
Streaming is the dominant ingestion pattern, but PrepareRange + per-ledger
GetLedger + Close is a stateful sequence that's easy to misuse and forces a
per-ledger XDR copy/decode. LedgerStream collapses that into one call and lets
consumers swap captive-core / datastore / RPC sources without touching their loop,
while each implementation still reaches its own internals for a zero-copy read.
The per-step read is lock-free: a stream exclusively owns its backend on a
single goroutine, so the only thing the backend's read lock guards — a concurrent
Close — cannot happen, and a blocked read is cancelled via ctx. Each yielded
slice is a borrow that aliases the backend's internal frame and is valid only
until the next iteration step; consumers copy if they need to retain it. Together
this removes both the per-ledger lock and the per-ledger copy from the hot path.
Each backend exposes an injectable factory seam so the streams are unit-testable
without real GCS/core/RPC.
Known limitations
on the interface and rawReader). Callers that retain must copy.
consumers unmarshal as needed.
release; GetLedger is untouched.
Testing
TestBufferedStorageStream, TestCaptiveCoreStream, and TestRPCStream exercise
each implementation through its factory seam (lifecycle, ordering, bounded/error/
early-break paths). Full ingest/ledgerbackend and ingest/loadtest suites pass.