Run fetch, process and database writes at the same time by aditya1702 · Pull Request #706 · stellar/wallet-backend

aditya1702 · 2026-08-26T01:08:34Z

Start here: ingestLiveLedgers in internal/services/ingest_live.go. The first three commits just clear the way for it.

Problem

Live ingestion ran fetch → process → write-to-DB one after another, so each ledger cost the total of all three. At high transaction volume that total can exceed the ledger close time even when each step on its own fits comfortably inside it.

Fix

Run the three steps at the same time, using an errgroup and channels: while ledger N is being written, N+1 is processed and N+2 is fetched. A ledger now costs the slowest step, not the sum.

Four rules shape the design:

Writes stay in ledger order. The cursor and the per-protocol compare-and-swap step from N-1 to N, so nothing can jump ahead. Writing is also the only step that touches the DB, so the advisory-lock health check moves there.
Classification planning stays with the write step. It reads the pool outside a transaction and must see the previous ledger's committed protocol_wasms rows.
Two buffers take turns, and are only cleared when picked up — so clearing one can never corrupt a write still reading it.
If any step fails, the process exits and restarts from the cursor. Ledgers already in flight are dropped, not written: the cancelled context fails both the lock check and the write transaction.

The three preparation commits: delete unused fields and the no-op --latest-ledger-cursor-name flag, simplify persistLedgerData's arguments, and reuse utils.RetryWithBackoff instead of a hand-written retry loop.

Two metrics change meaning — please re-check any alert on them

Ingestion.Duration is now per-ledger work, not wall-clock. It sums the three step times, and those steps overlap across different ledgers. With process=2s and write=3s the pipeline finishes a ledger every 3s but reports 5s. Anything comparing this histogram against the 5s close-time budget as a "are we keeping up" proxy will over-report — use LagLedgers for that instead.
RetriesTotal{db_persist} records one fewer retry per exhausted ladder (4, not 5, for a 5-attempt exhaustion). RetryWithBackoff counts a retry only when it is about to back off, which it skips on the final attempt. This makes db_persist consistent with ledger_fetch, which always counted this way.

Two things to watch

Channel depth is bounded by ledger count, not bytes. At full depth the pipeline holds 4 ledger metas, 2 buffers and 2 transaction slices — a few times the old loop's peak. Oversized ledgers under a memory limit are the axis to watch on the rig.
Throughput is bounded by the buffer count, not the channel depth. Processing cannot start a ledger without a free buffer, and a buffer comes back only when the write of the ledger it holds finishes. freeBuffers has room for every buffer that exists, so the hand-back never blocks and a deeper processed channel adds read-ahead, not throughput.

No performance numbers yet — please review the design and correctness.

PR 3 of 6 replacing #684 · schema → fixes → pipeline → parallel writes → table groups → final fixes

Copilot

Pull request overview

Pipelines live ledger fetching, processing, and persistence to improve ingestion throughput while preserving ordered writes.

Changes:

Adds a three-stage errgroup pipeline with rotating buffers.
Standardizes persistence retries and exhaustion reporting.
Removes unused ingestion fields and a deprecated CLI flag.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
`internal/utils/retry.go`	Adds retry-exhaustion classification.
`internal/utils/retry_test.go`	Tests retry error classification.
`internal/services/protocol_validation_dispatch_test.go`	Updates renamed retry reference.
`internal/services/ingest.go`	Removes unused fields and simplifies insertion results.
`internal/services/ingest_test.go`	Updates persistence tests and adds pipeline failure coverage.
`internal/services/ingest_live.go`	Implements the concurrent live-ingestion pipeline.
`internal/services/ingest_live_test.go`	Updates service construction.
`internal/services/ingest_backfill.go`	Adapts to simplified insertion results.
`internal/ingest/ingest.go`	Removes obsolete service dependencies.
`internal/indexer/indexer.go`	Narrows the buffer interface and removes unused state.
`internal/indexer/indexer_test.go`	Updates indexer fixtures.
`cmd/ingest.go`	Removes the deprecated cursor-name flag.

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

