Skip to content

ingest/ledgerbackend: add LedgerStream streaming ingestion API - #5944

Merged
tamirms merged 4 commits into
mainfrom
ledger-stream
Jun 4, 2026
Merged

ingest/ledgerbackend: add LedgerStream streaming ingestion API#5944
tamirms merged 4 commits into
mainfrom
ledger-stream

Conversation

@tamirms

@tamirms tamirms commented May 25, 2026

Copy link
Copy Markdown
Contributor

What

Introduce a streaming-first, interchangeable ingestion source interface:

type LedgerStream interface {
    // RawLedgers yields the raw XDR bytes of each ledger in r, in order.
    RawLedgers(ctx context.Context, r Range) iter.Seq2[[]byte, error]
}

with constructors for all three backends:

  • NewBufferedStorageStream(cfg, dsConfig, log) — GCS/S3 datastore
  • NewCaptiveCoreStream(config, log) — captive stellar-core
  • NewRPCStream(options, log) — RPC server

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

  • Yielded bytes are borrows, not safe to retain past the next step (documented
    on the interface and rawReader). Callers that retain must copy.
  • No decoded-LedgerCloseMeta streaming helper here — streaming yields raw XDR;
    consumers unmarshal as needed.
  • Single-consumer only: a LedgerStream is meant to be driven by one goroutine.
  • GetLedgerRaw removal is not a breaking change — it was never in a tagged
    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.

tamirms and others added 2 commits May 25, 2026 13:35
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]>
@tamirms
tamirms marked this pull request as ready for review May 25, 2026 17:36
Copilot AI review requested due to automatic review settings May 25, 2026 17:36

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 GetLedgerRaw from LedgerBackend and 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.

Comment thread ingest/ledgerbackend/buffered_storage_backend.go
Comment thread ingest/ledgerbackend/ledger_stream.go
Comment thread ingest/ledgerbackend/ledger_stream.go
Comment thread ingest/ledgerbackend/ledger_stream.go Outdated
Comment thread ingest/ledgerbackend/ledger_buffer.go
Comment thread ingest/ledgerbackend/buffered_meta_pipe_reader.go Outdated
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]>
@tamirms
tamirms requested a review from a team May 27, 2026 08:44
@tamirms tamirms added this to the platform sprint 72 milestone Jun 2, 2026
@tamirms tamirms moved this from To Do to Needs Review in Platform Scrum Jun 2, 2026

@Shaptic Shaptic left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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]>
@tamirms
tamirms merged commit ff1e140 into main Jun 4, 2026
11 checks passed
@github-project-automation github-project-automation Bot moved this from Needs Review to Done in Platform Scrum Jun 4, 2026
@tamirms
tamirms deleted the ledger-stream branch June 4, 2026 22:09
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

Status: Done

Development

Successfully merging this pull request may close these issues.

3 participants