Add streaming-loadtest ledger backend for apply-load replay - #678
Closed
aditya1702 wants to merge 1 commit into
Closed
Add streaming-loadtest ledger backend for apply-load replay#678aditya1702 wants to merge 1 commit into
aditya1702 wants to merge 1 commit into
Conversation
A dev-only LEDGER_BACKEND_TYPE=streaming-loadtest that reads stream-framed LedgerCloseMeta from named pipes written by stellar-core apply-load (one FIFO per transaction profile), renumbers each stream onto the consumer's requested sequence with per-pipe diffs, merges the per-sequence frames into one mixed-traffic ledger via the SDK's loadtest.MergeLedgers, and stamps monotone wall-clock close times (apply-load emits closeTime 0). Renumbering makes both sides restartable without a database reset: a restarted apply-load resets to raw sequence 1 and is mapped onto the next requested ledger; a restarted consumer resumes from its cursor. Pacing via --loadtest-ledger-close-duration bounds the ledger rate, and FIFO backpressure throttles the generators to match. Because apply-load's benchmark mode publishes no history archive, this backend type skips the archive connection and the cursor-0 checkpoint bootstrap: ingestion starts from ledger 1 on an empty database and balance state accumulates from the ledger stream. Everything downstream of the backend (live ingest loop, processors, persistence) is unchanged, so a load test exercises the same code path as production ingestion. Includes an opt-in corpus test (STREAMING_LOADTEST_CORPUS) that replays real apply-load output through the backend and the production transaction reader; verified against v27 sac/custom_token/soroswap corpora. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Pull request overview
Adds a development-only streaming ledger backend for replaying and merging stellar-core apply-load metadata through live ingestion.
Changes:
- Adds FIFO-based ledger reading, merging, renumbering, pacing, and restart handling.
- Adds configuration flags and archive-free startup behavior.
- Adds unit and opt-in corpus tests.
Reviewed changes
Copilot reviewed 11 out of 12 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
internal/services/ingest_live.go |
Adds archive-free startup. |
internal/ingest/streaming_loadtest_ledger_backend.go |
Implements the streaming backend. |
internal/ingest/streaming_loadtest_ledger_backend_test.go |
Tests backend behavior. |
internal/ingest/streaming_loadtest_corpus_test.go |
Adds real-corpus replay testing. |
internal/ingest/ledger_backend.go |
Registers the backend factory. |
internal/ingest/ingest.go |
Adds configuration and skips archive connection. |
go.mod |
Adds indirect dependencies. |
go.sum |
Records dependency checksums. |
cmd/utils/custom_set_value.go |
Adds string-list parsing. |
cmd/utils/custom_set_value_test.go |
Tests string-list parsing. |
cmd/protocol_migrate.go |
Rejects streaming migration use. |
cmd/ingest.go |
Adds flags and validation. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+261
to
+267
| } else if int64(frameSeq)+p.seqDiff != int64(sequence) { | ||
| // apply-load emits every ledger it closes, in order, with no | ||
| // gaps. A mismatch inside an epoch means frames were lost or | ||
| // reordered — unrecoverable, unlike a writer restart. | ||
| return xdr.LedgerCloseMeta{}, fmt.Errorf( | ||
| "sequence mismatch within stream epoch: raw ledger %d + diff %d != requested %d", | ||
| frameSeq, p.seqDiff, sequence) |
Comment on lines
+308
to
+315
| err = db.RunInTransaction(ctx, m.models.DB, func(dbTx pgx.Tx) error { | ||
| return m.initializeCursors(ctx, dbTx, startLedger) | ||
| }) | ||
| if err != nil { | ||
| return fmt.Errorf("initializing cursors without archive bootstrap: %w", err) | ||
| } | ||
| m.appMetrics.Ingestion.LatestLedger.Set(float64(startLedger)) | ||
| m.appMetrics.Ingestion.OldestLedger.Set(float64(startLedger)) |
Comment on lines
+200
to
+203
| if i == 0 { | ||
| merged = frame | ||
| if err := loadtest.UpdateLedgerSeqInLedgerEntries(&merged, mapSeq); err != nil { | ||
| return xdr.LedgerCloseMeta{}, fmt.Errorf("renumbering entries from pipe %s: %w", p.path, err) |
Comment on lines
+281
to
+284
| go func(path string, ch chan<- openPipeResult) { | ||
| f, err := os.OpenFile(path, os.O_RDONLY, 0) | ||
| ch <- openPipeResult{file: f, err: err} | ||
| }(p.path, p.opening) |
Comment on lines
+114
to
+115
| func NewStreamingLoadtestLedgerBackend(cfg StreamingLoadtestBackendConfig) (*StreamingLoadtestLedgerBackend, error) { | ||
| if len(cfg.MetaPipePaths) == 0 { |
Contributor
Author
|
Superseded by #679, which targets blend/pr6-integration-tests (what dev currently runs) so the backend can be tested against the live dev deployment. |
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
LEDGER_BACKEND_TYPE=streaming-loadtest: reads ledger meta from named pipes written bystellar-core apply-load(one FIFO per tx profile), merges the streams into one ledger sequence, renumbers so restarts on either side never need a DB reset, and stamps advancing close times (apply-load emits closeTime 0).--loadtest-meta-pipe-paths,--loadtest-ledger-close-duration(pacing; FIFO backpressure throttles the generators to match).Why
Load-test ingestion at rates the dev loadtest network can't produce (thousands of soroban TPL, sub-second ledgers). Only the ledger backend is swapped — the live ingest loop, processors, and persistence are unchanged, so results predict production behavior.
Testing
-race).STREAMING_LOADTEST_CORPUS) replayed real v27 sac/custom_token/soroswap apply-load output: 200 merged ledgers, ~57k txs, all parsed by the production transaction reader.rpc/datastorebehavior unchanged (validation branches exercised manually).🤖 Generated with Claude Code