Copilot

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

internal/utils/retry.go:82

A cancellation that occurs while the final attempt is running is labeled as retry exhaustion here. persistLedgerDataWithRetry consequently increments both RetryExhaustionsTotal and ErrorsTotal for a normal shutdown, despite its explicit cancellation branch; the added test only covers a context canceled before the first attempt. Check ctx.Err() before breaking from the final attempt, and add a regression test where the final fn cancels the context.

	return zero, fmt.Errorf("%w: failed after %d attempts: %w", ErrRetriesExhausted, maxRetries, lastErr)

Copilot

Pull request overview

Copilot reviewed 13 out of 13 changed files in this pull request and generated 1 comment.

Suppressed comments (3)

internal/services/ingest_live.go:665

If fetch or processing fails while this transaction is in flight, errgroup cancellation makes persistLedgerDataWithRetry return a cancellation, but this line records it as another ingest_live fault. That contradicts the helper's shutdown-not-fault handling and makes the error metric timing-dependent. Skip this increment when ctx.Err() != nil.

		if err := m.persistLedgerDataWithRetry(ctx, pl.seq, pl.meta, plan, pl.transactions, pl.buffer); err != nil {
			m.appMetrics.Ingestion.ErrorsTotal.WithLabelValues("ingest_live").Inc()
			return fmt.Errorf("persisting ledger %d: %w", pl.seq, err)

internal/services/ingest_live.go:605

A sibling-stage failure can cancel ctx while processLedger is running, causing this block to record the propagated cancellation as another independent ingest_live error. Guard the increment with ctx.Err() == nil so the top-level error counter records the root stage failure once rather than depending on pipeline timing.

		if err != nil {
			m.appMetrics.Ingestion.ErrorsTotal.WithLabelValues("ingest_live").Inc()
			return fmt.Errorf("processing ledger %d: %w", fl.seq, err)

internal/utils/retry.go:82

If fn returns because ctx is cancelled during the final attempt, control reaches this return without another context check and incorrectly adds ErrRetriesExhausted. persistLedgerDataWithRetry then reports retry exhaustion and fault metrics during an ordinary shutdown. Check ctx.Err() here before constructing the exhaustion error, and add a final-attempt cancellation regression case.

	return zero, fmt.Errorf("%w: failed after %d attempts: %w", ErrRetriesExhausted, maxRetries, lastErr)

Copilot

Copilot reviewed 13 out of 13 changed files in this pull request and generated no new comments.

Copilot

Copilot review overview

🔵 Needs a closer look

The concurrency, cancellation, cursor, and transactional-ordering changes require final human validation.

Review effort: Balanced
Findings: 1

Open (1)

Correct Duration semantics or include fetch time · New

+		ledgerDuration := pl.processDuration + classifyDuration + persistDuration
+		m.appMetrics.Ingestion.Duration.Observe(ledgerDuration.Seconds())


Copilot

Copilot review overview

🔵 Needs a closer look

The concurrent ingestion and transactional failure behavior warrants final human review and load validation.

Review effort: Balanced
Findings: 1

Open (1)

Correct Duration semantics or include fetch time

Copilot

Copilot review overview

🔵 Needs a closer look

The concurrency, cancellation, and ordered database-write guarantees warrant final human review, and the PR description retains an obsolete channel-capacity invariant.

Review effort: Balanced
Findings: 1 · 1

Open (2)

Correct Duration semantics or include fetch time
Update obsolete buffer-capacity invariant in PR description · New

+	// The channel holds every buffer that exists, so handing one back never
+	// blocks. The buffer count, not the channel depths, bounds the pipeline:
+	// process cannot start a ledger without a free buffer, so a deeper
+	// `processed` queue only helps if more buffers are added with it.


…flag ingestService carried four fields (appTracker, getLedgersLimit, knownContractIDs, contractMetadataService) and Indexer two (ingestionMetrics, networkPassphrase) that were assigned at construction and never read — consumers get these through their own constructors or through internal/ingest.Configs. Deleting them also drops AppTracker, GetLedgersLimit, and ContractMetadataService from IngestServiceConfig (the checkpoint service keeps its own metadata-service wiring). Also removes the --latest-ledger-cursor-name flag, which only fed a deprecation warning, and the phantom load_current_state phase label from the ProtocolStateProcessingDuration help text (never emitted).

Every remaining parameter of persistLedgerData had exactly one production value: cursorName was always the latest-ledger cursor (the blind-Update branch was reachable only from tests), ledgerMeta was never nil, and the numTxs/numOps returns only fed two counters the buffer already knows via GetNumberOfTransactions/GetNumberOfOperations. The signature is now (ctx, ledgerSeq, ledgerMeta, plan, contractData, buffer) error, the cursor write is unconditionally guarded, and the contractDataMemo nil-receiver branch (test-only) is gone. IndexerBufferInterface shrinks to its real seam — the indexer's fold-in plus the five getters insertIntoDB reads; everything else is reached on the concrete *IndexerBuffer. Tests now exercise the guarded cursor path persistLedgerData actually runs in production.

…Backoff persistLedgerDataWithRetry (né ingestProcessedDataWithRetry) hand-rolled the same attempt loop, exponential backoff, cap, and context handling that utils.RetryWithBackoff already provides — the body is now a single RetryWithBackoff call with isPermanentPersistError as the classifier. Metric semantics preserved: transient retries count RetriesTotal, permanent errors count ErrorsTotal and return immediately, exhaustion counts both plus RetryExhaustionsTotal, and context cancellation counts neither. RetryWithBackoff's exhaustion return now wraps the exported sentinel utils.ErrRetriesExhausted alongside the final error, which is how the wrapper tells exhaustion apart from a permanent-error exit.

…ist stages The live loop ran fetch, process_ledger, and the DB persist strictly in sequence, so the per-ledger time was their sum — which can exceed the ledger cadence at high transaction volume even when each stage individually fits inside it. The loop is now a three-stage errgroup pipeline over depth-1 channels: while ledger N persists, N+1 processes and N+2 is fetched, making the ledger time the slowest stage instead of the sum. Persist stays strictly sequential in ledger order (the guarded cursor and per-protocol CAS chain advance N-1 → N) and is the only writing stage, so the advisory-lock liveness probe moves there. Classification planning also stays on the persist stage: its known-hash lookup is a non-transactional pool read that must observe the previous ledger's committed protocol_wasms rows. Two IndexerBuffers rotate through a freeBuffers channel, cleared only at take so a clear can never tear a persist still reading the maps the getters alias. Any stage error cancels the pipeline and the process exits, resuming from the cursor on restart — the Duration metric now sums the ledger's stage times, since wall time under a pipeline would count queue wait.

The ingestLiveLedgers comment claimed a stage error cancels the pipeline and returns. Cancellation and the producer's channel close both fire, so a downstream stage picks either and a ledger already handed off can still be persisted first — which is what we want, since every ledger that lands moves the cursor forward for the restart. Say that instead. Also record why the stage hand-off channels carry a slot: the overlap comes from receiving input before claiming run resources, not from the depth, so the slot is there to ride out a backend's bursty prefetch refill.

… error The persist stage probes the advisory-lock session before every ledger, but the probe ran on the pipeline context immediately after the receive from `processed`. On cancellation that channel usually holds a ledger, so the select can pick the ledger over ctx.Done(); the probe then fails with "context canceled" and the stage reports a lock loss that never happened, incrementing ErrorsTotal{ingest_live}. A plain SIGTERM was enough to trigger it. Check cancellation before attributing the failure to the lock, using the same wording the fetch and process stages already use.

persistLedgerData's ledgerMeta parameter became a value, which dropped the `ledgerMeta != nil` half of the guard around the protocol-processor block. LedgerCloseTime() on a zero-value LedgerCloseMeta dereferences a nil V0 arm and panics, and ledgerSeq-1 underflows to 4294967295. Three tests passed xdr.LedgerCloseMeta{} and survived only because their services register no protocol processors — the first test to register one would panic instead of failing. Use dummyLedgerMeta, the fixture already in this package, so the trap is gone at the source rather than papered over with a guard the live path can never need.

Four comments no longer described the code: - The shutdown paragraph claimed in-flight ledgers are drained and persisted so a restart lands further ahead. They are not: errgroup cancels the group context as soon as a stage fails, which fails both the lock probe and the persist transaction, so a handed-off ledger is always dropped. - The freeBuffers comment justified why handing a buffer back never blocks at today's capacities without naming the invariant that makes it true: cap(freeBuffers) >= cap(processed) + 1. Raising `processed` on its own — the obvious tuning knob — deadlocks the pipeline. - Ingestion.Duration sums three stage times that overlap in wall-clock across different ledgers, so it measures per-ledger work and is no longer a keep-up proxy. Say so, and point at LagLedgers instead. - One stale reference to ingestProcessedDataWithRetry survived the rename. Also note that channel depth bounds the pipeline by ledger count, not bytes, since peak working set grew with the extra in-flight metas and buffers.

… an attempt RetryWithBackoff only checked the context between attempts, so a cancellation landing while fn ran fell through as an ordinary failure: on the last attempt it was tagged ErrRetriesExhausted, and on earlier ones it reached onRetry, bumping the retry counter and logging "retrying" for a shutdown. persistLedgerDataWithRetry then counted a clean SIGTERM as a db_persist exhaustion, against its own documented contract. Check ctx.Err() right after fn fails, before the permanent classifier and onRetry, and return the context error untagged in both cases.

… for lock loss When one stage fails, errgroup cancels the pipeline context and the other stages' in-flight calls fail with that cancellation, so a single fault incremented ErrorsTotal{ingest_live} a nondeterministic number of times. Every stage now counts only when ctx.Err() is nil. The advisory-lock probe checked cancellation before running, so a cancel landing mid-probe still surfaced as a lost lock; it now checks after. The buffer return to the rotation is paired with ctx.Done() so it can never strand the persist stage should the rotation ever fill.

The comment claimed raising the processed queue alone would deadlock the pipeline. It cannot: the free-buffer channel holds every buffer that exists, so the hand-back send always has room. The buffer count is what bounds throughput; say that instead.

…loss The lag-read test drives the lock probe through a cancelled context, the way the real probe fails during a shutdown, and asserts the run returns context.Canceled without the lock-loss wording or an ingest_live error count. Two assertions on a buffer built with no transactions are removed: they could not fail. The fetch-slot comment states the backend prefetch depth as it is: hundreds of ledgers, not a couple.

Copilot

Copilot review overview

🔵 Needs a closer look

The concurrent pipeline, cancellation behavior, and ordered transactional invariants warrant final human validation.

Review effort: Balanced
Findings: 1 · 1

Open (2)

Correct Duration semantics or include fetch time
Update obsolete buffer-capacity invariant in PR description

aditya1702 mentioned this pull request Aug 26, 2026

Pipeline live ingestion into fetch/process/persist with seven parallel write streams #684

Closed

aditya1702 changed the title ~~Pipeline live ingestion into fetch, process and persist stages~~ Run fetch, process and database writes at the same time Aug 26, 2026

aditya1702 force-pushed the persist/3-pipeline branch from f831563 to d247b7e Compare August 26, 2026 20:25

aditya1702 force-pushed the persist/3-pipeline branch from 95e047a to e917c69 Compare August 27, 2026 02:52

aditya1702 requested review from aristidesstaffieri and a balanced review from Copilot August 28, 2026 15:01

Copilot started reviewing on behalf of aditya1702 August 28, 2026 15:01

View session

Copilot AI reviewed Aug 28, 2026

View reviewed changes

Comment thread internal/utils/retry.go

Comment thread internal/services/ingest_live.go

aristidesstaffieri reviewed Aug 28, 2026

View reviewed changes

Comment thread internal/services/ingest_live.go Outdated

Copilot AI review requested due to automatic review settings September 3, 2026 23:46

aditya1702 force-pushed the persist/3-pipeline branch from e917c69 to 5953f78 Compare September 3, 2026 23:46

Copilot started reviewing on behalf of aditya1702 September 3, 2026 23:46

View session

Copilot AI reviewed Sep 3, 2026

View reviewed changes

Comment thread internal/services/ingest_live.go

Copilot AI review requested due to automatic review settings September 4, 2026 19:31

aditya1702 force-pushed the persist/3-pipeline branch from 5953f78 to b9f173b Compare September 4, 2026 19:31

Copilot started reviewing on behalf of aditya1702 September 4, 2026 19:31

View session

Copilot AI reviewed Sep 4, 2026

View reviewed changes

Comment thread internal/services/ingest_live.go

Copilot AI review requested due to automatic review settings September 9, 2026 21:07

aditya1702 force-pushed the persist/3-pipeline branch from b9f173b to b6fee8a Compare September 9, 2026 21:07

Copilot started reviewing on behalf of aditya1702 September 9, 2026 21:07

View session

Copilot AI reviewed Sep 9, 2026

View reviewed changes

JiahuiWho reviewed Sep 11, 2026

View reviewed changes

Comment thread internal/services/ingest_live.go

aditya1702 self-assigned this Sep 12, 2026

aditya1702 added this to the Wallet Backend O3 Performance Testing milestone Sep 12, 2026

aditya1702 force-pushed the persist/3-pipeline branch from b13cc19 to 3819abf Compare October 1, 2026 21:06

Copilot started reviewing on behalf of aditya1702 October 1, 2026 21:07

View session

Copilot AI reviewed Oct 1, 2026

View reviewed changes

Comment thread internal/services/ingest_live.go

Comment on lines +684 to +685

ledgerDuration := pl.processDuration + classifyDuration + persistDuration

m.appMetrics.Ingestion.Duration.Observe(ledgerDuration.Seconds())

aditya1702 force-pushed the persist/3-pipeline branch from 3819abf to 5a87fce Compare October 5, 2026 14:57

Copilot AI balanced review requested due to automatic review settings October 5, 2026 14:57

Copilot started reviewing on behalf of aditya1702 October 5, 2026 14:58

View session

Copilot AI reviewed Oct 5, 2026

View reviewed changes

Copilot AI balanced review requested due to automatic review settings October 5, 2026 17:35

aditya1702 force-pushed the persist/3-pipeline branch from 5a87fce to 50bfd28 Compare October 5, 2026 17:35

Copilot started reviewing on behalf of aditya1702 October 5, 2026 17:36

View session

Copilot AI reviewed Oct 5, 2026

View reviewed changes

Base automatically changed from persist/2-fixes to main October 5, 2026 17:44

aditya1702 added 11 commits October 6, 2026 12:04

aditya1702 force-pushed the persist/3-pipeline branch from 50bfd28 to 457474c Compare October 6, 2026 16:06

Copilot AI balanced review requested due to automatic review settings October 6, 2026 16:52

Copilot started reviewing on behalf of aditya1702 October 6, 2026 16:53

View session

Copilot AI reviewed Oct 6, 2026

View reviewed changes

aditya1702 merged commit 7eb6377 into main Oct 6, 2026
10 checks passed

aditya1702 deleted the persist/3-pipeline branch October 6, 2026 17:09

Navigation Menu

Repository navigation

Run fetch, process and database writes at the same time - #706

Run fetch, process and database writes at the same time#706
aditya1702 merged 12 commits into
mainfrom
persist/3-pipeline

aditya1702 commented Aug 26, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

		ledgerDuration := pl.processDuration + classifyDuration + persistDuration
		m.appMetrics.Ingestion.Duration.Observe(ledgerDuration.Seconds())

Conversation

aditya1702 commented Aug 26, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Problem

Fix

Two metrics change meaning — please re-check any alert on them

Two things to watch

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

🔵 Needs a closer look

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

🔵 Needs a closer look

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

🔵 Needs a closer look

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot review overview

🔵 Needs a closer look

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

aditya1702 commented Aug 26, 2026 •

edited

Loading