Write the big tables on separate database connections, in batches by aditya1702 · Pull Request #711 · stellar/wallet-backend

aditya1702 · 2026-08-26T16:45:02Z

Start here: the comment above persistLedgerData in internal/services/ingest_live.go. Its commit-order rules explain everything else in this PR.

1. Big tables are written on separate DB connections

The three big COPY groups ran one after another, in a single transaction on a single connection — so the write step cost their total, and one Postgres backend did all the index work.

They now run at the same time, each on its own connection and its own transaction. A fourth "coordinating" transaction handles everything else. No foreign keys point between these table groups, so they can't block each other.

Nothing becomes visible until all four are done. The three table transactions commit first; the coordinating transaction — which holds the cursor that decides which ledgers exist — commits last.

Failure before the first commit → everything rolls back, retried as before.
Failure after → returns ErrPartialPersist and is fatal. COPY has no ON CONFLICT, so retrying would hit duplicate-key errors.

The only mess a crash can leave is rows above the cursor, which DeleteRowsAboveLedger clears on startup.

2. Backlogs can be written in batches — off by default

--live-persist-max-batch-size defaults to 1: one commit per ledger, same as before this PR. Testnet and pubnet close times comfortably exceed persist time, so batching stays off. Raise it only where persist can't keep up.

When raised, up to that many consecutive ledgers get written in one commit. Two rules keep that correct:

A ledger carrying a contract we haven't classified before always starts its own batch. Its classification reads then see everything the previous batch committed.
Event membership is read through the write transaction, not the pool. A contract classified at the head of a batch is still uncommitted, so a pool read would miss it and silently drop that contract's SEP-41 events for the rest of the batch.

3. Reads are bounded by the cursor

Because the table transactions commit first, the tables briefly hold rows for ledgers the cursor hasn't reached — and after a crash they hold them until startup cleanup runs. Nothing stopped the API from serving those rows.

The problem isn't wrong data, it's incomplete data that looks complete: a transaction whose state changes hadn't landed yet returns an empty list, which a client can't tell from a real one. Rows also appeared, got cleaned up, then came back on re-ingestion.

Every read now filters out rows above the cursor. Applied at the five places a client asks for something directly:

Read	Bounded on
transaction by hash	`to_id`
operation by id	`id`
account's transactions	`tx_to_id`
account's operations	`operation_id`
account's state changes	`to_id`

Everything else is a batch load keyed by a row we already fetched, so bounding the five entry points covers those too.

Cost, measured on the dev mainnet DB (62GB, 543M rows): 0.06ms and two extra buffer reads. It's one primary-key lookup, done once per query. Same query plan, same number of chunks read.

4. `synchronous_commit = off` on the three table transactions

Durability is unchanged. The coordinating transaction commits normally and last, and its flush covers their write-ahead log records too.

Known, handled later in the stack

Correct at the default of 1. If you raise the batch size, three follow-ups matter:

What	Where
Mid-batch ledgers keep their committed contract membership	#709
Startup cleanup bounded by close time (else it scans every chunk)	#712
Pool-size guard + true per-commit persist metric	#709

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

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

Copilot

Pull request overview

Parallelizes live-ledger persistence, adds backlog batching, and prevents API reads from exposing rows beyond the committed ingest cursor.

Changes:

Splits persistence across coordinated transactions with ordered commits and recovery.
Adds configurable ledger batching and classification-aware batch boundaries.
Bounds direct transaction, operation, and state-change reads by the ingest cursor.

Reviewed changes

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

Show a summary per file

File	Description
`cmd/ingest.go`	Adds the live persist batch-size flag.
`internal/ingest/ingest.go`	Propagates batch configuration.
`internal/metrics/ingestion.go`	Adds persist batch-size metrics.
`internal/services/ingest.go`	Stores batching and classification state.
`internal/services/ingest_live.go`	Implements parallel transactions, batching, and reconciliation.
`internal/services/ingest_live_test.go`	Tests partial-persist error classification.
`internal/services/ingest_test.go`	Updates persistence tests and adds batch scenarios.
`internal/data/query_utils.go`	Adds the shared cursor-bound SQL helper.
`internal/data/transactions.go`	Bounds direct transaction reads.
`internal/data/transactions_test.go`	Tests transaction cursor visibility.
`internal/data/operations.go`	Bounds direct operation reads.
`internal/data/statechanges.go`	Bounds account state-change reads.
`internal/data/statechanges_test.go`	Tests state-change cursor visibility.
`internal/data/protocol_contracts.go`	Supports transaction-scoped membership reads.
`internal/data/protocol_contracts_test.go`	Updates membership-query tests.
`internal/data/mocks.go`	Updates the protocol-contract model mock.
`internal/data/ingest_store.go`	Adds startup cleanup above the cursor.
`internal/data/ingest_store_test.go`	Tests cleanup across bulk tables.

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

+	for i, s := range siblings {
+		if commitErr := siblingTxs[i].Commit(ctx); commitErr != nil {
+			if i > 0 {
+				return fmt.Errorf("committing %s for %s: %w: %w", s.name, label, ErrPartialPersist, commitErr)
+			}
+			return fmt.Errorf("committing %s for %s: %w", s.name, label, commitErr)
+		}
+	}
+	// The coordinating transaction commits strictly last: it carries the
+	// cursor, so its commit is the point at which the batch's ledgers exist.
+	if commitErr := coordTx.Commit(ctx); commitErr != nil {
+		return fmt.Errorf("committing coordinating transaction for %s: %w: %w", label, ErrPartialPersist, commitErr)
+	}


-			return fmt.Errorf("inserting processed data into db for ledger %d: %w", ledgerSeq, txErr)
+			if historySwapped {
+				persistStart := time.Now()
+				persistErr := processor.PersistHistory(ctx, dbTx)


+		classifyShare := classifyDuration / time.Duration(len(batch))
+		persistShare := persistDuration / time.Duration(len(batch))
+		for _, pl := range batch {
+			m.appMetrics.Ingestion.PhaseDuration.WithLabelValues("insert_into_db").Observe(persistShare.Seconds())
+


+		// cursor, so a crash between those commits orphans (at most) the single
+		// ledger past the cursor. Fatal on failure — ingesting over the orphans


+// those commits leaves orphaned bulk rows for (at most) the single ledger past
+// the committed cursor — and they must be cleared before that ledger is


+func (m *ingestService) persistLedgerData(ctx context.Context, items []persistItem) error {
+	label := batchLabel(items)


+	var queryBuilder strings.Builder
+	fmt.Fprintf(&queryBuilder, `SELECT %s FROM operations WHERE id = $1`, columns)
+	appendIngestCursorBound(&queryBuilder, "id")


Copilot

Pull request overview

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

Suppressed comments (8)

internal/services/ingest_live.go:424

PersistHistory writes SEP-41/Blend history into state_changes through the coordinating transaction, while the sibling goroutine is concurrently copying the indexer's rows into the same hypertable. At a chunk boundary, either insert can wait on chunk creation/locks held by the other transaction; the commit barrier then waits for both goroutines before committing the lock holder, creating an application-level deadlock PostgreSQL cannot detect. Route all state_changes producers through the same sibling transaction (serialized because a pgx.Tx cannot be used concurrently) while keeping the cursor CAS on the coordinator.

			if historySwapped {
				persistStart := time.Now()
				persistErr := processor.PersistHistory(ctx, dbTx)

internal/services/ingest_live.go:221

The commit barrier still uses the cancellable pipeline context. If SIGTERM cancels ctx after one sibling commits, the next Commit fails immediately and turns an otherwise graceful shutdown into ErrPartialPersist, forcing crash recovery. Once the barrier starts, all sibling and coordinator commits must use a context detached from cancellation so the set finishes committing.

	for i, s := range siblings {
		if commitErr := siblingTxs[i].Commit(ctx); commitErr != nil {

internal/services/ingest_live.go:158

Live ingestion already pins one pool connection for the advisory lock, and this code now needs four more simultaneously (coordinator plus three siblings). Because --db-max-conns is configurable, values below 5 make one of these sequential Acquire calls wait forever while the already-open transactions retain their connections. Validate the pool capacity before ingestion starts and fail with an actionable error.

	siblingTxs := make([]pgx.Tx, len(siblings))
	for i, s := range siblings {
		conn, acquireErr := m.models.DB.Acquire(ctx)
		if acquireErr != nil {
			return fmt.Errorf("acquiring %s connection for %s: %w", s.name, label, acquireErr)

internal/services/ingest_live.go:825

Tracking only the contract ID makes an upgrade of an already-seen contract look classified even when it is now bound to a different wasm. Such a ledger can ride mid-batch without a plan, so its new binding is never classified before protocol processing and its events/state can be assigned using stale membership. Track the observed contract→wasm binding and cut whenever that binding changes.

	for contractID := range buffer.GetProtocolContracts() {
		if _, seen := m.classifiedContracts[contractID]; !seen {
			return true

internal/services/ingest_live.go:946

A non-head ledger can re-buffer a previously classified contract, but it receives a nil plan here. getEffectiveProtocolContracts removes every buffered contract from the committed set and only re-adds it when classification[wasmHash] matches; with a nil plan, the re-observed contract disappears and that ledger's protocol events/state changes are silently skipped even though membership is committed. Define nil as “do not reclassify” for these safe re-observations, rather than as an empty classification.

		items[0].plan = plan

internal/services/ingest_live.go:712

With a full batch of N buffers held by persist, N+1 leaves only one buffer for process, so the next backlog cannot accumulate to the configured batch size while the current batch writes. Because draining is non-blocking, subsequent commits collapse to one or a few ledgers instead of N under sustained load. The rotation needs N buffers for the active persist batch and N for the next process/queue batch.

	// Buffers rotate between the process and persist stages: process fills
	// one while persist drains the others, and reuse keeps the asset-parse
	// memo warm and the maps' backing arrays allocated across ledgers. The
	// rotation holds one more buffer than the persist batch cap — a full
	// batch in flight still leaves one for process to fill — and the
	// channel's capacity matches, so handing a buffer back never blocks.
	freeBuffers := make(chan *indexer.IndexerBuffer, batchCap+1)
	for range batchCap + 1 {
		freeBuffers <- indexer.NewIndexerBuffer()

internal/services/ingest_live.go:967

Dividing commit wall time and recording one observation per ledger makes insert_into_db report artificially low latency exactly when batching is active (for example, one 4s commit of 8 ledgers becomes eight 0.5s samples). This hides an overloaded persist stage from p99-based alerts. Record one persistDuration observation per actual commit; keep persistShare only for the per-ledger aggregate duration if that metric intentionally remains amortized.

		classifyShare := classifyDuration / time.Duration(len(batch))
		persistShare := persistDuration / time.Duration(len(batch))
		for _, pl := range batch {
			m.appMetrics.Ingestion.PhaseDuration.WithLabelValues("insert_into_db").Observe(persistShare.Seconds())

internal/services/ingest_live.go:846

These process-lifetime seen sets are append-only, so memory grows with every unique contract and wasm ever observed; this also happens when the batch cap is 1, where the sets are never needed. Cap-and-clear the sets (a miss only causes a conservative batch cut), and skip tracking entirely in unbatched mode.

func (m *ingestService) markClassificationInputsSeen(batch []processedLedger) {
	for _, pl := range batch {
		for hash := range pl.buffer.GetProtocolWasms() {
			m.classifiedWasms[hash] = struct{}{}
		}
		for hash := range pl.buffer.GetProtocolWasmBytecodes() {
			m.classifiedWasms[hash] = struct{}{}
		}
		for contractID := range pl.buffer.GetProtocolContracts() {
			m.classifiedContracts[contractID] = struct{}{}

+	err := db.RunInTransaction(ctx, m.DB, func(dbTx pgx.Tx) error {
+		for _, target := range targets {
+			start := time.Now()
+			tag, execErr := dbTx.Exec(ctx, fmt.Sprintf(`DELETE FROM %s WHERE %s >= $1`, target.table, target.column), minTOID)


Copilot

Pull request overview

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

Suppressed comments (7)

internal/services/ingest_live.go:428

PersistHistory runs on the coordinating transaction, but both the Blend and SEP-41 implementations call StateChanges.BatchCopy, concurrently with the state_changes sibling transaction above. At a TimescaleDB chunk boundary one transaction can wait for the other while g.Wait() prevents the sibling from reaching its commit, hanging ingestion indefinitely. Route all protocol-history state-change COPYs through the state-changes sibling transaction and serialize them there before the commit barrier.

			if historySwapped {
				persistStart := time.Now()
				persistErr := processor.PersistHistory(ctx, dbTx)

internal/services/ingest_live.go:233

The commit barrier uses the cancellable pipeline context. If SIGTERM cancels it after one sibling commits, the next Commit(ctx) fails with cancellation and is classified as ErrPartialPersist, turning a normal shutdown into a fatal partial persist. Once staging succeeds, detach cancellation for every sibling and coordinator commit so the barrier completes as one unit.

	for i, s := range siblings {
		if commitErr := siblingTxs[i].Commit(ctx); commitErr != nil {

internal/services/ingest_live.go:716

With an N-ledger batch in persist, batchCap+1 leaves only one buffer for the process stage, so it can queue at most one ledger before blocking. Consequently the next greedy drain usually forms a size-1 batch even under sustained backlog, defeating the batching option. The rotation needs room for the current N buffers plus the next N buffered/queued ledgers.

	freeBuffers := make(chan *indexer.IndexerBuffer, batchCap+1)
	for range batchCap + 1 {
		freeBuffers <- indexer.NewIndexerBuffer()

internal/services/ingest_live.go:830

Collapsing contracts by ID to the last binding loses bindings needed by earlier ledgers in the same batch. For example, if ledger N re-observes an already-classified C→W1 and emits events, then N+1 upgrades C to W2, only W2 reaches knownHashes; while staging N, classification[W1] is empty and getEffectiveProtocolContracts drops C, silently losing N's protocol events/history. Preserve every per-ledger binding needed for classification, or cut the batch at a rebind, while still persisting the final binding last.

		maps.Copy(wasms, pl.buffer.GetProtocolWasms())
		maps.Copy(bytecodes, pl.buffer.GetProtocolWasmBytecodes())
		maps.Copy(contracts, pl.buffer.GetProtocolContracts())

internal/data/ingest_store.go:296

These hypertables are partitioned by ledger_created_at, but the startup DELETE supplies only a TOID predicate. TOID chunk-skipping may reject rows, yet TimescaleDB still has to consider every historical chunk for each of the five statements, making restart reconciliation grow with the entire database. Add a lower ledger_created_at bound derived from the cursor ledger's close time so old chunks are excluded statically.

			tag, execErr := dbTx.Exec(ctx, fmt.Sprintf(`DELETE FROM %s WHERE %s >= $1`, target.table, target.column), minTOID)

internal/services/ingest_live.go:156

This path now requires five simultaneous pool connections: the advisory-lock connection is held for the run, then this function holds the coordinator plus three siblings. Because DBMaxConns remains configurable below five, the final Acquire can wait forever while this same call holds every available connection. Validate the live-ingest pool capacity before opening the transaction set (or use a dedicated write pool).

		conn, acquireErr := m.models.DB.Acquire(ctx)

internal/data/operations.go:28

The analogous transaction and state-change cursor bounds add regression coverage, but neither operation entry point is exercised with a row above the cursor. Add an operation-model test that seeds at-cursor and ahead rows and verifies both GetByID and the account list hide the ahead row; otherwise a query-shape regression here can expose partially committed operations unnoticed.

	appendIngestCursorBound(&queryBuilder, "id")

+	classifyShare := classifyDuration / time.Duration(len(batch))
+	persistShare := persistDuration / time.Duration(len(batch))
+	for _, pl := range batch {
+		m.appMetrics.Ingestion.PhaseDuration.WithLabelValues("insert_into_db").Observe(persistShare.Seconds())


+	if len(m.protocolProcessors) > 0 {
+		ledgerCloseTime := ledgerMeta.LedgerCloseTime()
+		contractEvents := buffer.GetContractEvents()
+		expected := strconv.FormatUint(uint64(ledgerSeq-1), 10)


aristidesstaffieri · 2026-08-28T18:23:16Z

+	}
+
+	// Stream the COPY families and stage the coordinated writes concurrently.
+	// The table sets are disjoint (no FKs among them), and the goroutines only


are they actually disjoint? It looks like at internal/services/ingest_live.go:150 we do a COPY of the state changes into state_changes while processor.PersistHistory also write to that table.

Copilot

Pull request overview

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

Suppressed comments (5)

internal/services/ingest_live.go:233

Once the first sibling commits, cancellation from SIGTERM or another pipeline stage can make a later Commit(ctx) fail with context.Canceled, manufacturing a fatal partial persist during routine shutdown. Detach the commit barrier from cancellation once it begins so the batch commits fully or not at all.

		if commitErr := siblingTxs[i].Commit(ctx); commitErr != nil {

internal/services/ingest_live.go:428

PersistHistory for SEP-41 also COPYs into state_changes, so this coordinator transaction and the state-change sibling can write the same hypertable concurrently. At a new TimescaleDB chunk, one transaction can wait on a chunk-creation lock held by the other while the commit barrier waits for both goroutines, creating an application-level deadlock PostgreSQL cannot detect. Route protocol history through the state-change sibling transaction and serialize use of that pgx.Tx.

				persistErr := processor.PersistHistory(ctx, dbTx)

internal/services/ingest_live.go:709

The rotation cannot refill a full batch while the previous full batch still owns its buffers: with only batchCap+1 buffers, process runs out before the processed queue can reach the configured cap, so sustained backlogs settle below the requested batch size. Size the rotation for both the in-flight batch and the refill path.

	freeBuffers := make(chan *indexer.IndexerBuffer, batchCap+1)
	for range batchCap + 1 {
		freeBuffers <- indexer.NewIndexerBuffer()

internal/data/ingest_store.go:296

These hypertables are partitioned by ledger_created_at; a TOID-only DELETE cannot statically exclude old uncompressed chunks, so every restart scans the full history of all five large tables even when there are no orphan rows. Resolve the cursor ledger's close time and add a monotonic ledger_created_at lower bound, retaining the TOID predicate for exactness.

			tag, execErr := dbTx.Exec(ctx, fmt.Sprintf(`DELETE FROM %s WHERE %s >= $1`, target.table, target.column), minTOID)

internal/services/ingest_live.go:958

Dividing commit wall time by batch size makes insert_into_db look healthier precisely when persist is backlogged; for example, a 10-second commit of 10 ledgers is reported as ten 1-second samples. Record one observation with the actual commit duration and use persist_batch_size to expose amortization.

		m.appMetrics.Ingestion.PhaseDuration.WithLabelValues("insert_into_db").Observe(persistShare.Seconds())

+		maps.Copy(wasms, pl.buffer.GetProtocolWasms())
+		maps.Copy(bytecodes, pl.buffer.GetProtocolWasmBytecodes())
+		maps.Copy(contracts, pl.buffer.GetProtocolContracts())


Copilot

Pull request overview

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

Suppressed comments (5)

internal/services/ingest_live.go:428

PersistHistory is still given the coordinating transaction, but the SEP-41 processor COPYs its history into state_changes (internal/services/sep41/processor.go:356-368) while the state-change sibling concurrently COPYs the same hypertable at lines 150-151. At a TimescaleDB chunk boundary, one transaction can wait on a lock held by the other, while the sibling cannot commit until this coordinator goroutine finishes, producing an application-level deadlock. Route protocol history through the state-change sibling transaction and serialize both producers while leaving the protocol cursor on the coordinator.

				persistErr := processor.PersistHistory(ctx, dbTx)

internal/services/ingest_live.go:814

The last-write-wins contract merge loses per-ledger classification history. If ledger N binds contract C to already-classified W1 and ledger N+1 rebinds C to W2, this map retains only C→W2, so the plan omits W1 (as the new LastBindingWins test asserts). Yet stageCoordinatedWrites processes ledger N with this same plan; getEffectiveProtocolContracts removes C's committed binding and cannot re-add C under W1, silently dropping N's protocol events/state. Split the batch before a binding change or retain the classification needed by each ledger.

		maps.Copy(contracts, pl.buffer.GetProtocolContracts())

internal/services/ingest_live.go:700

This pool is too small to sustain configured-size batches. While N buffers are held by the current persist, the next batch needs N−1 queued buffers plus one buffer being processed, for 2N total. With only N+1, processing stalls after one queued ledger during a full persist, so subsequent commits cannot reliably coalesce to the configured cap and the throughput feature defeats itself.

	freeBuffers := make(chan *indexer.IndexerBuffer, batchCap+1)
	for range batchCap + 1 {
		freeBuffers <- indexer.NewIndexerBuffer()

internal/services/ingest_live.go:233

The commit barrier uses the cancellable pipeline context. A SIGTERM can therefore arrive after one sibling commits and make the next Commit return context.Canceled, turning an ordinary shutdown into ErrPartialPersist and requiring startup reconciliation. Once all staging succeeds and the barrier starts, use a context detached from cancellation so the set finishes committing (or fails for an actual database error).

		if commitErr := siblingTxs[i].Commit(ctx); commitErr != nil {

internal/services/ingest_live.go:949

Dividing the commit wall time by batch size and emitting one sample per ledger hides the actual database commit latency: for example, a 10-second 10-ledger commit is reported as ten 1-second insert_into_db observations. This makes the phase p99 look healthy precisely when persistence is overloaded. Observe persistDuration once per commit; keep persistShare only for the per-ledger aggregate duration.

	persistShare := persistDuration / time.Duration(len(batch))
	for _, pl := range batch {
		m.appMetrics.Ingestion.PhaseDuration.WithLabelValues("insert_into_db").Observe(persistShare.Seconds())

…with coordinated late commits The live persist ran its three COPY families — transactions(+accounts), operations(+accounts), state_changes — sequentially inside one transaction on one connection, so the insert phase cost their sum and a single Postgres backend did all the index maintenance. They now stream concurrently on three sibling connections, each in its own transaction, while the coordinating transaction stages everything else (assets, contracts, classification, protocol state, token changes, cursor). The table sets are disjoint with no FKs among them, and every goroutine only reads the quiescent buffer. Commits are the visibility point and are held until all four transactions have done their work: siblings commit first (sub-ms each), the coordinating transaction — whose cursor is the authority on which ledgers exist — strictly last. A failure before the first commit rolls everything back and stays retryable exactly as before; a failure after it wraps the new ErrPartialPersist sentinel, which isPermanentPersistError classifies as fatal, because COPY has no ON CONFLICT and re-running the ledger would collide on primary keys. The only crash state this ordering can produce is orphaned bulk rows for the single ledger past the committed cursor, so startup runs IngestStoreModel.DeleteRowsAboveLedger before resuming: one transaction of TOID-bounded deletes (rows of ledgers > cursor are exactly rows with TOID >= toid.New(cursor+1,0,0)) across the five bulk tables, kept chunk-local by each table's chunk skipping. Backfill keeps the single-transaction insertIntoDB path unchanged.

…ommits When the process stage finishes ledgers faster than persist drains them, the persist stage now folds up to --live-persist-max-batch-size (default 5) consecutive ledgers into one commit set: each sibling COPY streams the whole batch on its connection, the coordinating transaction stages the batch's ledgers in order, and the commit barrier fires once — amortizing COPY setup, index-page churn, and fsyncs across the backlog. While the pipeline keeps pace every batch has size 1 and behavior is unchanged. A ledger with classification inputs always opens its own batch: its plan's pool reads see exactly what the previous batch committed, which preserves the deployed-contract-sees-prior-wasm invariant. The cursor stays the authority — one guarded update lands on the batch's last ledger, and a pre-commit failure rolls back and retries the whole batch. wallet_ingestion_persist_batch_size observes coalescing; per-ledger duration histograms record each ledger's amortized share so panels stay comparable across batch sizes.

…on WAL flush SET LOCAL synchronous_commit = off on each sibling session removes up to three serialized fsync waits per persist. Durability is unchanged: the coordinating transaction commits synchronously and strictly last, and its flush covers all earlier WAL including the sibling commit records, so a durable cursor implies durable siblings; rows a crash could lose are exactly the unacknowledged ones startup reconciliation deletes.

Batching is opt-in per deployment: at mainnet/testnet cadence the close time comfortably exceeds the persist time, so every ledger commits on its own; high-TPL/short-block deployments raise the flag to amortize backlogs.

… the pool BatchGetByContractIDs queried the pool, so a mid-batch ledger's membership lookup could not see protocol_contracts rows staged earlier in the same batch's uncommitted coordinating transaction. A contract deployed and classified at the batch head, then invoked mid-batch without touching its instance entry, buffered nothing on the later ledger — its events were silently dropped for event-only processors (SEP-41; Blend re-reads through the transaction and was unaffected). Unreachable at the default batch size of 1; the loadtest rig runs 3. The lookup now takes a db.Querier and live ingestion passes the coordinating transaction, mirroring GetByProtocolID.

The sibling COPY transactions commit before the coordinating transaction that carries the cursor, so the bulk tables hold rows for ledgers that are not yet part of the served chain: briefly between those commits, and after a crash until startup reconciliation runs. No read query bounded on the cursor, so those rows were served. A transaction whose state changes had not landed looked like a transaction with none — an empty list a client cannot distinguish from a real one — and rows appeared, were reconciled away, then reappeared on re-ingestion. appendIngestCursorBound bounds a TOID column by the cursor, applied at the five read paths a client reaches directly: transactions and operations by key, and the account-scoped transaction, operation and state-change lists. Everything else in the read layer is a dataloader keyed by a parent row's TOID, so a bounded root leaves its children bounded too. The bound resolves the cursor inline rather than taking it from Go: one snapshot covers both the cursor and the rows, where two reads could disagree. Measured on dev mainnet (62GB, 543M rows, 30 daily chunks): +0.06ms and two buffers — one ingest_store PK probe as an InitPlan, evaluated once per execution — with the plan shape and chunk counts unchanged. The subquery aggregates so it returns one never-NULL row, keeping the comparison a bare scalar that pushes into the rowstore index condition and the compressed batches' min/max metadata; an outer COALESCE keeps the index condition but loses metadata pruning on all 28 compressed chunks.

The batch cut existed because the classification plan was built per ledger. A plan resolves a contract's wasm verdict from protocol_wasms when that wasm was uploaded in an earlier ledger — a pool read that cannot see rows the current batch has staged but not committed. So a ledger carrying classification inputs had to open its own batch, and proving which ledgers could safely ride mid-batch needed seen-sets of every wasm and contract the process had ever committed, a size cap because that state was unbounded, and a gate to stop the default batch cap of 1 paying for any of it. One plan now covers the whole batch, built from the union of its ledgers' buffered wasms, bytecodes and contracts. prepareClassificationPlan already classifies a contract's wasm from buffered bytecode when both arrive in the same call — its own thisBatch set — so supplying them together removes the database read the cut was protecting. Every batch is safe to form whatever its ledgers carry, and batches now reach the configured cap instead of settling below it. The plan moves off persistItem to a parameter of persistLedgerData, and its validator side writes are applied once per batch rather than per ledger. That is the same point in the write order as before — Apply always ran before any of the batch's wasm rows landed, and its writes hold no foreign key into protocol_wasms. The persist stage is also split into named steps: nextPersistBatch (which turns the labelled drain loop into a function, so `break drain` becomes a return and the pending slice no longer outlives an iteration), classifyBatch, persistBatch and recordBatchPersisted. Tests: the merge property is pinned by asserting the batch's own upload never reaches GetClassifiedByHashes, plus last-binding-wins for a contract rebound inside one batch. Test_persistBatchCut is removed with the cut.

Copilot

Pull request overview

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

Suppressed comments (9)

internal/services/ingest_live.go:430

PersistHistory uses the coordinating transaction, but SEP-41's implementation also BatchCopys into state_changes while the sibling transaction is writing that same hypertable. At a new TimescaleDB chunk boundary, one backend can wait on chunk creation while the coordinator waits for the sibling at the barrier, causing ingestion to hang. Route protocol-history rows through the state_changes sibling and serialize both producers on that one transaction.

			if historySwapped {
				persistStart := time.Now()
				persistErr := processor.PersistHistory(ctx, dbTx)
				m.appMetrics.Ingestion.ProtocolStateProcessingDuration.WithLabelValues(protocolID, "persist_history").Observe(time.Since(persistStart).Seconds())
				if persistErr != nil {

internal/services/ingest_live.go:233

A shutdown can cancel ctx after one sibling has committed; the next Commit(ctx) then fails because of cancellation and turns a normal restart into a fatal partial persist. Once the barrier starts, use a non-cancellable context for every sibling and coordinator commit so the set finishes consistently.

	for i, s := range siblings {
		if commitErr := siblingTxs[i].Commit(ctx); commitErr != nil {

internal/services/ingest_live.go:815

This last-write-wins map drops earlier bindings for the same contract. If ledger N binds a new contract to an already-classified W1 and N+1 rebinds it to W2, only W2 is looked up by prepareClassificationPlan; while staging N, classification lacks W1, so getEffectiveProtocolContracts drops the contract and its protocol events/state. Preserve every encountered binding hash for classification, or split the batch at rebinding, while retaining the final binding only for persisted end state.

	for _, pl := range batch {
		maps.Copy(wasms, pl.buffer.GetProtocolWasms())
		maps.Copy(bytecodes, pl.buffer.GetProtocolWasmBytecodes())
		maps.Copy(contracts, pl.buffer.GetProtocolContracts())
	}

internal/services/ingest_live.go:700

With a cap of N, persist can hold N buffers while process needs N more to build the next backlog (N-1 queued plus one being filled). Allocating only N+1 leaves process starved during a full persist, so subsequent batches collapse toward size 1 and the batching option cannot provide sustained throughput. Size the rotation to 2*batchCap.

	freeBuffers := make(chan *indexer.IndexerBuffer, batchCap+1)
	for range batchCap + 1 {
		freeBuffers <- indexer.NewIndexerBuffer()

internal/services/ingest_live.go:949

Dividing commit wall time by batch size and recording N samples hides an overloaded persist stage: a 4-second commit of four ledgers appears as four 1-second observations. PhaseDuration is used to grade the slowest stage against ledger close time, so record the actual persistDuration once per commit; keep persistShare only for the per-ledger aggregate duration.

func (m *ingestService) recordBatchPersisted(ctx context.Context, batch []processedLedger, classifyDuration, persistDuration time.Duration, freeBuffers chan<- *indexer.IndexerBuffer, latestIngested *atomic.Uint32) {
	classifyShare := classifyDuration / time.Duration(len(batch))
	persistShare := persistDuration / time.Duration(len(batch))
	for _, pl := range batch {
		m.appMetrics.Ingestion.PhaseDuration.WithLabelValues("insert_into_db").Observe(persistShare.Seconds())

internal/data/ingest_store.go:298

These deletes only constrain the TOID column, while all five hypertables are partitioned by ledger_created_at. Startup reconciliation therefore has to consider historical chunks on every restart, which becomes a long startup pause on a deep database. Resolve the cursor ledger's close time and include it as a partition-column lower bound in each delete so old chunks are excluded directly.

		for _, target := range targets {
			start := time.Now()
			tag, execErr := dbTx.Exec(ctx, fmt.Sprintf(`DELETE FROM %s WHERE %s >= $1`, target.table, target.column), minTOID)
			m.Metrics.QueryDuration.WithLabelValues("DeleteRowsAboveLedger", target.table).Observe(time.Since(start).Seconds())
			m.Metrics.QueriesTotal.WithLabelValues("DeleteRowsAboveLedger", target.table).Inc()

internal/services/ingest_live.go:545

This comment still says a crash can orphan at most one ledger, but a partial commit now leaves the entire coalesced batch above the cursor. Describe the bound as up to livePersistMaxBatchSize ledgers so operators understand the cleanup scope when batching is enabled.

		// Remove any bulk rows a crashed run left above the cursor before that
		// ledger is re-ingested: persistLedgerData commits the sibling COPY
		// transactions before the coordinating transaction that carries the
		// cursor, so a crash between those commits orphans (at most) the single
		// ledger past the cursor. Fatal on failure — ingesting over the orphans
		// would collide on the bulk tables' primary keys anyway.

internal/data/ingest_store.go:276

With batching enabled, a crash between sibling and coordinator commits can leave all ledgers in that batch above the cursor, not only one ledger. Update this API documentation to reflect that reconciliation may remove up to one full persist batch.

// DeleteRowsAboveLedger removes every row belonging to a ledger past the given
// one from the five bulk-COPY tables, in one transaction. It is live
// ingestion's startup reconciliation: sibling COPY transactions commit before
// the coordinating transaction that carries the cursor, so a crash between
// those commits leaves orphaned bulk rows for (at most) the single ledger past
// the committed cursor — and they must be cleared before that ledger is

internal/data/operations.go:28

The transaction and state-change cursor bounds have regression tests that seed rows on both sides of the cursor, but neither new operation bound is exercised: existing operation tests run without a cursor row and only verify the unbounded fallback. Add the same at/above-cursor coverage for GetByID and BatchGetByAccountAddress so a misplaced predicate cannot expose partially committed operation rows.

	var queryBuilder strings.Builder
	fmt.Fprintf(&queryBuilder, `SELECT %s FROM operations WHERE id = $1`, columns)
	appendIngestCursorBound(&queryBuilder, "id")

+func (m *IngestStoreModel) DeleteRowsAboveLedger(ctx context.Context, ledger uint32) error {
+	minTOID := toid.New(int32(ledger+1), 0, 0).ToInt64()


aditya1702 mentioned this pull request Aug 26, 2026

Stream bulk COPYs on sibling connections and coalesce persist backlogs #707

Closed

aditya1702 force-pushed the persist/4-parallel-writes branch 2 times, most recently from ca1e434 to 090ee0d Compare August 27, 2026 13:54

Copilot AI balanced review requested due to automatic review settings August 27, 2026 17:18

Copilot started reviewing on behalf of aditya1702 August 27, 2026 17:19 View session

aditya1702 force-pushed the persist/4-parallel-writes branch from b53c0dc to 35089da Compare August 27, 2026 17:24

Copilot AI reviewed Aug 27, 2026

View reviewed changes

Copilot AI review requested due to automatic review settings August 27, 2026 17:24

Copilot started reviewing on behalf of aditya1702 August 27, 2026 17:24 View session

Copilot AI reviewed Aug 27, 2026

View reviewed changes

Copilot AI review requested due to automatic review settings August 27, 2026 20:41

aditya1702 force-pushed the persist/4-parallel-writes branch from 35089da to 489bcbf Compare August 27, 2026 20:41

Copilot started reviewing on behalf of aditya1702 August 27, 2026 20:42 View session

Copilot AI reviewed Aug 27, 2026

View reviewed changes

aditya1702 requested a review from aristidesstaffieri August 28, 2026 14:34

aristidesstaffieri reviewed Aug 28, 2026

View reviewed changes

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

aditya1702 force-pushed the persist/4-parallel-writes branch from 489bcbf to 2fefea4 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

Comment on lines +821 to +823

maps.Copy(wasms, pl.buffer.GetProtocolWasms())

maps.Copy(bytecodes, pl.buffer.GetProtocolWasmBytecodes())

maps.Copy(contracts, pl.buffer.GetProtocolContracts())

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

aditya1702 force-pushed the persist/4-parallel-writes branch from 2fefea4 to 1b28f26 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

aditya1702 added 2 commits September 9, 2026 17:04

aditya1702 added 5 commits September 9, 2026 17:04

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

aditya1702 force-pushed the persist/4-parallel-writes branch from 1b28f26 to 0422d9b 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

Comment thread internal/data/ingest_store.go

Comment on lines +281 to +282

func (m *IngestStoreModel) DeleteRowsAboveLedger(ctx context.Context, ledger uint32) error {

minTOID := toid.New(int32(ledger+1), 0, 0).ToInt64()

aditya1702 self-assigned this Sep 12, 2026

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

Navigation Menu

Uh oh!

Write the big tables on separate database connections, in batches - #711

Write the big tables on separate database connections, in batches#711
aditya1702 wants to merge 7 commits into
persist/3-pipelinefrom
persist/4-parallel-writes

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!

Copilot AI left a comment

Uh oh!

aristidesstaffieri Aug 28, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		// cursor, so a crash between those commits orphans (at most) the single
		// ledger past the cursor. Fatal on failure — ingesting over the orphans

		// those commits leaves orphaned bulk rows for (at most) the single ledger past
		// the committed cursor — and they must be cleared before that ledger is

		func (m *ingestService) persistLedgerData(ctx context.Context, items []persistItem) error {
		label := batchLabel(items)

		func (m *IngestStoreModel) DeleteRowsAboveLedger(ctx context.Context, ledger uint32) error {
		minTOID := toid.New(int32(ledger+1), 0, 0).ToInt64()

Uh oh!

Conversation

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

Uh oh!

1. Big tables are written on separate DB connections

2. Backlogs can be written in batches — off by default

3. Reads are bounded by the cursor

4. synchronous_commit = off on the three table transactions

Known, handled later in the stack

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!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

aristidesstaffieri Aug 28, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

aditya1702 commented Aug 26, 2026 •

edited

Loading

4. `synchronous_commit = off` on the three table transactions