From f00fe860590c1e8fa0d313259d6bb53f908cd10d Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Mon, 10 Aug 2026 23:49:18 -0400
Subject: [PATCH 01/11] perf(services): stream the accounts-link tables on
 their own COPY siblings
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The live persist path's sibling COPY streams grow from three to five:
transactions_accounts and operations_accounts — the two largest
index-maintenance payloads, each carrying an account_id-leading unique
PK — stream concurrently with their parent tables instead of serially
behind them. TransactionModel and OperationModel each split BatchCopy
into a parent-table COPY and BatchCopyAccounts for the link table, with
the link COPY's duration and batch size recorded under its own metric
labels rather than folded into the parent's. Each sibling now writes
exactly one table, so the disjointness at the commit barrier is
per-table; the backfill path calls the five inserts in the old order
and is behaviorally unchanged.

Persist holds six connections at its commit barrier; the pool default
rises to 12 so the persist path never queues on Acquire behind the
advisory-lock session or pool-side classification reads.
---
 internal/data/operations.go        | 109 ++++++++++++++++------------
 internal/data/operations_test.go   |  12 +++-
 internal/data/transactions.go      | 111 +++++++++++++++++------------
 internal/data/transactions_test.go |  12 +++-
 internal/db/db.go                  |   6 +-
 internal/services/ingest.go        |  46 +++++++++---
 internal/services/ingest_live.go   |  29 +++++---
 7 files changed, 210 insertions(+), 115 deletions(-)

diff --git a/internal/data/operations.go b/internal/data/operations.go
index e01b7fbf4..9b16d2ca1 100644
--- a/internal/data/operations.go
+++ b/internal/data/operations.go
@@ -356,6 +356,8 @@ func (m *OperationModel) BatchGetByStateChangeIDs(ctx context.Context, scToIDs [
 // BatchCopy inserts operations using pgx's binary COPY protocol.
 // Uses pgx.Tx for binary format which is faster than lib/pq's text format.
 // Uses native pgtype types for optimal performance (see https://github.com/jackc/pgx/issues/763).
+// The operations_accounts links are written separately by BatchCopyAccounts, so
+// the two tables can stream on independent connections.
 //
 // IMPORTANT: BatchCopy will FAIL if any duplicate records exist. The PostgreSQL COPY
 // protocol does not support conflict handling. Callers must ensure no duplicates exist
@@ -364,7 +366,6 @@ func (m *OperationModel) BatchCopy(
 	ctx context.Context,
 	pgxTx pgx.Tx,
 	operations []*types.Operation,
-	stellarAddressesByOpID map[int64]map[string]struct{},
 ) (int, error) {
 	if len(operations) == 0 {
 		return 0, nil
@@ -407,55 +408,73 @@ func (m *OperationModel) BatchCopy(
 		return 0, fmt.Errorf("expected %d rows copied, got %d", len(operations), copyCount)
 	}
 
-	// COPY operations_accounts using pgx binary format with native pgtype types. Upstream participants handling ensures that
-	// account address is not NULL here.
-	if len(stellarAddressesByOpID) > 0 {
-		// Build OpID -> LedgerCreatedAt lookup from operations
-		ledgerCreatedAtByOpID := make(map[int64]time.Time, len(operations))
-		for _, op := range operations {
-			ledgerCreatedAtByOpID[op.ID] = op.LedgerCreatedAt
-		}
+	duration := time.Since(start).Seconds()
+	m.Metrics.QueryDuration.WithLabelValues("BatchCopy", "operations").Observe(duration)
+	m.Metrics.BatchSize.WithLabelValues("BatchCopy", "operations").Observe(float64(len(operations)))
+	m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "operations").Inc()
 
-		var oaRows [][]any
-		for opID, addresses := range stellarAddressesByOpID {
-			ledgerCreatedAt := ledgerCreatedAtByOpID[opID]
-			ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
-			opIDPgtype := pgtype.Int8{Int64: opID, Valid: true}
-			for addr := range addresses {
-				addrBytes, addrErr := types.AddressBytea(addr).Value()
-				if addrErr != nil {
-					return 0, fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
-				}
-				oaRows = append(oaRows, []any{
-					ledgerCreatedAtPgtype,
-					opIDPgtype,
-					addrBytes,
-				})
-			}
-		}
+	return len(operations), nil
+}
 
-		_, err = pgxTx.CopyFrom(
-			ctx,
-			pgx.Identifier{"operations_accounts"},
-			[]string{"ledger_created_at", "operation_id", "account_id"},
-			pgx.CopyFromRows(oaRows),
-		)
-		if err != nil {
-			duration := time.Since(start).Seconds()
-			m.Metrics.QueryDuration.WithLabelValues("BatchCopy", "operations").Observe(duration)
-			m.Metrics.BatchSize.WithLabelValues("BatchCopy", "operations").Observe(float64(len(operations)))
-			m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "operations").Inc()
-			m.Metrics.QueryErrors.WithLabelValues("BatchCopy", "operations_accounts", utils.GetDBErrorType(err)).Inc()
-			return 0, fmt.Errorf("pgx CopyFrom operations_accounts: %w", err)
-		}
+// BatchCopyAccounts inserts the operations_accounts links using pgx's binary
+// COPY protocol, and returns the number of link rows written. operations
+// supplies each operation's ledger_created_at, the link table's partition
+// column; the two arguments come from the same buffer and are read only.
+//
+// IMPORTANT: like BatchCopy, this FAILS on duplicates — COPY has no conflict
+// handling.
+func (m *OperationModel) BatchCopyAccounts(
+	ctx context.Context,
+	pgxTx pgx.Tx,
+	operations []*types.Operation,
+	stellarAddressesByOpID map[int64]map[string]struct{},
+) (int, error) {
+	if len(stellarAddressesByOpID) == 0 {
+		return 0, nil
+	}
+
+	start := time.Now()
 
-		m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "operations_accounts").Inc()
+	// Build OpID -> LedgerCreatedAt lookup from operations
+	ledgerCreatedAtByOpID := make(map[int64]time.Time, len(operations))
+	for _, op := range operations {
+		ledgerCreatedAtByOpID[op.ID] = op.LedgerCreatedAt
+	}
+
+	// COPY operations_accounts using pgx binary format with native pgtype types. Upstream participants handling ensures
+	// that account address is not NULL here.
+	var oaRows [][]any
+	for opID, addresses := range stellarAddressesByOpID {
+		ledgerCreatedAt := ledgerCreatedAtByOpID[opID]
+		ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
+		opIDPgtype := pgtype.Int8{Int64: opID, Valid: true}
+		for addr := range addresses {
+			addrBytes, addrErr := types.AddressBytea(addr).Value()
+			if addrErr != nil {
+				return 0, fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
+			}
+			oaRows = append(oaRows, []any{
+				ledgerCreatedAtPgtype,
+				opIDPgtype,
+				addrBytes,
+			})
+		}
 	}
 
+	_, err := pgxTx.CopyFrom(
+		ctx,
+		pgx.Identifier{"operations_accounts"},
+		[]string{"ledger_created_at", "operation_id", "account_id"},
+		pgx.CopyFromRows(oaRows),
+	)
 	duration := time.Since(start).Seconds()
-	m.Metrics.QueryDuration.WithLabelValues("BatchCopy", "operations").Observe(duration)
-	m.Metrics.BatchSize.WithLabelValues("BatchCopy", "operations").Observe(float64(len(operations)))
-	m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "operations").Inc()
+	m.Metrics.QueryDuration.WithLabelValues("BatchCopyAccounts", "operations_accounts").Observe(duration)
+	m.Metrics.BatchSize.WithLabelValues("BatchCopyAccounts", "operations_accounts").Observe(float64(len(oaRows)))
+	m.Metrics.QueriesTotal.WithLabelValues("BatchCopyAccounts", "operations_accounts").Inc()
+	if err != nil {
+		m.Metrics.QueryErrors.WithLabelValues("BatchCopyAccounts", "operations_accounts", utils.GetDBErrorType(err)).Inc()
+		return 0, fmt.Errorf("pgx CopyFrom operations_accounts: %w", err)
+	}
 
-	return len(operations), nil
+	return len(oaRows), nil
 }
diff --git a/internal/data/operations_test.go b/internal/data/operations_test.go
index 282133f2e..2d49d1981 100644
--- a/internal/data/operations_test.go
+++ b/internal/data/operations_test.go
@@ -147,7 +147,10 @@ func Test_OperationModel_BatchCopy(t *testing.T) {
 			pgxTx, err := conn.Begin(ctx)
 			require.NoError(t, err)
 
-			gotCount, err := m.BatchCopy(ctx, pgxTx, tc.operations, tc.stellarAddressesByOpID)
+			gotCount, err := m.BatchCopy(ctx, pgxTx, tc.operations)
+			if err == nil {
+				_, err = m.BatchCopyAccounts(ctx, pgxTx, tc.operations, tc.stellarAddressesByOpID)
+			}
 
 			if tc.wantErrContains != "" {
 				require.Error(t, err)
@@ -739,11 +742,16 @@ func BenchmarkOperationModel_BatchCopy(b *testing.B) {
 				}
 				b.StartTimer()
 
-				_, err = m.BatchCopy(ctx, pgxTx, ops, addressesByOpID)
+				_, err = m.BatchCopy(ctx, pgxTx, ops)
 				if err != nil {
 					pgxTx.Rollback(ctx)
 					b.Fatalf("BatchCopy failed: %v", err)
 				}
+				_, err = m.BatchCopyAccounts(ctx, pgxTx, ops, addressesByOpID)
+				if err != nil {
+					pgxTx.Rollback(ctx)
+					b.Fatalf("BatchCopyAccounts failed: %v", err)
+				}
 
 				b.StopTimer()
 				if err := pgxTx.Commit(ctx); err != nil {
diff --git a/internal/data/transactions.go b/internal/data/transactions.go
index a8ce68374..c6e2ea3bd 100644
--- a/internal/data/transactions.go
+++ b/internal/data/transactions.go
@@ -204,6 +204,8 @@ func (m *TransactionModel) BatchGetByStateChangeIDs(ctx context.Context, scToIDs
 // BatchCopy inserts transactions using pgx's binary COPY protocol.
 // Uses pgx.Tx for binary format which is faster than lib/pq's text format.
 // Uses native pgtype types for optimal performance (see https://github.com/jackc/pgx/issues/763).
+// The transactions_accounts links are written separately by BatchCopyAccounts,
+// so the two tables can stream on independent connections.
 //
 // IMPORTANT: BatchCopy will FAIL if any duplicate records exist. The PostgreSQL COPY
 // protocol does not support conflict handling. Callers must ensure no duplicates exist
@@ -212,7 +214,6 @@ func (m *TransactionModel) BatchCopy(
 	ctx context.Context,
 	pgxTx pgx.Tx,
 	txs []*types.Transaction,
-	stellarAddressesByToID map[int64]map[string]struct{},
 ) (int, error) {
 	if len(txs) == 0 {
 		return 0, nil
@@ -220,8 +221,7 @@ func (m *TransactionModel) BatchCopy(
 
 	start := time.Now()
 
-	// COPY transactions using pgx binary format with native pgtype types. Upstream participants handling ensures that
-	// account address is not NULL here.
+	// COPY transactions using pgx binary format with native pgtype types.
 	copyCount, err := pgxTx.CopyFrom(
 		ctx,
 		pgx.Identifier{"transactions"},
@@ -260,54 +260,73 @@ func (m *TransactionModel) BatchCopy(
 		return 0, fmt.Errorf("expected %d rows copied, got %d", len(txs), copyCount)
 	}
 
-	// COPY transactions_accounts using pgx binary format with native pgtype types
-	if len(stellarAddressesByToID) > 0 {
-		// Build ToID -> LedgerCreatedAt lookup from transactions
-		ledgerCreatedAtByToID := make(map[int64]time.Time, len(txs))
-		for _, tx := range txs {
-			ledgerCreatedAtByToID[tx.ToID] = tx.LedgerCreatedAt
-		}
+	duration := time.Since(start).Seconds()
+	m.Metrics.QueryDuration.WithLabelValues("BatchCopy", "transactions").Observe(duration)
+	m.Metrics.BatchSize.WithLabelValues("BatchCopy", "transactions").Observe(float64(len(txs)))
+	m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "transactions").Inc()
 
-		var taRows [][]any
-		for toID, addresses := range stellarAddressesByToID {
-			ledgerCreatedAt := ledgerCreatedAtByToID[toID]
-			ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
-			toIDPgtype := pgtype.Int8{Int64: toID, Valid: true}
-			for addr := range addresses {
-				addrBytes, addrErr := types.AddressBytea(addr).Value()
-				if addrErr != nil {
-					return 0, fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
-				}
-				taRows = append(taRows, []any{
-					ledgerCreatedAtPgtype,
-					toIDPgtype,
-					addrBytes,
-				})
-			}
-		}
+	return len(txs), nil
+}
 
-		_, err = pgxTx.CopyFrom(
-			ctx,
-			pgx.Identifier{"transactions_accounts"},
-			[]string{"ledger_created_at", "tx_to_id", "account_id"},
-			pgx.CopyFromRows(taRows),
-		)
-		if err != nil {
-			duration := time.Since(start).Seconds()
-			m.Metrics.QueryDuration.WithLabelValues("BatchCopy", "transactions").Observe(duration)
-			m.Metrics.BatchSize.WithLabelValues("BatchCopy", "transactions").Observe(float64(len(txs)))
-			m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "transactions").Inc()
-			m.Metrics.QueryErrors.WithLabelValues("BatchCopy", "transactions_accounts", utils.GetDBErrorType(err)).Inc()
-			return 0, fmt.Errorf("pgx CopyFrom transactions_accounts: %w", err)
-		}
+// BatchCopyAccounts inserts the transactions_accounts links using pgx's binary
+// COPY protocol, and returns the number of link rows written. txs supplies each
+// transaction's ledger_created_at, the link table's partition column; the two
+// arguments come from the same buffer and are read only.
+//
+// IMPORTANT: like BatchCopy, this FAILS on duplicates — COPY has no conflict
+// handling.
+func (m *TransactionModel) BatchCopyAccounts(
+	ctx context.Context,
+	pgxTx pgx.Tx,
+	txs []*types.Transaction,
+	stellarAddressesByToID map[int64]map[string]struct{},
+) (int, error) {
+	if len(stellarAddressesByToID) == 0 {
+		return 0, nil
+	}
+
+	start := time.Now()
 
-		m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "transactions_accounts").Inc()
+	// Build ToID -> LedgerCreatedAt lookup from transactions
+	ledgerCreatedAtByToID := make(map[int64]time.Time, len(txs))
+	for _, tx := range txs {
+		ledgerCreatedAtByToID[tx.ToID] = tx.LedgerCreatedAt
+	}
+
+	// COPY transactions_accounts using pgx binary format with native pgtype types. Upstream participants handling
+	// ensures that account address is not NULL here.
+	var taRows [][]any
+	for toID, addresses := range stellarAddressesByToID {
+		ledgerCreatedAt := ledgerCreatedAtByToID[toID]
+		ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
+		toIDPgtype := pgtype.Int8{Int64: toID, Valid: true}
+		for addr := range addresses {
+			addrBytes, addrErr := types.AddressBytea(addr).Value()
+			if addrErr != nil {
+				return 0, fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
+			}
+			taRows = append(taRows, []any{
+				ledgerCreatedAtPgtype,
+				toIDPgtype,
+				addrBytes,
+			})
+		}
 	}
 
+	_, err := pgxTx.CopyFrom(
+		ctx,
+		pgx.Identifier{"transactions_accounts"},
+		[]string{"ledger_created_at", "tx_to_id", "account_id"},
+		pgx.CopyFromRows(taRows),
+	)
 	duration := time.Since(start).Seconds()
-	m.Metrics.QueryDuration.WithLabelValues("BatchCopy", "transactions").Observe(duration)
-	m.Metrics.BatchSize.WithLabelValues("BatchCopy", "transactions").Observe(float64(len(txs)))
-	m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "transactions").Inc()
+	m.Metrics.QueryDuration.WithLabelValues("BatchCopyAccounts", "transactions_accounts").Observe(duration)
+	m.Metrics.BatchSize.WithLabelValues("BatchCopyAccounts", "transactions_accounts").Observe(float64(len(taRows)))
+	m.Metrics.QueriesTotal.WithLabelValues("BatchCopyAccounts", "transactions_accounts").Inc()
+	if err != nil {
+		m.Metrics.QueryErrors.WithLabelValues("BatchCopyAccounts", "transactions_accounts", utils.GetDBErrorType(err)).Inc()
+		return 0, fmt.Errorf("pgx CopyFrom transactions_accounts: %w", err)
+	}
 
-	return len(txs), nil
+	return len(taRows), nil
 }
diff --git a/internal/data/transactions_test.go b/internal/data/transactions_test.go
index e70b745f3..2863cba1a 100644
--- a/internal/data/transactions_test.go
+++ b/internal/data/transactions_test.go
@@ -145,7 +145,10 @@ func Test_TransactionModel_BatchCopy(t *testing.T) {
 			pgxTx, err := conn.Begin(ctx)
 			require.NoError(t, err)
 
-			gotCount, err := m.BatchCopy(ctx, pgxTx, tc.txs, tc.stellarAddressesByToID)
+			gotCount, err := m.BatchCopy(ctx, pgxTx, tc.txs)
+			if err == nil {
+				_, err = m.BatchCopyAccounts(ctx, pgxTx, tc.txs, tc.stellarAddressesByToID)
+			}
 
 			if tc.wantErrContains != "" {
 				require.Error(t, err)
@@ -463,11 +466,16 @@ func BenchmarkTransactionModel_BatchCopy(b *testing.B) {
 				}
 				b.StartTimer()
 
-				_, err = m.BatchCopy(ctx, pgxTx, txs, addressesByToID)
+				_, err = m.BatchCopy(ctx, pgxTx, txs)
 				if err != nil {
 					pgxTx.Rollback(ctx)
 					b.Fatalf("BatchCopy failed: %v", err)
 				}
+				_, err = m.BatchCopyAccounts(ctx, pgxTx, txs, addressesByToID)
+				if err != nil {
+					pgxTx.Rollback(ctx)
+					b.Fatalf("BatchCopyAccounts failed: %v", err)
+				}
 
 				b.StopTimer()
 				if err := pgxTx.Commit(ctx); err != nil {
diff --git a/internal/db/db.go b/internal/db/db.go
index 99233c0a9..abe64a700 100644
--- a/internal/db/db.go
+++ b/internal/db/db.go
@@ -17,7 +17,11 @@ import (
 
 const (
 	DefaultMaxConnIdleTime time.Duration = 10 * time.Second
-	DefaultMaxConns        int32         = 10
+	// Live persist alone holds 6 connections at its commit barrier (5 COPY
+	// siblings + the coordinator), plus the advisory-lock session and
+	// pool-side classification reads; 12 leaves headroom so no persist-path
+	// Acquire queues behind an unrelated consumer.
+	DefaultMaxConns        int32         = 12
 	DefaultMinConns        int32         = 5
 	DefaultMaxConnLifetime time.Duration = 5 * time.Minute
 )
diff --git a/internal/services/ingest.go b/internal/services/ingest.go
index b7f907a5c..81547a65c 100644
--- a/internal/services/ingest.go
+++ b/internal/services/ingest.go
@@ -276,10 +276,16 @@ func (m *ingestService) insertIntoDB(ctx context.Context, dbTx pgx.Tx, buffer in
 	opParticipants := buffer.GetOperationsParticipants()
 	stateChanges := buffer.GetStateChanges()
 
-	if err := m.insertTransactions(ctx, dbTx, txs, txParticipants); err != nil {
+	if err := m.insertTransactions(ctx, dbTx, txs); err != nil {
 		return err
 	}
-	if err := m.insertOperations(ctx, dbTx, ops, opParticipants); err != nil {
+	if err := m.insertTransactionsAccounts(ctx, dbTx, txs, txParticipants); err != nil {
+		return err
+	}
+	if err := m.insertOperations(ctx, dbTx, ops); err != nil {
+		return err
+	}
+	if err := m.insertOperationsAccounts(ctx, dbTx, ops, opParticipants); err != nil {
 		return err
 	}
 	if err := m.insertStateChanges(ctx, dbTx, stateChanges); err != nil {
@@ -289,30 +295,54 @@ func (m *ingestService) insertIntoDB(ctx context.Context, dbTx pgx.Tx, buffer in
 	return nil
 }
 
-// insertTransactions batch inserts transactions with their participants into the database.
-func (m *ingestService) insertTransactions(ctx context.Context, pgxTx pgx.Tx, txs []*types.Transaction, stellarAddressesByToID map[int64]map[string]struct{}) error {
+// insertTransactions batch inserts transactions into the database.
+func (m *ingestService) insertTransactions(ctx context.Context, pgxTx pgx.Tx, txs []*types.Transaction) error {
 	if len(txs) == 0 {
 		return nil
 	}
-	_, err := m.models.Transactions.BatchCopy(ctx, pgxTx, txs, stellarAddressesByToID)
+	_, err := m.models.Transactions.BatchCopy(ctx, pgxTx, txs)
 	if err != nil {
 		return fmt.Errorf("batch inserting transactions: %w", err)
 	}
 	return nil
 }
 
-// insertOperations batch inserts operations with their participants into the database.
-func (m *ingestService) insertOperations(ctx context.Context, pgxTx pgx.Tx, ops []*types.Operation, stellarAddressesByOpID map[int64]map[string]struct{}) error {
+// insertTransactionsAccounts batch inserts the transactions_accounts links into the database.
+func (m *ingestService) insertTransactionsAccounts(ctx context.Context, pgxTx pgx.Tx, txs []*types.Transaction, stellarAddressesByToID map[int64]map[string]struct{}) error {
+	if len(stellarAddressesByToID) == 0 {
+		return nil
+	}
+	_, err := m.models.Transactions.BatchCopyAccounts(ctx, pgxTx, txs, stellarAddressesByToID)
+	if err != nil {
+		return fmt.Errorf("batch inserting transactions accounts: %w", err)
+	}
+	return nil
+}
+
+// insertOperations batch inserts operations into the database.
+func (m *ingestService) insertOperations(ctx context.Context, pgxTx pgx.Tx, ops []*types.Operation) error {
 	if len(ops) == 0 {
 		return nil
 	}
-	_, err := m.models.Operations.BatchCopy(ctx, pgxTx, ops, stellarAddressesByOpID)
+	_, err := m.models.Operations.BatchCopy(ctx, pgxTx, ops)
 	if err != nil {
 		return fmt.Errorf("batch inserting operations: %w", err)
 	}
 	return nil
 }
 
+// insertOperationsAccounts batch inserts the operations_accounts links into the database.
+func (m *ingestService) insertOperationsAccounts(ctx context.Context, pgxTx pgx.Tx, ops []*types.Operation, stellarAddressesByOpID map[int64]map[string]struct{}) error {
+	if len(stellarAddressesByOpID) == 0 {
+		return nil
+	}
+	_, err := m.models.Operations.BatchCopyAccounts(ctx, pgxTx, ops, stellarAddressesByOpID)
+	if err != nil {
+		return fmt.Errorf("batch inserting operations accounts: %w", err)
+	}
+	return nil
+}
+
 // insertStateChanges batch inserts state changes and records metrics.
 func (m *ingestService) insertStateChanges(ctx context.Context, pgxTx pgx.Tx, stateChanges []types.StateChange) error {
 	if len(stateChanges) == 0 {
diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index dc818f185..0a9054a75 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -101,10 +101,10 @@ func batchLabel(items []persistItem) string {
 }
 
 // persistLedgerData persists a batch of consecutive ledgers in one commit
-// set. The three bulk COPY families — transactions(+accounts),
-// operations(+accounts), state_changes — stream concurrently on sibling
-// connections, each in its own transaction covering every ledger in the
-// batch, while the coordinating transaction stages everything else (assets,
+// set. The five bulk COPY families — transactions, transactions_accounts,
+// operations, operations_accounts, state_changes — stream concurrently on
+// sibling connections, each in its own transaction covering every ledger in
+// the batch, while the coordinating transaction stages everything else (assets,
 // contracts, classification, protocol state, token changes, cursor) ledger
 // by ledger in order — the per-protocol CAS chain advances N-1 → N inside
 // the transaction, and the guarded cursor's final value is the batch's last
@@ -142,10 +142,16 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 		run  func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error
 	}{
 		{"transactions", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
-			return m.insertTransactions(ctx, dbTx, it.buffer.GetTransactions(), it.buffer.GetTransactionsParticipants())
+			return m.insertTransactions(ctx, dbTx, it.buffer.GetTransactions())
+		}},
+		{"transactions_accounts", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
+			return m.insertTransactionsAccounts(ctx, dbTx, it.buffer.GetTransactions(), it.buffer.GetTransactionsParticipants())
 		}},
 		{"operations", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
-			return m.insertOperations(ctx, dbTx, it.buffer.GetOperations(), it.buffer.GetOperationsParticipants())
+			return m.insertOperations(ctx, dbTx, it.buffer.GetOperations())
+		}},
+		{"operations_accounts", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
+			return m.insertOperationsAccounts(ctx, dbTx, it.buffer.GetOperations(), it.buffer.GetOperationsParticipants())
 		}},
 		{"state_changes", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
 			return m.insertStateChanges(ctx, dbTx, it.buffer.GetStateChanges())
@@ -181,9 +187,10 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 	}
 
 	// Stream the COPY families and stage the coordinated writes concurrently.
-	// The table sets are disjoint (no FKs among them), and the goroutines only
-	// read the quiescent buffers, so the four transactions never contend.
-	// Within each transaction the batch's ledgers run in order.
+	// Every sibling owns exactly one table and the table sets are disjoint (no
+	// FKs among them), and the goroutines only read the quiescent buffers, so
+	// the six transactions never contend. Within each transaction the batch's
+	// ledgers run in order.
 	g, gctx := errgroup.WithContext(ctx)
 	for i, s := range siblings {
 		g.Go(func() error {
@@ -205,7 +212,7 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 		return nil
 	})
 	if err = g.Wait(); err != nil {
-		// Nothing has committed: the deferred rollbacks discard all four
+		// Nothing has committed: the deferred rollbacks discard all six
 		// transactions and the batch is cleanly retryable.
 		return fmt.Errorf("persisting ledger data for %s: %w", label, err)
 	}
@@ -233,7 +240,7 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 	return nil
 }
 
-// stageCoordinatedWrites runs every per-ledger write except the three bulk
+// stageCoordinatedWrites runs every per-ledger write except the five bulk
 // COPY families on the coordinating transaction: trustline assets, SAC
 // contract tokens, protocol classification and wasm/contract rows,
 // CAS-gated protocol state, token changes, and finally the guarded cursor.

From fc0450608bc14e84b47cdc964e3accccac3eae35 Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Mon, 10 Aug 2026 23:51:43 -0400
Subject: [PATCH 02/11] perf(data): stop the balance upserts from rewriting
 unchanged rows in random order

The native_balances and trustline_balances batch upserts arrive in Go
map-iteration order and unconditionally rewrite every conflicting row.
Both cost heap buffer touches far above what the row count warrants and
bloat the table with dead tuples. Two changes: rows sort by the primary-key columns before the
UNNEST arrays are built, so the btree descent and heap touches run in
key order; and the DO UPDATE carries an IS DISTINCT FROM guard, so an
identical row produces no new tuple version, no dead tuple, no WAL,
and no index churn.
---
 internal/data/native_balances.go    | 56 +++++++++++++++++--------
 internal/data/trustline_balances.go | 64 +++++++++++++++++++++--------
 2 files changed, 86 insertions(+), 34 deletions(-)

diff --git a/internal/data/native_balances.go b/internal/data/native_balances.go
index 104992685..d78837773 100644
--- a/internal/data/native_balances.go
+++ b/internal/data/native_balances.go
@@ -3,9 +3,11 @@
 package data
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -85,14 +87,11 @@ func (m *NativeBalanceModel) BatchUpsert(ctx context.Context, dbTx pgx.Tx, upser
 	start := time.Now()
 
 	if len(upserts) > 0 {
-		accountIDs := make([][]byte, len(upserts))
-		balances := make([]int64, len(upserts))
-		minimumBalances := make([]int64, len(upserts))
-		buyingLiabilities := make([]int64, len(upserts))
-		sellingLiabilities := make([]int64, len(upserts))
-		numSubentries := make([]int32, len(upserts))
-		ledgerNumbers := make([]int64, len(upserts))
-
+		type upsertRow struct {
+			accountID []byte
+			nb        NativeBalance
+		}
+		rows := make([]upsertRow, len(upserts))
 		for i, nb := range upserts {
 			raw, err := nb.AccountID.Value()
 			if err != nil {
@@ -102,15 +101,35 @@ func (m *NativeBalanceModel) BatchUpsert(ctx context.Context, dbTx pgx.Tx, upser
 			if !ok {
 				return fmt.Errorf("converting account address to bytes for upsert: expected []byte, got %T", raw)
 			}
-			accountIDs[i] = rawBytes
-			balances[i] = nb.Balance
-			minimumBalances[i] = nb.MinimumBalance
-			buyingLiabilities[i] = nb.BuyingLiabilities
-			sellingLiabilities[i] = nb.SellingLiabilities
-			numSubentries[i] = int32(nb.NumSubEntries)
-			ledgerNumbers[i] = int64(nb.LedgerNumber)
+			rows[i] = upsertRow{accountID: rawBytes, nb: nb}
+		}
+		// Upserts arrive in Go map-iteration order; sorting by the PK column
+		// descends the btree and touches heap pages in key order instead of
+		// scattering thousands of random probes across the relation.
+		slices.SortFunc(rows, func(a, b upsertRow) int {
+			return bytes.Compare(a.accountID, b.accountID)
+		})
+
+		accountIDs := make([][]byte, len(rows))
+		balances := make([]int64, len(rows))
+		minimumBalances := make([]int64, len(rows))
+		buyingLiabilities := make([]int64, len(rows))
+		sellingLiabilities := make([]int64, len(rows))
+		numSubentries := make([]int32, len(rows))
+		ledgerNumbers := make([]int64, len(rows))
+		for i, row := range rows {
+			accountIDs[i] = row.accountID
+			balances[i] = row.nb.Balance
+			minimumBalances[i] = row.nb.MinimumBalance
+			buyingLiabilities[i] = row.nb.BuyingLiabilities
+			sellingLiabilities[i] = row.nb.SellingLiabilities
+			numSubentries[i] = int32(row.nb.NumSubEntries)
+			ledgerNumbers[i] = int64(row.nb.LedgerNumber)
 		}
 
+		// The WHERE clause turns updates that would rewrite an identical row
+		// into no-ops: no new tuple version, no dead tuple, no WAL, no index
+		// churn — the update fires only when some column actually differs.
 		const upsertQuery = `
 			INSERT INTO native_balances (account_id, balance, minimum_balance, buying_liabilities, selling_liabilities, num_subentries, last_modified_ledger)
 			SELECT * FROM UNNEST($1::bytea[], $2::bigint[], $3::bigint[], $4::bigint[], $5::bigint[], $6::int[], $7::bigint[])
@@ -120,7 +139,12 @@ func (m *NativeBalanceModel) BatchUpsert(ctx context.Context, dbTx pgx.Tx, upser
 				buying_liabilities = EXCLUDED.buying_liabilities,
 				selling_liabilities = EXCLUDED.selling_liabilities,
 				num_subentries = EXCLUDED.num_subentries,
-				last_modified_ledger = EXCLUDED.last_modified_ledger`
+				last_modified_ledger = EXCLUDED.last_modified_ledger
+			WHERE (native_balances.balance, native_balances.minimum_balance, native_balances.buying_liabilities,
+			       native_balances.selling_liabilities, native_balances.num_subentries, native_balances.last_modified_ledger)
+			      IS DISTINCT FROM
+			      (EXCLUDED.balance, EXCLUDED.minimum_balance, EXCLUDED.buying_liabilities,
+			       EXCLUDED.selling_liabilities, EXCLUDED.num_subentries, EXCLUDED.last_modified_ledger)`
 
 		if _, err := dbTx.Exec(ctx, upsertQuery, accountIDs, balances, minimumBalances, buyingLiabilities, sellingLiabilities, numSubentries, ledgerNumbers); err != nil {
 			m.Metrics.QueryDuration.WithLabelValues("BatchUpsert", "native_balances").Observe(time.Since(start).Seconds())
diff --git a/internal/data/trustline_balances.go b/internal/data/trustline_balances.go
index 9e19dd93d..b7ee7fc72 100644
--- a/internal/data/trustline_balances.go
+++ b/internal/data/trustline_balances.go
@@ -3,8 +3,10 @@
 package data
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"slices"
 	"time"
 
 	"github.com/google/uuid"
@@ -118,15 +120,11 @@ func (m *TrustlineBalanceModel) BatchUpsert(ctx context.Context, dbTx pgx.Tx, up
 	start := time.Now()
 
 	if len(upserts) > 0 {
-		accountIDs := make([][]byte, len(upserts))
-		assetIDs := make([]uuid.UUID, len(upserts))
-		balances := make([]int64, len(upserts))
-		limits := make([]int64, len(upserts))
-		buyingLiabilities := make([]int64, len(upserts))
-		sellingLiabilities := make([]int64, len(upserts))
-		flags := make([]int32, len(upserts))
-		ledgerNumbers := make([]int64, len(upserts))
-
+		type upsertRow struct {
+			accountID []byte
+			tl        TrustlineBalance
+		}
+		rows := make([]upsertRow, len(upserts))
 		for i, tl := range upserts {
 			raw, err := tl.AccountID.Value()
 			if err != nil {
@@ -136,16 +134,41 @@ func (m *TrustlineBalanceModel) BatchUpsert(ctx context.Context, dbTx pgx.Tx, up
 			if !ok {
 				return fmt.Errorf("converting account address to bytes for upsert: expected []byte, got %T", raw)
 			}
-			accountIDs[i] = rawBytes
-			assetIDs[i] = tl.AssetID
-			balances[i] = tl.Balance
-			limits[i] = tl.Limit
-			buyingLiabilities[i] = tl.BuyingLiabilities
-			sellingLiabilities[i] = tl.SellingLiabilities
-			flags[i] = int32(tl.Flags)
-			ledgerNumbers[i] = int64(tl.LedgerNumber)
+			rows[i] = upsertRow{accountID: rawBytes, tl: tl}
+		}
+		// Upserts arrive in Go map-iteration order; sorting by the PK's
+		// leading columns descends the btree and touches heap pages in key
+		// order instead of scattering thousands of random probes across the
+		// relation.
+		slices.SortFunc(rows, func(a, b upsertRow) int {
+			if c := bytes.Compare(a.accountID, b.accountID); c != 0 {
+				return c
+			}
+			return bytes.Compare(a.tl.AssetID[:], b.tl.AssetID[:])
+		})
+
+		accountIDs := make([][]byte, len(rows))
+		assetIDs := make([]uuid.UUID, len(rows))
+		balances := make([]int64, len(rows))
+		limits := make([]int64, len(rows))
+		buyingLiabilities := make([]int64, len(rows))
+		sellingLiabilities := make([]int64, len(rows))
+		flags := make([]int32, len(rows))
+		ledgerNumbers := make([]int64, len(rows))
+		for i, row := range rows {
+			accountIDs[i] = row.accountID
+			assetIDs[i] = row.tl.AssetID
+			balances[i] = row.tl.Balance
+			limits[i] = row.tl.Limit
+			buyingLiabilities[i] = row.tl.BuyingLiabilities
+			sellingLiabilities[i] = row.tl.SellingLiabilities
+			flags[i] = int32(row.tl.Flags)
+			ledgerNumbers[i] = int64(row.tl.LedgerNumber)
 		}
 
+		// The WHERE clause turns updates that would rewrite an identical row
+		// into no-ops: no new tuple version, no dead tuple, no WAL, no index
+		// churn — the update fires only when some column actually differs.
 		const upsertQuery = `
 			INSERT INTO trustline_balances (
 				account_id, asset_id, balance, trust_limit,
@@ -158,7 +181,12 @@ func (m *TrustlineBalanceModel) BatchUpsert(ctx context.Context, dbTx pgx.Tx, up
 				buying_liabilities = EXCLUDED.buying_liabilities,
 				selling_liabilities = EXCLUDED.selling_liabilities,
 				flags = EXCLUDED.flags,
-				last_modified_ledger = EXCLUDED.last_modified_ledger`
+				last_modified_ledger = EXCLUDED.last_modified_ledger
+			WHERE (trustline_balances.balance, trustline_balances.trust_limit, trustline_balances.buying_liabilities,
+			       trustline_balances.selling_liabilities, trustline_balances.flags, trustline_balances.last_modified_ledger)
+			      IS DISTINCT FROM
+			      (EXCLUDED.balance, EXCLUDED.trust_limit, EXCLUDED.buying_liabilities,
+			       EXCLUDED.selling_liabilities, EXCLUDED.flags, EXCLUDED.last_modified_ledger)`
 
 		if _, err := dbTx.Exec(ctx, upsertQuery, accountIDs, assetIDs, balances, limits, buyingLiabilities, sellingLiabilities, flags, ledgerNumbers); err != nil {
 			m.Metrics.QueryDuration.WithLabelValues("BatchUpsert", "trustline_balances").Observe(time.Since(start).Seconds())

From 4b8193493d49cbc873f20af64b50bd2ab4662926 Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Mon, 10 Aug 2026 23:57:51 -0400
Subject: [PATCH 03/11] fix(indexer): key the buffer's canonical transactions
 by ToID, not hash
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The buffer stored transactions keyed by hash while tracking their
participants keyed by ToID. A merged ledger can carry the same envelope
at several tx-set positions — distinct ToIDs, one hash — and the two maps
then disagreed:
duplicate transactions were silently dropped from the transactions COPY
while their participant links survived, whose ledger_created_at lookup
missed and COPYed the zero timestamp into a year-0001 chunk (observed
on the rig: 6 orphaned transactions_accounts rows, 5 missing
transaction rows across bootstrap ledgers). Real networks cannot repeat
a hash within or across ledgers, so both maps now share the ToID key
domain by construction, mirroring the operations pair.

BatchCopyAccounts on both link tables now errors on a ToID/opID with no
parent row instead of silently writing a zero timestamp.
---
 internal/data/operations.go             |  8 ++++++-
 internal/data/transactions.go           |  8 ++++++-
 internal/indexer/indexer_buffer.go      | 32 +++++++++++++++----------
 internal/indexer/indexer_buffer_test.go | 23 ++++++++++++++++++
 4 files changed, 56 insertions(+), 15 deletions(-)

diff --git a/internal/data/operations.go b/internal/data/operations.go
index 9b16d2ca1..c0b017059 100644
--- a/internal/data/operations.go
+++ b/internal/data/operations.go
@@ -445,7 +445,13 @@ func (m *OperationModel) BatchCopyAccounts(
 	// that account address is not NULL here.
 	var oaRows [][]any
 	for opID, addresses := range stellarAddressesByOpID {
-		ledgerCreatedAt := ledgerCreatedAtByOpID[opID]
+		ledgerCreatedAt, ok := ledgerCreatedAtByOpID[opID]
+		if !ok {
+			// A silent miss would COPY a zero timestamp — a year-0001 chunk in
+			// the hypertable — and means the caller's operations and
+			// participants disagree about which IDs exist.
+			return 0, fmt.Errorf("no operation supplies ledger_created_at for operation_id %d", opID)
+		}
 		ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
 		opIDPgtype := pgtype.Int8{Int64: opID, Valid: true}
 		for addr := range addresses {
diff --git a/internal/data/transactions.go b/internal/data/transactions.go
index c6e2ea3bd..6780fb6ba 100644
--- a/internal/data/transactions.go
+++ b/internal/data/transactions.go
@@ -297,7 +297,13 @@ func (m *TransactionModel) BatchCopyAccounts(
 	// ensures that account address is not NULL here.
 	var taRows [][]any
 	for toID, addresses := range stellarAddressesByToID {
-		ledgerCreatedAt := ledgerCreatedAtByToID[toID]
+		ledgerCreatedAt, ok := ledgerCreatedAtByToID[toID]
+		if !ok {
+			// A silent miss would COPY a zero timestamp — a year-0001 chunk in
+			// the hypertable — and means the caller's transactions and
+			// participants disagree about which ToIDs exist.
+			return 0, fmt.Errorf("no transaction supplies ledger_created_at for to_id %d", toID)
+		}
 		ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
 		toIDPgtype := pgtype.Int8{Int64: toID, Valid: true}
 		for addr := range addresses {
diff --git a/internal/indexer/indexer_buffer.go b/internal/indexer/indexer_buffer.go
index 760b24f3f..d91a83b71 100644
--- a/internal/indexer/indexer_buffer.go
+++ b/internal/indexer/indexer_buffer.go
@@ -21,7 +21,7 @@ import (
 //
 // ARCHITECTURE:
 // 1. Canonical Storage Layer:
-//   - txByHash: Single pointer per unique transaction (keyed by hash)
+//   - txByToID: Single pointer per unique transaction (keyed by ToID)
 //   - opByID: Single pointer per unique operation (keyed by ID)
 //   - This layer owns the actual data and ensures only ONE copy exists in memory
 //
@@ -30,6 +30,13 @@ import (
 //   - participantsByOpID: Maps each operation ID to a set of participant IDs (a map used as a set)
 //   - Efficiently tracks which participants interacted with each tx/op
 //
+// Both layers of each pair share one key domain (ToID for transactions, ID for
+// operations), so a participant entry can never exist without its canonical
+// row. ToID is the only key that holds that invariant: a hash identifies an
+// envelope, not a position, so keying by hash collapses two tx-set positions
+// carrying the same envelope into one canonical row while both ToIDs keep
+// their participant entries.
+//
 // MEMORY OPTIMIZATION:
 // When multiple participants interact with the same transaction or operation, they all point
 // to the SAME canonical pointer instead of storing duplicate copies.
@@ -67,7 +74,7 @@ type ContractEventKey struct {
 }
 
 type IndexerBuffer struct {
-	txByHash                       map[string]*types.Transaction
+	txByToID                       map[int64]*types.Transaction
 	participantsByToID             map[int64]map[string]struct{}
 	opByID                         map[int64]*types.Operation
 	participantsByOpID             map[int64]map[string]struct{}
@@ -103,7 +110,7 @@ type IndexerBuffer struct {
 // All maps are pre-allocated to avoid nil map access.
 func NewIndexerBuffer() *IndexerBuffer {
 	return &IndexerBuffer{
-		txByHash:                       make(map[string]*types.Transaction),
+		txByToID:                       make(map[int64]*types.Transaction),
 		participantsByToID:             make(map[int64]map[string]struct{}),
 		opByID:                         make(map[int64]*types.Operation),
 		participantsByOpID:             make(map[int64]map[string]struct{}),
@@ -129,7 +136,7 @@ func NewIndexerBuffer() *IndexerBuffer {
 }
 
 // PushTransaction adds a transaction and associates it with a participant.
-// Uses canonical pointer pattern: stores one copy of each transaction (by hash) and tracks
+// Uses canonical pointer pattern: stores one copy of each transaction (by ToID) and tracks
 // which participants interacted with it. Multiple participants can reference the same transaction.
 func (b *IndexerBuffer) PushTransaction(participant string, transaction *types.Transaction) {
 	b.recordTransaction(participant, transaction)
@@ -138,17 +145,16 @@ func (b *IndexerBuffer) PushTransaction(participant string, transaction *types.T
 // recordTransaction is the shared internal helper that stores a transaction pointer and
 // records the participant:
 //
-// 1. Check if transaction already exists in txByHash
+// 1. Check if transaction already exists in txByToID
 // 2. If not, store the transaction pointer
 // 3. Add participant to this transaction's participant set in participantsByToID
 func (b *IndexerBuffer) recordTransaction(participant string, transaction *types.Transaction) {
-	txHash := transaction.Hash.String()
-	if _, exists := b.txByHash[txHash]; !exists {
-		b.txByHash[txHash] = transaction
+	toID := transaction.ToID
+	if _, exists := b.txByToID[toID]; !exists {
+		b.txByToID[toID] = transaction
 	}
 
 	// Track this participant by ToID
-	toID := transaction.ToID
 	participants, exists := b.participantsByToID[toID]
 	if !exists {
 		participants = make(map[string]struct{})
@@ -161,7 +167,7 @@ func (b *IndexerBuffer) recordTransaction(participant string, transaction *types
 
 // GetNumberOfTransactions returns the count of unique transactions in the buffer.
 func (b *IndexerBuffer) GetNumberOfTransactions() int {
-	return len(b.txByHash)
+	return len(b.txByToID)
 }
 
 // GetNumberOfOperations returns the count of unique operations in the buffer.
@@ -171,8 +177,8 @@ func (b *IndexerBuffer) GetNumberOfOperations() int {
 
 // GetTransactions returns all unique transactions.
 func (b *IndexerBuffer) GetTransactions() []*types.Transaction {
-	txs := make([]*types.Transaction, 0, len(b.txByHash))
-	for _, txPtr := range b.txByHash {
+	txs := make([]*types.Transaction, 0, len(b.txByToID))
+	for _, txPtr := range b.txByToID {
 		txs = append(txs, txPtr)
 	}
 
@@ -521,7 +527,7 @@ func (b *IndexerBuffer) IngestTransactionResult(r *TransactionResult) {
 // key, so changes from two different ledgers must never coexist in those maps.
 func (b *IndexerBuffer) Clear() {
 	// Clear maps (keep allocated backing arrays)
-	clear(b.txByHash)
+	clear(b.txByToID)
 	clear(b.participantsByToID)
 	clear(b.opByID)
 	clear(b.participantsByOpID)
diff --git a/internal/indexer/indexer_buffer_test.go b/internal/indexer/indexer_buffer_test.go
index e63e9f7bc..d73eeecea 100644
--- a/internal/indexer/indexer_buffer_test.go
+++ b/internal/indexer/indexer_buffer_test.go
@@ -51,6 +51,29 @@ func TestIndexerBuffer_PushTransaction(t *testing.T) {
 		// Assert GetAllTransactions
 		assert.ElementsMatch(t, []*types.Transaction{&tx1, &tx2}, indexerBuffer.GetTransactions())
 	})
+
+	t.Run("🟢 same hash at two ToIDs keeps both transactions", func(t *testing.T) {
+		// The same envelope can occupy several tx-set positions: distinct
+		// ToIDs, one hash. Every ToID with a participant entry must have its
+		// canonical transaction row, or the participant link is COPYed with a
+		// zero ledger_created_at and the transaction row is silently dropped.
+		indexerBuffer := NewIndexerBuffer()
+
+		const sharedHash = "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760"
+		tx1 := types.Transaction{Hash: sharedHash, ToID: 1}
+		tx2 := types.Transaction{Hash: sharedHash, ToID: 2}
+
+		indexerBuffer.PushTransaction("alice", &tx1)
+		indexerBuffer.PushTransaction("bob", &tx2)
+
+		assert.Equal(t, 2, indexerBuffer.GetNumberOfTransactions())
+		assert.ElementsMatch(t, []*types.Transaction{&tx1, &tx2}, indexerBuffer.GetTransactions())
+
+		txParticipants := indexerBuffer.GetTransactionsParticipants()
+		for toID := range txParticipants {
+			assert.Contains(t, []int64{tx1.ToID, tx2.ToID}, toID)
+		}
+	})
 }
 
 func TestIndexerBuffer_PushOperation(t *testing.T) {

From 6c731d3aa266570f285b6379dd0b24cd2ef585d2 Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Tue, 11 Aug 2026 00:10:51 -0400
Subject: [PATCH 04/11] =?UTF-8?q?chore(cmd):=20default=20live=20persist=20?=
 =?UTF-8?q?batching=20to=201=20=E2=80=94=20commit=20per=20ledger?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 cmd/ingest.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/cmd/ingest.go b/cmd/ingest.go
index c15f0b0c4..e3468063b 100644
--- a/cmd/ingest.go
+++ b/cmd/ingest.go
@@ -70,10 +70,10 @@ func (c *ingestCmd) Command() *cobra.Command {
 		},
 		{
 			Name:        "live-persist-max-batch-size",
-			Usage:       "Maximum consecutive ledgers coalesced into one persist commit when live ingestion falls behind. 1 persists every ledger in its own commit; larger values amortize inserts across a backlog at the cost of coarser crash recovery and visibility latency under load.",
+			Usage:       "Maximum consecutive ledgers coalesced into one persist commit when live ingestion falls behind. The default 1 persists every ledger in its own commit — right for networks whose close time comfortably exceeds the persist time; raise it on high-TPL/short-block deployments to amortize inserts across a backlog at the cost of coarser crash recovery and visibility latency under load.",
 			OptType:     types.Int,
 			ConfigKey:   &cfg.LivePersistMaxBatchSize,
-			FlagDefault: 5,
+			FlagDefault: 1,
 			Required:    false,
 		},
 		{

From 4eb61423462349260a08920583c78a12b1d242d4 Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Tue, 11 Aug 2026 10:46:59 -0400
Subject: [PATCH 05/11] perf(data): bound the startup-reconciliation deletes by
 the cursor ledger's close time
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

DeleteRowsAboveLedger scanned every chunk TOID chunk-skipping stats
could not exclude — stats only cover compressed chunks, and three of
the five tables have no TOID-leading index. On a database carrying hours
of full ledgers that scan runs long enough to hold up startup entirely. Close times are monotone, so every orphan above the cursor
carries ledger_created_at at or after the cursor ledger's own close
time; binding that as a partition-column predicate statically excludes
all older chunks regardless of compression, and turns state_changes'
scan into a primary-key range seek. A cursor ledger with no
transactions row leaves the bound unresolvable and the deletes fall
back to the unbounded scan.
---
 internal/data/ingest_store.go      | 36 +++++++++++++++++++++++++-----
 internal/data/ingest_store_test.go | 15 +++++++++++++
 2 files changed, 46 insertions(+), 5 deletions(-)

diff --git a/internal/data/ingest_store.go b/internal/data/ingest_store.go
index 7a0ff0e41..b87d6cf16 100644
--- a/internal/data/ingest_store.go
+++ b/internal/data/ingest_store.go
@@ -272,14 +272,34 @@ func (m *IngestStoreModel) GetOldestLedger(ctx context.Context) (uint32, error)
 // 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
+// those commits leaves orphaned bulk rows for (at most) the persist batch past
+// the committed cursor — and they must be cleared before those ledgers are
 // re-ingested, because COPY has no ON CONFLICT and would collide on the
 // primary keys. Rows of ledgers > ledger are exactly rows whose TOID column is
-// >= the first TOID of ledger+1; each table has chunk skipping on its TOID
-// column, so the deletes stay bounded to the newest chunks.
+// >= the first TOID of ledger+1.
+//
+// The deletes also bound ledger_created_at by the cursor ledger's own close
+// time: close times are monotone, so every orphan carries a timestamp at or
+// after it, and the partition-column predicate statically excludes all older
+// chunks — including uncompressed ones, which TOID chunk-skipping stats do
+// not cover — instead of sequentially scanning them on the tables whose PK
+// does not lead with the TOID column. Without a resolvable bound (the cursor
+// ledger left no transactions row) the deletes scan unbounded, which is the
+// correct-but-slow degenerate case.
 func (m *IngestStoreModel) DeleteRowsAboveLedger(ctx context.Context, ledger uint32) error {
 	minTOID := toid.New(int32(ledger+1), 0, 0).ToInt64()
+
+	// The cursor ledger's close time, from its own transactions rows
+	// ([first TOID of ledger, first TOID of ledger+1)) — a PK-prefix seek.
+	var bound time.Time
+	boundErr := m.DB.QueryRow(ctx,
+		`SELECT ledger_created_at FROM transactions WHERE to_id >= $1 AND to_id < $2 LIMIT 1`,
+		toid.New(int32(ledger), 0, 0).ToInt64(), minTOID,
+	).Scan(&bound)
+	if boundErr != nil && !errors.Is(boundErr, pgx.ErrNoRows) {
+		return fmt.Errorf("resolving close-time bound for ledger %d: %w", ledger, boundErr)
+	}
+
 	targets := []struct {
 		table  string
 		column string
@@ -292,8 +312,14 @@ func (m *IngestStoreModel) DeleteRowsAboveLedger(ctx context.Context, ledger uin
 	}
 	err := db.RunInTransaction(ctx, m.DB, func(dbTx pgx.Tx) error {
 		for _, target := range targets {
+			query := fmt.Sprintf(`DELETE FROM %s WHERE %s >= $1`, target.table, target.column)
+			args := []any{minTOID}
+			if !bound.IsZero() {
+				query = fmt.Sprintf(`DELETE FROM %s WHERE ledger_created_at >= $2 AND %s >= $1`, target.table, target.column)
+				args = append(args, bound)
+			}
 			start := time.Now()
-			tag, execErr := dbTx.Exec(ctx, fmt.Sprintf(`DELETE FROM %s WHERE %s >= $1`, target.table, target.column), minTOID)
+			tag, execErr := dbTx.Exec(ctx, query, args...)
 			m.Metrics.QueryDuration.WithLabelValues("DeleteRowsAboveLedger", target.table).Observe(time.Since(start).Seconds())
 			m.Metrics.QueriesTotal.WithLabelValues("DeleteRowsAboveLedger", target.table).Inc()
 			if execErr != nil {
diff --git a/internal/data/ingest_store_test.go b/internal/data/ingest_store_test.go
index daaac5e39..ea3985e8c 100644
--- a/internal/data/ingest_store_test.go
+++ b/internal/data/ingest_store_test.go
@@ -782,4 +782,19 @@ func Test_IngestStoreModel_DeleteRowsAboveLedger(t *testing.T) {
 	for table, c := range countRows(t) {
 		assert.Equal(t, [2]int{1, 0}, c, "%s must be unchanged by a reconciliation no-op", table)
 	}
+
+	// The close-time bound comes from the cursor ledger's own transactions
+	// rows. With no such row the bound is unresolvable and the deletes fall
+	// back to scanning unbounded — slower, but orphans must still go.
+	seedLedger(t, cursorLedger+2, "2026-01-01T00:00:10Z")
+	_, delErr := dbConnectionPool.Exec(ctx, `DELETE FROM transactions`)
+	require.NoError(t, delErr)
+	require.NoError(t, m.DeleteRowsAboveLedger(ctx, cursorLedger))
+	for table, c := range countRows(t) {
+		if table == "transactions" {
+			assert.Equal(t, [2]int{0, 0}, c, "transactions was emptied by the test itself")
+			continue
+		}
+		assert.Equal(t, [2]int{1, 0}, c, "%s must lose its orphan even without a resolvable close-time bound", table)
+	}
 }

From 2f9dbf521ab51ebd8e3e2133c84de3fc107e0911 Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Tue, 11 Aug 2026 15:51:04 -0400
Subject: [PATCH 06/11] fix(services): size the buffer rotation so the persist
 batch cap is reachable

A buffer stays checked out from the moment the process stage takes it until
the batch carrying it commits, so the rotation needs 2*cap buffers to sustain
a full batch: cap for the batch in flight, cap-1 queued in processed, one
being filled. It held cap+1, so process starved on freeBuffers after filling
(cap+1)-k and the batch settled at the fixed point k = (cap+1)-k.

The batch size was therefore pinned below the configured cap at every load,
never reaching it however deep the backlog. The new test pins the old
behaviour: with the pool sized the old way, batches settle below a cap of 3.
---
 internal/services/ingest_live.go | 18 ++++---
 internal/services/ingest_test.go | 88 ++++++++++++++++++++++++++++++++
 2 files changed, 100 insertions(+), 6 deletions(-)

diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index 0a9054a75..550d148fa 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -689,12 +689,18 @@ func (m *ingestService) ingestLiveLedgers(ctx context.Context, startLedger uint3
 
 	// 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 {
+	// memo warm and the maps' backing arrays allocated across ledgers.
+	//
+	// A buffer stays checked out from the moment process takes it until the
+	// batch carrying it commits, so a full batch needs 2*batchCap buffers in
+	// circulation: batchCap held by the in-flight batch, batchCap-1 queued in
+	// processed, and one process is filling. The spare on top keeps handing a
+	// buffer back from ever blocking. Sizing the rotation any tighter caps
+	// the batch below batchCap no matter how deep the backlog — process runs
+	// out of buffers before the queue can refill, and the batch settles at
+	// the size where held and refillable buffers balance.
+	freeBuffers := make(chan *indexer.IndexerBuffer, 2*batchCap+1)
+	for range 2*batchCap + 1 {
 		freeBuffers <- indexer.NewIndexerBuffer()
 	}
 
diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go
index 791d4b31c..13298d1cf 100644
--- a/internal/services/ingest_test.go
+++ b/internal/services/ingest_test.go
@@ -14,6 +14,7 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/testutil"
+	dto "github.com/prometheus/client_model/go"
 	"github.com/stellar/go-stellar-sdk/ingest/ledgerbackend"
 	"github.com/stellar/go-stellar-sdk/keypair"
 	"github.com/stellar/go-stellar-sdk/network"
@@ -1541,6 +1542,93 @@ func Test_persistBatchCut(t *testing.T) {
 	assert.Equal(t, 2, svc.persistBatchCut(pending))
 }
 
+// Test_ingestLiveLedgers_batchReachesConfiguredCap pins the sizing of the
+// process↔persist buffer rotation. A buffer stays checked out until the batch
+// carrying it commits, so sustaining a full batch takes the cap's worth of
+// buffers for the batch in flight plus the cap's worth for the process stage to
+// refill behind it. Sized any tighter, process starves on freeBuffers before the
+// queue refills and the batch settles strictly below livePersistMaxBatchSize no
+// matter how deep the backlog — leaving the configured cap unreachable at every
+// load, which the batch-size histogram reports as a suspiciously constant value.
+func Test_ingestLiveLedgers_batchReachesConfiguredCap(t *testing.T) {
+	const batchCap = 3
+
+	dbt := dbtest.Open(t)
+	defer dbt.Close()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	pool, err := db.OpenDBConnectionPool(ctx, dbt.DSN)
+	require.NoError(t, err)
+	defer pool.Close()
+
+	const startLedger = uint32(51) // not a multiple of oldestLedgerSyncInterval (100)
+	setupDBCursors(t, ctx, pool, startLedger-1, startLedger-1)
+
+	m := metrics.NewMetrics(prometheus.NewRegistry())
+	models, err := data.NewModels(pool, m.DB)
+	require.NoError(t, err)
+
+	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
+	mockTokenIngestionService.On("ProcessTokenChanges",
+		mock.Anything, // ctx
+		mock.Anything, // dbTx
+		mock.Anything, // trustlineChangesByTrustlineKey
+		mock.Anything, // accountChangesByAccountID
+		mock.Anything, // sacBalanceChangesByKey
+		mock.Anything, // lpShareChangesByKey
+		mock.Anything, // lpChangesByPoolID
+	).Return(nil).Maybe()
+
+	// Supply is unbounded and every ledger is empty, so fetch and process both
+	// outrun the persist stage's commit round-trip: a backlog is always present
+	// for the batch to coalesce.
+	mockBackend := &LedgerBackendMock{}
+	mockBackend.On("GetLedger", mock.Anything, mock.Anything).Return(dummyLedgerMeta(1), nil).Maybe()
+	mockBackend.On("GetLatestLedgerSequence", mock.Anything).
+		Return(uint32(0), errors.New("lag gauge unused in this test")).Maybe()
+
+	svc, err := NewIngestService(IngestServiceConfig{
+		IngestionMode:           IngestionModeLive,
+		Models:                  models,
+		RPCService:              &RPCServiceMock{},
+		LedgerBackend:           mockBackend,
+		TokenIngestionService:   mockTokenIngestionService,
+		Metrics:                 m,
+		Network:                 network.TestNetworkPassphrase,
+		NetworkPassphrase:       network.TestNetworkPassphrase,
+		Archive:                 &HistoryArchiveMock{},
+		LivePersistMaxBatchSize: batchCap,
+	})
+	require.NoError(t, err)
+
+	done := make(chan error, 1)
+	go func() { done <- svc.ingestLiveLedgers(ctx, startLedger, func(context.Context) error { return nil }) }()
+
+	// Grade the mean over many commits, not a single observation: while the
+	// rotation is still filling at startup the process stage can hand off more
+	// than the steady state sustains, so one large batch proves nothing.
+	var h *dto.Histogram
+	require.Eventually(t, func() bool {
+		var mt dto.Metric
+		require.NoError(t, m.Ingestion.PersistBatchSize.Write(&mt))
+		h = mt.GetHistogram()
+		return h.GetSampleCount() >= 20
+	}, 30*time.Second, 50*time.Millisecond, "pipeline did not reach 20 persist commits")
+
+	mean := h.GetSampleSum() / float64(h.GetSampleCount())
+	assert.Greaterf(t, mean, float64(batchCap)-0.5,
+		"mean persist batch %.2f over %d commits fell short of the configured cap %d: the process↔persist buffer rotation cannot refill a full batch",
+		mean, h.GetSampleCount(), batchCap)
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("ingestLiveLedgers did not return after context cancellation")
+	}
+}
+
 // oneLedger wraps a single ledger's persist payload as the batch slice
 // persistLedgerData and persistLedgerDataWithRetry consume.
 func oneLedger(seq uint32, meta xdr.LedgerCloseMeta, plan *ClassificationPlan, contractData *contractDataMemo, buffer *indexer.IndexerBuffer) []persistItem {

From 033f3797ad8c2d7c3fdaddacc4553a944a13d4ed Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Thu, 13 Aug 2026 11:07:27 -0400
Subject: [PATCH 07/11] perf(ingest): move FK-free balance upserts onto their
 own persist sibling

The native-balance and liquidity-pool upserts ran serially inside the
coordinating transaction, lengthening its critical path while the COPY
siblings streamed concurrently. Their tables carry no foreign keys, so a
sixth sibling now carries them; the trustline and SAC balances stay on the
coordinating transaction because their parent rows (trustline_assets,
contract_tokens) are staged uncommitted there, and a sibling committing
first would fail FK checks against committed state.

Reconciliation (DeleteRowsAboveLedger) does not cover the balance tables:
a crash can leave balance rows above the committed cursor, which is safe
because the upserts are idempotent and reapply when those ledgers
re-ingest.
---
 internal/db/db.go                         |  9 +--
 internal/services/ingest_live.go          | 79 ++++++++++++---------
 internal/services/ingest_test.go          | 84 ++++++++++++++++-------
 internal/services/mocks.go                |  9 ++-
 internal/services/token_ingestion.go      | 37 ++++++----
 internal/services/token_ingestion_test.go |  7 +-
 6 files changed, 147 insertions(+), 78 deletions(-)

diff --git a/internal/db/db.go b/internal/db/db.go
index abe64a700..6b34d1686 100644
--- a/internal/db/db.go
+++ b/internal/db/db.go
@@ -17,10 +17,11 @@ import (
 
 const (
 	DefaultMaxConnIdleTime time.Duration = 10 * time.Second
-	// Live persist alone holds 6 connections at its commit barrier (5 COPY
-	// siblings + the coordinator), plus the advisory-lock session and
-	// pool-side classification reads; 12 leaves headroom so no persist-path
-	// Acquire queues behind an unrelated consumer.
+	// Live persist alone holds 7 connections at its commit barrier (5 COPY
+	// siblings + the balances sibling + the coordinator), plus the
+	// advisory-lock session and up to 2 transient pool-side classification
+	// reads; 12 leaves headroom so no persist-path Acquire queues behind an
+	// unrelated consumer.
 	DefaultMaxConns        int32         = 12
 	DefaultMinConns        int32         = 5
 	DefaultMaxConnLifetime time.Duration = 5 * time.Minute
diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index 550d148fa..49ec58783 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -102,20 +102,22 @@ func batchLabel(items []persistItem) string {
 
 // persistLedgerData persists a batch of consecutive ledgers in one commit
 // set. The five bulk COPY families — transactions, transactions_accounts,
-// operations, operations_accounts, state_changes — stream concurrently on
-// sibling connections, each in its own transaction covering every ledger in
-// the batch, while the coordinating transaction stages everything else (assets,
-// contracts, classification, protocol state, token changes, cursor) ledger
-// by ledger in order — the per-protocol CAS chain advances N-1 → N inside
-// the transaction, and the guarded cursor's final value is the batch's last
-// ledger. All the slow work happens uncommitted and invisible; only after
-// every stream and the coordinator succeed do the commits fire, siblings
-// first and the coordinating transaction strictly last. The cursor it
-// carries is the authority: the only crash state this ordering can produce
-// is orphaned bulk rows above the committed cursor, which
-// DeleteRowsAboveLedger removes at startup. A failure before the first
-// commit rolls everything back and the whole batch is cleanly retryable; a
-// failure after it wraps ErrPartialPersist and is fatal.
+// operations, operations_accounts, state_changes — plus the FK-free balance
+// upserts stream concurrently on sibling connections, each in its own
+// transaction covering every ledger in the batch, while the coordinating
+// transaction stages everything else (assets, contracts, classification,
+// protocol state, trustline/SAC balances, cursor) ledger by ledger in order
+// — the per-protocol CAS chain advances N-1 → N inside the transaction, and
+// the guarded cursor's final value is the batch's last ledger. All the slow
+// work happens uncommitted and invisible; only after every stream and the
+// coordinator succeed do the commits fire, siblings first and the
+// coordinating transaction strictly last. The cursor it carries is the
+// authority: the only crash state this ordering can produce is sibling rows
+// above the committed cursor — DeleteRowsAboveLedger removes the bulk
+// families' at startup, and the balances sibling's idempotent upserts simply
+// reapply when those ledgers re-ingest. A failure before the first commit
+// rolls everything back and the whole batch is cleanly retryable; a failure
+// after it wraps ErrPartialPersist and is fatal.
 //
 // Only the first ledger of a batch may carry a classification plan: a
 // plan's pool reads see exactly the state the previous batch committed (see
@@ -156,6 +158,18 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 		{"state_changes", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
 			return m.insertStateChanges(ctx, dbTx, it.buffer.GetStateChanges())
 		}},
+		// The FK-free balance upserts ride their own sibling so the coordinating
+		// transaction's serial path stays short. Their tables have no foreign
+		// keys, so nothing here can observe the coordinator's uncommitted rows;
+		// the trustline and SAC balances, whose parents the coordinator stages,
+		// remain with it in stageCoordinatedWrites.
+		{"balances", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
+			return m.tokenIngestionService.ProcessNativeAndPoolChanges(ctx, dbTx,
+				it.buffer.GetAccountChanges(),
+				it.buffer.GetLiquidityPoolShareChanges(),
+				it.buffer.GetLiquidityPoolChanges(),
+			)
+		}},
 	}
 	siblingTxs := make([]pgx.Tx, len(siblings))
 	for i, s := range siblings {
@@ -179,18 +193,19 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 		// last, and its flush covers all earlier WAL — including these
 		// commit records — so a durable cursor implies durable siblings. A
 		// crash inside the window can only lose rows the cursor never
-		// acknowledged, exactly what startup reconciliation
-		// (DeleteRowsAboveLedger) removes anyway.
+		// acknowledged: startup reconciliation (DeleteRowsAboveLedger)
+		// removes the bulk families' anyway, and the balances sibling's
+		// idempotent upserts reapply on re-ingest.
 		if _, setErr := tx.Exec(ctx, "SET LOCAL synchronous_commit = off"); setErr != nil {
 			return fmt.Errorf("disabling synchronous commit on %s for %s: %w", s.name, label, setErr)
 		}
 	}
 
-	// Stream the COPY families and stage the coordinated writes concurrently.
-	// Every sibling owns exactly one table and the table sets are disjoint (no
-	// FKs among them), and the goroutines only read the quiescent buffers, so
-	// the six transactions never contend. Within each transaction the batch's
-	// ledgers run in order.
+	// Stream the sibling writes and stage the coordinated writes concurrently.
+	// Every sibling owns a disjoint set of tables with no FKs among them or to
+	// any other transaction's tables, and the goroutines only read the
+	// quiescent buffers, so the transactions never contend. Within each
+	// transaction the batch's ledgers run in order.
 	g, gctx := errgroup.WithContext(ctx)
 	for i, s := range siblings {
 		g.Go(func() error {
@@ -212,8 +227,8 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 		return nil
 	})
 	if err = g.Wait(); err != nil {
-		// Nothing has committed: the deferred rollbacks discard all six
-		// transactions and the batch is cleanly retryable.
+		// Nothing has committed: the deferred rollbacks discard every
+		// transaction and the batch is cleanly retryable.
 		return fmt.Errorf("persisting ledger data for %s: %w", label, err)
 	}
 
@@ -240,10 +255,10 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 	return nil
 }
 
-// stageCoordinatedWrites runs every per-ledger write except the five bulk
-// COPY families on the coordinating transaction: trustline assets, SAC
-// contract tokens, protocol classification and wasm/contract rows,
-// CAS-gated protocol state, token changes, and finally the guarded cursor.
+// stageCoordinatedWrites runs every per-ledger write the siblings don't own
+// on the coordinating transaction: trustline assets, SAC contract tokens,
+// protocol classification and wasm/contract rows, CAS-gated protocol state,
+// the trustline and SAC balance changes, and finally the guarded cursor.
 func (m *ingestService) stageCoordinatedWrites(
 	ctx context.Context,
 	dbTx pgx.Tx,
@@ -447,13 +462,13 @@ func (m *ingestService) stageCoordinatedWrites(
 		}
 	}
 
-	// 5. Process token changes (trustline add/remove/update, native balance, SAC balance)
-	if txErr = m.tokenIngestionService.ProcessTokenChanges(ctx, dbTx,
+	// 5. Apply the FK-bound balance changes. Trustline and SAC balance rows
+	// reference the trustline_assets and contract_tokens rows staged in steps
+	// 1-2 of this same transaction, so they cannot leave it; the FK-free
+	// balance families run on the "balances" sibling instead.
+	if txErr = m.tokenIngestionService.ProcessTrustlineAndSACChanges(ctx, dbTx,
 		buffer.GetTrustlineChanges(),
-		buffer.GetAccountChanges(),
 		buffer.GetSACBalanceChanges(),
-		buffer.GetLiquidityPoolShareChanges(),
-		buffer.GetLiquidityPoolChanges(),
 	); txErr != nil {
 		return fmt.Errorf("processing token changes for ledger %d: %w", ledgerSeq, txErr)
 	}
diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go
index 13298d1cf..bb2111e8f 100644
--- a/internal/services/ingest_test.go
+++ b/internal/services/ingest_test.go
@@ -1570,12 +1570,16 @@ func Test_ingestLiveLedgers_batchReachesConfiguredCap(t *testing.T) {
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTokenChanges",
+	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 		mock.Anything, // ctx
 		mock.Anything, // dbTx
 		mock.Anything, // trustlineChangesByTrustlineKey
-		mock.Anything, // accountChangesByAccountID
 		mock.Anything, // sacBalanceChangesByKey
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+		mock.Anything, // ctx
+		mock.Anything, // dbTx
+		mock.Anything, // accountChangesByAccountID
 		mock.Anything, // lpShareChangesByKey
 		mock.Anything, // lpChangesByPoolID
 	).Return(nil).Maybe()
@@ -1665,12 +1669,16 @@ func Test_persistLedgerDataWithRetry(t *testing.T) {
 
 		// Mock AccountTokenService to succeed
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTokenChanges",
+		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
 			mock.Anything, // trustlineChangesByTrustlineKey
-			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // sacBalanceChangesByKey
+		).Return(nil)
+		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
+			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // lpShareChangesByKey
 			mock.Anything, // lpChangesByPoolID
 		).Return(nil)
@@ -1743,15 +1751,19 @@ func Test_persistLedgerDataWithRetry(t *testing.T) {
 
 		// Mock AccountTokenService to return error (simulating DB failure)
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTokenChanges",
+		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
 			mock.Anything, // trustlineChangesByTrustlineKey
-			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // sacBalanceChangesByKey
+		).Return(fmt.Errorf("db connection failed"))
+		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
+			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // lpShareChangesByKey
 			mock.Anything, // lpChangesByPoolID
-		).Return(fmt.Errorf("db connection failed"))
+		).Return(nil).Maybe()
 
 		svc, err := NewIngestService(IngestServiceConfig{
 			IngestionMode:         IngestionModeLive,
@@ -1821,24 +1833,25 @@ func Test_persistLedgerDataWithRetry(t *testing.T) {
 
 		// Mock AccountTokenService to fail once then succeed
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTokenChanges",
+		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
 			mock.Anything, // trustlineChangesByTrustlineKey
-			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // sacBalanceChangesByKey
-			mock.Anything, // lpShareChangesByKey
-			mock.Anything, // lpChangesByPoolID
 		).Return(fmt.Errorf("transient error")).Once()
-		mockTokenIngestionService.On("ProcessTokenChanges",
+		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
 			mock.Anything, // trustlineChangesByTrustlineKey
-			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // sacBalanceChangesByKey
+		).Return(nil).Once()
+		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
+			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // lpShareChangesByKey
 			mock.Anything, // lpChangesByPoolID
-		).Return(nil).Once()
+		).Return(nil).Maybe()
 
 		svc, err := NewIngestService(IngestServiceConfig{
 			IngestionMode:         IngestionModeLive,
@@ -2027,12 +2040,16 @@ func Test_persistLedgerData_ProtocolCASGating(t *testing.T) {
 		require.NoError(t, err)
 
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTokenChanges",
+		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
 			mock.Anything, // trustlineChangesByTrustlineKey
-			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // sacBalanceChangesByKey
+		).Return(nil).Maybe()
+		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
+			mock.Anything, // accountChangesByAccountID
 			mock.Anything, // lpShareChangesByKey
 			mock.Anything, // lpChangesByPoolID
 		).Return(nil).Maybe()
@@ -2756,9 +2773,11 @@ func Test_persistLedgerData_ClassificationPlan(t *testing.T) {
 		require.NoError(t, err)
 
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTokenChanges",
+		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 			mock.Anything, mock.Anything, mock.Anything, mock.Anything,
-			mock.Anything, mock.Anything, mock.Anything,
+		).Return(nil).Maybe()
+		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+			mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
 		).Return(nil).Maybe()
 
 		svc, err := NewIngestService(IngestServiceConfig{
@@ -2926,12 +2945,16 @@ func Test_ingestService_ingestLiveLedgers_LagReadDoesNotBlockConsumer(t *testing
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTokenChanges",
+	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
 		mock.Anything, // ctx
 		mock.Anything, // dbTx
 		mock.Anything, // trustlineChangesByTrustlineKey
-		mock.Anything, // accountChangesByAccountID
 		mock.Anything, // sacBalanceChangesByKey
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+		mock.Anything, // ctx
+		mock.Anything, // dbTx
+		mock.Anything, // accountChangesByAccountID
 		mock.Anything, // lpShareChangesByKey
 		mock.Anything, // lpChangesByPoolID
 	).Return(nil).Maybe()
@@ -3061,8 +3084,11 @@ func Test_ingestService_ingestLiveLedgers_StageErrorStopsPipeline(t *testing.T)
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTokenChanges",
-		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
 	).Return(nil).Maybe()
 
 	cursorReached := func(target uint32) bool {
@@ -3136,8 +3162,11 @@ func Test_persistLedgerData_Batch(t *testing.T) {
 		require.NoError(t, err)
 
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTokenChanges",
-			mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+			mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		).Return(nil).Maybe()
+		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+			mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
 		).Return(nil).Maybe()
 
 		svc, err := NewIngestService(IngestServiceConfig{
@@ -3236,8 +3265,11 @@ func Test_persistLedgerData_SiblingFailureRollsBackEverything(t *testing.T) {
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTokenChanges",
-		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
+		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
 	).Return(nil).Maybe()
 
 	svc, err := NewIngestService(IngestServiceConfig{
diff --git a/internal/services/mocks.go b/internal/services/mocks.go
index 6bea76a3e..7a761140d 100644
--- a/internal/services/mocks.go
+++ b/internal/services/mocks.go
@@ -126,8 +126,13 @@ type TokenIngestionServiceMock struct {
 
 var _ TokenIngestionService = (*TokenIngestionServiceMock)(nil)
 
-func (m *TokenIngestionServiceMock) ProcessTokenChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, accountChangesByAccountID map[string]types.AccountChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange, lpShareChangesByKey map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange, lpChangesByPoolID map[string]types.LiquidityPoolChange) error {
-	args := m.Called(ctx, dbTx, trustlineChangesByTrustlineKey, accountChangesByAccountID, sacBalanceChangesByKey, lpShareChangesByKey, lpChangesByPoolID)
+func (m *TokenIngestionServiceMock) ProcessTrustlineAndSACChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error {
+	args := m.Called(ctx, dbTx, trustlineChangesByTrustlineKey, sacBalanceChangesByKey)
+	return args.Error(0)
+}
+
+func (m *TokenIngestionServiceMock) ProcessNativeAndPoolChanges(ctx context.Context, dbTx pgx.Tx, accountChangesByAccountID map[string]types.AccountChange, lpShareChangesByKey map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange, lpChangesByPoolID map[string]types.LiquidityPoolChange) error {
+	args := m.Called(ctx, dbTx, accountChangesByAccountID, lpShareChangesByKey, lpChangesByPoolID)
 	return args.Error(0)
 }
 
diff --git a/internal/services/token_ingestion.go b/internal/services/token_ingestion.go
index 548881bcd..05efba0c6 100644
--- a/internal/services/token_ingestion.go
+++ b/internal/services/token_ingestion.go
@@ -15,10 +15,21 @@ import (
 )
 
 // TokenIngestionService provides write access to account token storage during live ingestion.
+// The balance writes are split by foreign-key dependence: trustline and SAC balances reference
+// parent rows (trustline_assets, contract_tokens) that live persist stages uncommitted on the
+// coordinating transaction, so they must run on that same transaction; native-balance and
+// liquidity-pool tables carry no foreign keys, so their writes may run on any transaction —
+// live persist gives them their own sibling, concurrent with the coordinator.
 type TokenIngestionService interface {
-	// ProcessTokenChanges applies trustline, native, SAC, and liquidity-pool balance changes to
-	// PostgreSQL. This is called by the indexer for each ledger's state changes during live ingestion.
-	ProcessTokenChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, accountChangesByAccountID map[string]types.AccountChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange, lpShareChangesByKey map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange, lpChangesByPoolID map[string]types.LiquidityPoolChange) error
+	// ProcessTrustlineAndSACChanges applies trustline and SAC balance changes. dbTx must be
+	// the transaction on which this ledger's trustline_assets and contract_tokens rows are
+	// staged: both tables' foreign keys are checked against committed state, and a
+	// transaction that commits before those parents would fail with SQLSTATE 23503.
+	ProcessTrustlineAndSACChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error
+	// ProcessNativeAndPoolChanges applies native-balance and liquidity-pool changes. The
+	// target tables have no foreign keys, and every upsert is idempotent (ON CONFLICT with
+	// an IS DISTINCT FROM guard), so the writes are safe on any transaction.
+	ProcessNativeAndPoolChanges(ctx context.Context, dbTx pgx.Tx, accountChangesByAccountID map[string]types.AccountChange, lpShareChangesByKey map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange, lpChangesByPoolID map[string]types.LiquidityPoolChange) error
 }
 
 // Verify interface compliance at compile time
@@ -56,20 +67,22 @@ func NewTokenIngestionService(cfg TokenIngestionServiceConfig) *tokenIngestionSe
 	}
 }
 
-// ProcessTokenChanges processes token changes and stores them in PostgreSQL.
-// This is called by the indexer for each ledger's state changes during live ingestion.
-func (s *tokenIngestionService) ProcessTokenChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, accountChangesByAccountID map[string]types.AccountChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange, lpShareChangesByKey map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange, lpChangesByPoolID map[string]types.LiquidityPoolChange) error {
-	if len(trustlineChangesByTrustlineKey) == 0 && len(accountChangesByAccountID) == 0 && len(sacBalanceChangesByKey) == 0 && len(lpShareChangesByKey) == 0 && len(lpChangesByPoolID) == 0 {
-		return nil
-	}
-
+// ProcessTrustlineAndSACChanges applies trustline and SAC balance changes on the
+// transaction that stages their trustline_assets / contract_tokens parent rows.
+func (s *tokenIngestionService) ProcessTrustlineAndSACChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error {
 	if err := s.processTrustlineChanges(ctx, dbTx, trustlineChangesByTrustlineKey); err != nil {
 		return err
 	}
-	if err := s.processNativeBalanceChanges(ctx, dbTx, accountChangesByAccountID); err != nil {
+	if err := s.processSACBalanceChanges(ctx, dbTx, sacBalanceChangesByKey); err != nil {
 		return err
 	}
-	if err := s.processSACBalanceChanges(ctx, dbTx, sacBalanceChangesByKey); err != nil {
+	return nil
+}
+
+// ProcessNativeAndPoolChanges applies native-balance and liquidity-pool changes; the
+// target tables have no foreign keys, so any transaction may carry them.
+func (s *tokenIngestionService) ProcessNativeAndPoolChanges(ctx context.Context, dbTx pgx.Tx, accountChangesByAccountID map[string]types.AccountChange, lpShareChangesByKey map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange, lpChangesByPoolID map[string]types.LiquidityPoolChange) error {
+	if err := s.processNativeBalanceChanges(ctx, dbTx, accountChangesByAccountID); err != nil {
 		return err
 	}
 	if err := s.processLiquidityPoolChanges(ctx, dbTx, lpChangesByPoolID); err != nil {
diff --git a/internal/services/token_ingestion_test.go b/internal/services/token_ingestion_test.go
index 35a886958..470778daf 100644
--- a/internal/services/token_ingestion_test.go
+++ b/internal/services/token_ingestion_test.go
@@ -337,7 +337,7 @@ func TestGetAccountTrustlineBalances(t *testing.T) {
 	})
 }
 
-func TestProcessTokenChanges(t *testing.T) {
+func TestProcessBalanceChanges(t *testing.T) {
 	ctx := context.Background()
 
 	dbt := dbtest.Open(t)
@@ -369,7 +369,10 @@ func TestProcessTokenChanges(t *testing.T) {
 		})
 
 		err = db.RunInTransaction(ctx, dbConnectionPool, func(dbTx pgx.Tx) error {
-			return service.ProcessTokenChanges(ctx, dbTx, map[indexer.TrustlineChangeKey]types.TrustlineChange{}, make(map[string]types.AccountChange), make(map[indexer.SACBalanceChangeKey]types.SACBalanceChange), make(map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange), make(map[string]types.LiquidityPoolChange))
+			if tlErr := service.ProcessTrustlineAndSACChanges(ctx, dbTx, map[indexer.TrustlineChangeKey]types.TrustlineChange{}, make(map[indexer.SACBalanceChangeKey]types.SACBalanceChange)); tlErr != nil {
+				return tlErr
+			}
+			return service.ProcessNativeAndPoolChanges(ctx, dbTx, make(map[string]types.AccountChange), make(map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange), make(map[string]types.LiquidityPoolChange))
 		})
 		assert.NoError(t, err)
 	})

From 50e54ce24f6b470ab0792ca18eaaecb0bec475d9 Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Thu, 13 Aug 2026 13:07:16 -0400
Subject: [PATCH 08/11] perf(ingest): give trustline assets and balances their
 own persist sibling

Each balance family now rides the transaction that stages its foreign-key
parents: trustline_assets and trustline_balances move from the
coordinating transaction onto a new sibling, alongside the balances
sibling that carries pools and pool-share balances under the new pool
foreign key. Every FK is checked at its own transaction's commit, and the
coordinator's serial path shrinks to contracts, classification, protocol
state, SAC balances, and the cursor.

SAC balances stay coordinated: their parent (contract_tokens) is also
written by the classification path there, and a same-key insert from two
concurrent transactions could deadlock at the commit barrier.
---
 internal/db/db.go                         | 10 ++--
 internal/services/ingest_live.go          | 69 +++++++++++-----------
 internal/services/ingest_test.go          | 71 +++++++++++++++++------
 internal/services/mocks.go                |  9 ++-
 internal/services/token_ingestion.go      | 51 ++++++++--------
 internal/services/token_ingestion_test.go |  5 +-
 6 files changed, 132 insertions(+), 83 deletions(-)

diff --git a/internal/db/db.go b/internal/db/db.go
index 6b34d1686..a013190f8 100644
--- a/internal/db/db.go
+++ b/internal/db/db.go
@@ -17,11 +17,11 @@ import (
 
 const (
 	DefaultMaxConnIdleTime time.Duration = 10 * time.Second
-	// Live persist alone holds 7 connections at its commit barrier (5 COPY
-	// siblings + the balances sibling + the coordinator), plus the
-	// advisory-lock session and up to 2 transient pool-side classification
-	// reads; 12 leaves headroom so no persist-path Acquire queues behind an
-	// unrelated consumer.
+	// Live persist alone holds 8 connections at its commit barrier (5 COPY
+	// siblings + the balances and trustlines siblings + the coordinator),
+	// plus the advisory-lock session and up to 2 transient pool-side
+	// classification reads; 12 leaves headroom so no persist-path Acquire
+	// queues behind an unrelated consumer.
 	DefaultMaxConns        int32         = 12
 	DefaultMinConns        int32         = 5
 	DefaultMaxConnLifetime time.Duration = 5 * time.Minute
diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index 49ec58783..fcbb0a379 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -102,19 +102,20 @@ func batchLabel(items []persistItem) string {
 
 // persistLedgerData persists a batch of consecutive ledgers in one commit
 // set. The five bulk COPY families — transactions, transactions_accounts,
-// operations, operations_accounts, state_changes — plus the FK-free balance
-// upserts stream concurrently on sibling connections, each in its own
+// operations, operations_accounts, state_changes — plus the balance
+// families (native/pool on one sibling, trustline assets and balances on
+// another) stream concurrently on sibling connections, each in its own
 // transaction covering every ledger in the batch, while the coordinating
-// transaction stages everything else (assets, contracts, classification,
-// protocol state, trustline/SAC balances, cursor) ledger by ledger in order
-// — the per-protocol CAS chain advances N-1 → N inside the transaction, and
-// the guarded cursor's final value is the batch's last ledger. All the slow
-// work happens uncommitted and invisible; only after every stream and the
+// transaction stages everything else (contracts, classification, protocol
+// state, SAC balances, cursor) ledger by ledger in order — the per-protocol
+// CAS chain advances N-1 → N inside the transaction, and the guarded
+// cursor's final value is the batch's last ledger. All the slow work
+// happens uncommitted and invisible; only after every stream and the
 // coordinator succeed do the commits fire, siblings first and the
 // coordinating transaction strictly last. The cursor it carries is the
 // authority: the only crash state this ordering can produce is sibling rows
 // above the committed cursor — DeleteRowsAboveLedger removes the bulk
-// families' at startup, and the balances sibling's idempotent upserts simply
+// families' at startup, and the balance siblings' idempotent upserts simply
 // reapply when those ledgers re-ingest. A failure before the first commit
 // rolls everything back and the whole batch is cleanly retryable; a failure
 // after it wraps ErrPartialPersist and is fatal.
@@ -158,11 +159,12 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 		{"state_changes", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
 			return m.insertStateChanges(ctx, dbTx, it.buffer.GetStateChanges())
 		}},
-		// The FK-free balance upserts ride their own sibling so the coordinating
-		// transaction's serial path stays short. Their tables have no foreign
-		// keys, so nothing here can observe the coordinator's uncommitted rows;
-		// the trustline and SAC balances, whose parents the coordinator stages,
-		// remain with it in stageCoordinatedWrites.
+		// Each balance family rides the transaction that stages its FK parents,
+		// so the coordinating transaction's serial path stays short and every
+		// foreign key is checked within one commit. The SAC balances remain in
+		// stageCoordinatedWrites: their parent (contract_tokens) is also written
+		// by the classification path there, and a same-key insert from two
+		// concurrent transactions could deadlock at the commit barrier.
 		{"balances", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
 			return m.tokenIngestionService.ProcessNativeAndPoolChanges(ctx, dbTx,
 				it.buffer.GetAccountChanges(),
@@ -170,6 +172,14 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 				it.buffer.GetLiquidityPoolChanges(),
 			)
 		}},
+		{"trustlines", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
+			if uniqueAssets := it.buffer.GetUniqueTrustlineAssets(); len(uniqueAssets) > 0 {
+				if err := m.models.TrustlineAsset.BatchInsert(ctx, dbTx, uniqueAssets); err != nil {
+					return fmt.Errorf("inserting trustline assets: %w", err)
+				}
+			}
+			return m.tokenIngestionService.ProcessTrustlineChanges(ctx, dbTx, it.buffer.GetTrustlineChanges())
+		}},
 	}
 	siblingTxs := make([]pgx.Tx, len(siblings))
 	for i, s := range siblings {
@@ -256,9 +266,9 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 }
 
 // stageCoordinatedWrites runs every per-ledger write the siblings don't own
-// on the coordinating transaction: trustline assets, SAC contract tokens,
-// protocol classification and wasm/contract rows, CAS-gated protocol state,
-// the trustline and SAC balance changes, and finally the guarded cursor.
+// on the coordinating transaction: SAC contract tokens, protocol
+// classification and wasm/contract rows, CAS-gated protocol state, the SAC
+// balance changes, and finally the guarded cursor.
 func (m *ingestService) stageCoordinatedWrites(
 	ctx context.Context,
 	dbTx pgx.Tx,
@@ -268,15 +278,7 @@ func (m *ingestService) stageCoordinatedWrites(
 	contractData *contractDataMemo,
 	buffer *indexer.IndexerBuffer,
 ) error {
-	// 1. Insert unique trustline assets (FK prerequisite for trustline balances)
-	uniqueAssets := buffer.GetUniqueTrustlineAssets()
-	if len(uniqueAssets) > 0 {
-		if txErr := m.models.TrustlineAsset.BatchInsert(ctx, dbTx, uniqueAssets); txErr != nil {
-			return fmt.Errorf("inserting trustline assets for ledger %d: %w", ledgerSeq, txErr)
-		}
-	}
-
-	// 2. Insert new SAC contract tokens (filter existing, insert)
+	// 1. Insert new SAC contract tokens (filter existing, insert)
 	contracts, txErr := m.prepareNewSACContracts(ctx, dbTx, buffer.GetSACContracts())
 	if txErr != nil {
 		return fmt.Errorf("preparing contract tokens for ledger %d: %w", ledgerSeq, txErr)
@@ -288,7 +290,7 @@ func (m *ingestService) stageCoordinatedWrites(
 		log.Ctx(ctx).Infof("inserted %d SAC contract tokens", len(contracts))
 	}
 
-	// 3. Apply protocol classification (black-box per protocol). plan was
+	// 2. Apply protocol classification (black-box per protocol). plan was
 	// computed by prepareClassificationPlan before this transaction opened,
 	// so any RPC calls (e.g. SEP-41 metadata) already happened;
 	// ApplyClassificationPlan only performs each validator's DB writes
@@ -333,7 +335,7 @@ func (m *ingestService) stageCoordinatedWrites(
 		}
 	}
 
-	// 4. Per-protocol CAS-gated state production. The compare-and-swap on each
+	// 3. Per-protocol CAS-gated state production. The compare-and-swap on each
 	// protocol cursor is the authoritative gate — exactly one of live ingestion or
 	// protocol-migrate wins a given ledger. Staging (ProcessLedger) and persistence
 	// run only for cursors that win the swap, so a protocol still backfilling (its
@@ -462,18 +464,17 @@ func (m *ingestService) stageCoordinatedWrites(
 		}
 	}
 
-	// 5. Apply the FK-bound balance changes. Trustline and SAC balance rows
-	// reference the trustline_assets and contract_tokens rows staged in steps
-	// 1-2 of this same transaction, so they cannot leave it; the FK-free
-	// balance families run on the "balances" sibling instead.
-	if txErr = m.tokenIngestionService.ProcessTrustlineAndSACChanges(ctx, dbTx,
-		buffer.GetTrustlineChanges(),
+	// 4. Apply the SAC balance changes. Their rows reference the
+	// contract_tokens rows staged in step 1 of this same transaction, so they
+	// cannot leave it; the other balance families run on the "balances" and
+	// "trustlines" siblings, each alongside its own foreign-key parents.
+	if txErr = m.tokenIngestionService.ProcessSACBalanceChanges(ctx, dbTx,
 		buffer.GetSACBalanceChanges(),
 	); txErr != nil {
 		return fmt.Errorf("processing token changes for ledger %d: %w", ledgerSeq, txErr)
 	}
 
-	// 6. Advance the latest-ledger cursor. The update is guarded: a session that
+	// 5. Advance the latest-ledger cursor. The update is guarded: a session that
 	// silently lost its advisory lock (server-side failover, see startLiveIngestion's
 	// checkLockSession) must not blindly overwrite a value a second instance already
 	// advanced, or the cursor could regress.
diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go
index bb2111e8f..d0c0e8118 100644
--- a/internal/services/ingest_test.go
+++ b/internal/services/ingest_test.go
@@ -1570,10 +1570,14 @@ func Test_ingestLiveLedgers_batchReachesConfiguredCap(t *testing.T) {
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+	mockTokenIngestionService.On("ProcessTrustlineChanges",
 		mock.Anything, // ctx
 		mock.Anything, // dbTx
 		mock.Anything, // trustlineChangesByTrustlineKey
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessSACBalanceChanges",
+		mock.Anything, // ctx
+		mock.Anything, // dbTx
 		mock.Anything, // sacBalanceChangesByKey
 	).Return(nil).Maybe()
 	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
@@ -1669,10 +1673,14 @@ func Test_persistLedgerDataWithRetry(t *testing.T) {
 
 		// Mock AccountTokenService to succeed
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+		mockTokenIngestionService.On("ProcessTrustlineChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
 			mock.Anything, // trustlineChangesByTrustlineKey
+		).Return(nil)
+		mockTokenIngestionService.On("ProcessSACBalanceChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
 			mock.Anything, // sacBalanceChangesByKey
 		).Return(nil)
 		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
@@ -1751,12 +1759,16 @@ func Test_persistLedgerDataWithRetry(t *testing.T) {
 
 		// Mock AccountTokenService to return error (simulating DB failure)
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+		mockTokenIngestionService.On("ProcessSACBalanceChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
-			mock.Anything, // trustlineChangesByTrustlineKey
 			mock.Anything, // sacBalanceChangesByKey
 		).Return(fmt.Errorf("db connection failed"))
+		mockTokenIngestionService.On("ProcessTrustlineChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
+			mock.Anything, // trustlineChangesByTrustlineKey
+		).Return(nil).Maybe()
 		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
@@ -1833,18 +1845,21 @@ func Test_persistLedgerDataWithRetry(t *testing.T) {
 
 		// Mock AccountTokenService to fail once then succeed
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+		mockTokenIngestionService.On("ProcessSACBalanceChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
-			mock.Anything, // trustlineChangesByTrustlineKey
 			mock.Anything, // sacBalanceChangesByKey
 		).Return(fmt.Errorf("transient error")).Once()
-		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+		mockTokenIngestionService.On("ProcessSACBalanceChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
-			mock.Anything, // trustlineChangesByTrustlineKey
 			mock.Anything, // sacBalanceChangesByKey
 		).Return(nil).Once()
+		mockTokenIngestionService.On("ProcessTrustlineChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
+			mock.Anything, // trustlineChangesByTrustlineKey
+		).Return(nil).Maybe()
 		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
@@ -2040,10 +2055,14 @@ func Test_persistLedgerData_ProtocolCASGating(t *testing.T) {
 		require.NoError(t, err)
 
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+		mockTokenIngestionService.On("ProcessTrustlineChanges",
 			mock.Anything, // ctx
 			mock.Anything, // dbTx
 			mock.Anything, // trustlineChangesByTrustlineKey
+		).Return(nil).Maybe()
+		mockTokenIngestionService.On("ProcessSACBalanceChanges",
+			mock.Anything, // ctx
+			mock.Anything, // dbTx
 			mock.Anything, // sacBalanceChangesByKey
 		).Return(nil).Maybe()
 		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
@@ -2773,8 +2792,11 @@ func Test_persistLedgerData_ClassificationPlan(t *testing.T) {
 		require.NoError(t, err)
 
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
-			mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mockTokenIngestionService.On("ProcessTrustlineChanges",
+			mock.Anything, mock.Anything, mock.Anything,
+		).Return(nil).Maybe()
+		mockTokenIngestionService.On("ProcessSACBalanceChanges",
+			mock.Anything, mock.Anything, mock.Anything,
 		).Return(nil).Maybe()
 		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
 			mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
@@ -2945,10 +2967,14 @@ func Test_ingestService_ingestLiveLedgers_LagReadDoesNotBlockConsumer(t *testing
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
+	mockTokenIngestionService.On("ProcessTrustlineChanges",
 		mock.Anything, // ctx
 		mock.Anything, // dbTx
 		mock.Anything, // trustlineChangesByTrustlineKey
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessSACBalanceChanges",
+		mock.Anything, // ctx
+		mock.Anything, // dbTx
 		mock.Anything, // sacBalanceChangesByKey
 	).Return(nil).Maybe()
 	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
@@ -3084,8 +3110,11 @@ func Test_ingestService_ingestLiveLedgers_StageErrorStopsPipeline(t *testing.T)
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
-		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+	mockTokenIngestionService.On("ProcessTrustlineChanges",
+		mock.Anything, mock.Anything, mock.Anything,
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessSACBalanceChanges",
+		mock.Anything, mock.Anything, mock.Anything,
 	).Return(nil).Maybe()
 	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
 		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
@@ -3162,8 +3191,11 @@ func Test_persistLedgerData_Batch(t *testing.T) {
 		require.NoError(t, err)
 
 		mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-		mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
-			mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+		mockTokenIngestionService.On("ProcessTrustlineChanges",
+			mock.Anything, mock.Anything, mock.Anything,
+		).Return(nil).Maybe()
+		mockTokenIngestionService.On("ProcessSACBalanceChanges",
+			mock.Anything, mock.Anything, mock.Anything,
 		).Return(nil).Maybe()
 		mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
 			mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
@@ -3265,8 +3297,11 @@ func Test_persistLedgerData_SiblingFailureRollsBackEverything(t *testing.T) {
 	require.NoError(t, err)
 
 	mockTokenIngestionService := NewTokenIngestionServiceMock(t)
-	mockTokenIngestionService.On("ProcessTrustlineAndSACChanges",
-		mock.Anything, mock.Anything, mock.Anything, mock.Anything,
+	mockTokenIngestionService.On("ProcessTrustlineChanges",
+		mock.Anything, mock.Anything, mock.Anything,
+	).Return(nil).Maybe()
+	mockTokenIngestionService.On("ProcessSACBalanceChanges",
+		mock.Anything, mock.Anything, mock.Anything,
 	).Return(nil).Maybe()
 	mockTokenIngestionService.On("ProcessNativeAndPoolChanges",
 		mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything,
diff --git a/internal/services/mocks.go b/internal/services/mocks.go
index 7a761140d..3612e1a4d 100644
--- a/internal/services/mocks.go
+++ b/internal/services/mocks.go
@@ -126,8 +126,13 @@ type TokenIngestionServiceMock struct {
 
 var _ TokenIngestionService = (*TokenIngestionServiceMock)(nil)
 
-func (m *TokenIngestionServiceMock) ProcessTrustlineAndSACChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error {
-	args := m.Called(ctx, dbTx, trustlineChangesByTrustlineKey, sacBalanceChangesByKey)
+func (m *TokenIngestionServiceMock) ProcessTrustlineChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange) error {
+	args := m.Called(ctx, dbTx, trustlineChangesByTrustlineKey)
+	return args.Error(0)
+}
+
+func (m *TokenIngestionServiceMock) ProcessSACBalanceChanges(ctx context.Context, dbTx pgx.Tx, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error {
+	args := m.Called(ctx, dbTx, sacBalanceChangesByKey)
 	return args.Error(0)
 }
 
diff --git a/internal/services/token_ingestion.go b/internal/services/token_ingestion.go
index 05efba0c6..3c929d541 100644
--- a/internal/services/token_ingestion.go
+++ b/internal/services/token_ingestion.go
@@ -15,20 +15,25 @@ import (
 )
 
 // TokenIngestionService provides write access to account token storage during live ingestion.
-// The balance writes are split by foreign-key dependence: trustline and SAC balances reference
-// parent rows (trustline_assets, contract_tokens) that live persist stages uncommitted on the
-// coordinating transaction, so they must run on that same transaction; native-balance and
-// liquidity-pool tables carry no foreign keys, so their writes may run on any transaction —
-// live persist gives them their own sibling, concurrent with the coordinator.
+// The balance writes are split by where their foreign-key parents are staged: each balance
+// family runs on the transaction that carries its parent rows, so every FK check passes at
+// that transaction's own commit. Trustline balances ride the sibling that inserts
+// trustline_assets, pool-share balances ride the sibling that upserts liquidity_pools, and
+// SAC balances stay on the coordinating transaction because their parent (contract_tokens)
+// is also written by the classification path there — splitting those writers across
+// concurrent transactions could deadlock at the commit barrier on a same-key insert.
 type TokenIngestionService interface {
-	// ProcessTrustlineAndSACChanges applies trustline and SAC balance changes. dbTx must be
-	// the transaction on which this ledger's trustline_assets and contract_tokens rows are
-	// staged: both tables' foreign keys are checked against committed state, and a
-	// transaction that commits before those parents would fail with SQLSTATE 23503.
-	ProcessTrustlineAndSACChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error
-	// ProcessNativeAndPoolChanges applies native-balance and liquidity-pool changes. The
-	// target tables have no foreign keys, and every upsert is idempotent (ON CONFLICT with
-	// an IS DISTINCT FROM guard), so the writes are safe on any transaction.
+	// ProcessTrustlineChanges applies trustline balance changes. dbTx must be the
+	// transaction on which this ledger's trustline_assets rows are staged: the balance
+	// rows' foreign key is checked against committed state at commit, and a transaction
+	// committing before its parents would fail with SQLSTATE 23503.
+	ProcessTrustlineChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange) error
+	// ProcessSACBalanceChanges applies SAC balance changes. dbTx must be the transaction
+	// on which this ledger's contract_tokens rows are staged.
+	ProcessSACBalanceChanges(ctx context.Context, dbTx pgx.Tx, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error
+	// ProcessNativeAndPoolChanges applies native-balance and liquidity-pool changes.
+	// Pools are upserted before pool-share balances, so the share rows' foreign key is
+	// satisfied within this same transaction; native balances have no parent.
 	ProcessNativeAndPoolChanges(ctx context.Context, dbTx pgx.Tx, accountChangesByAccountID map[string]types.AccountChange, lpShareChangesByKey map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange, lpChangesByPoolID map[string]types.LiquidityPoolChange) error
 }
 
@@ -67,16 +72,16 @@ func NewTokenIngestionService(cfg TokenIngestionServiceConfig) *tokenIngestionSe
 	}
 }
 
-// ProcessTrustlineAndSACChanges applies trustline and SAC balance changes on the
-// transaction that stages their trustline_assets / contract_tokens parent rows.
-func (s *tokenIngestionService) ProcessTrustlineAndSACChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error {
-	if err := s.processTrustlineChanges(ctx, dbTx, trustlineChangesByTrustlineKey); err != nil {
-		return err
-	}
-	if err := s.processSACBalanceChanges(ctx, dbTx, sacBalanceChangesByKey); err != nil {
-		return err
-	}
-	return nil
+// ProcessTrustlineChanges applies trustline balance changes on the transaction
+// that stages their trustline_assets parent rows.
+func (s *tokenIngestionService) ProcessTrustlineChanges(ctx context.Context, dbTx pgx.Tx, trustlineChangesByTrustlineKey map[indexer.TrustlineChangeKey]types.TrustlineChange) error {
+	return s.processTrustlineChanges(ctx, dbTx, trustlineChangesByTrustlineKey)
+}
+
+// ProcessSACBalanceChanges applies SAC balance changes on the transaction that
+// stages their contract_tokens parent rows.
+func (s *tokenIngestionService) ProcessSACBalanceChanges(ctx context.Context, dbTx pgx.Tx, sacBalanceChangesByKey map[indexer.SACBalanceChangeKey]types.SACBalanceChange) error {
+	return s.processSACBalanceChanges(ctx, dbTx, sacBalanceChangesByKey)
 }
 
 // ProcessNativeAndPoolChanges applies native-balance and liquidity-pool changes; the
diff --git a/internal/services/token_ingestion_test.go b/internal/services/token_ingestion_test.go
index 470778daf..429c4cca0 100644
--- a/internal/services/token_ingestion_test.go
+++ b/internal/services/token_ingestion_test.go
@@ -369,9 +369,12 @@ func TestProcessBalanceChanges(t *testing.T) {
 		})
 
 		err = db.RunInTransaction(ctx, dbConnectionPool, func(dbTx pgx.Tx) error {
-			if tlErr := service.ProcessTrustlineAndSACChanges(ctx, dbTx, map[indexer.TrustlineChangeKey]types.TrustlineChange{}, make(map[indexer.SACBalanceChangeKey]types.SACBalanceChange)); tlErr != nil {
+			if tlErr := service.ProcessTrustlineChanges(ctx, dbTx, map[indexer.TrustlineChangeKey]types.TrustlineChange{}); tlErr != nil {
 				return tlErr
 			}
+			if sacErr := service.ProcessSACBalanceChanges(ctx, dbTx, make(map[indexer.SACBalanceChangeKey]types.SACBalanceChange)); sacErr != nil {
+				return sacErr
+			}
 			return service.ProcessNativeAndPoolChanges(ctx, dbTx, make(map[string]types.AccountChange), make(map[indexer.LiquidityPoolShareChangeKey]types.LiquidityPoolShareChange), make(map[string]types.LiquidityPoolChange))
 		})
 		assert.NoError(t, err)

From 13649dea9f3f850056164ae2afe42878ddc6737b Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Thu, 20 Aug 2026 15:24:30 -0400
Subject: [PATCH 09/11] fix(ingest): resolve event membership on the persist
 transaction, not the pool
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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.
---
 internal/data/mocks.go                   |  4 +-
 internal/data/protocol_contracts.go      | 11 ++++--
 internal/data/protocol_contracts_test.go |  4 +-
 internal/services/ingest_live.go         |  6 ++-
 internal/services/ingest_test.go         | 48 +++++++++++++++++++++++-
 5 files changed, 62 insertions(+), 11 deletions(-)

diff --git a/internal/data/mocks.go b/internal/data/mocks.go
index 61fb4a225..7bda0dcb4 100644
--- a/internal/data/mocks.go
+++ b/internal/data/mocks.go
@@ -373,8 +373,8 @@ func (m *ProtocolContractsModelMock) GetByProtocolID(ctx context.Context, q db.Q
 	return args.Get(0).([]ProtocolContracts), args.Error(1)
 }
 
-func (m *ProtocolContractsModelMock) BatchGetByContractIDs(ctx context.Context, contractIDs [][]byte) (map[string][]ProtocolContracts, error) {
-	args := m.Called(ctx, contractIDs)
+func (m *ProtocolContractsModelMock) BatchGetByContractIDs(ctx context.Context, q db.Querier, contractIDs [][]byte) (map[string][]ProtocolContracts, error) {
+	args := m.Called(ctx, q, contractIDs)
 	if args.Get(0) == nil {
 		return nil, args.Error(1)
 	}
diff --git a/internal/data/protocol_contracts.go b/internal/data/protocol_contracts.go
index 404191555..c09b6d2d1 100644
--- a/internal/data/protocol_contracts.go
+++ b/internal/data/protocol_contracts.go
@@ -27,7 +27,7 @@ type ProtocolContracts struct {
 type ProtocolContractsModelInterface interface {
 	BatchInsert(ctx context.Context, dbTx pgx.Tx, contracts []ProtocolContracts) error
 	GetByProtocolID(ctx context.Context, q db.Querier, protocolID string) ([]ProtocolContracts, error)
-	BatchGetByContractIDs(ctx context.Context, contractIDs [][]byte) (map[string][]ProtocolContracts, error)
+	BatchGetByContractIDs(ctx context.Context, q db.Querier, contractIDs [][]byte) (map[string][]ProtocolContracts, error)
 	GetByWasmHashes(ctx context.Context, wasmHashes []types.HashBytea) ([]ProtocolContracts, error)
 }
 
@@ -143,8 +143,11 @@ func (m *ProtocolContractsModel) GetByWasmHashes(ctx context.Context, wasmHashes
 // BatchGetByContractIDs returns, for the given contract IDs, the protocol_contracts
 // rows whose wasm is classified to a protocol, grouped by protocol ID. Live ingestion
 // uses this to resolve membership for only the contracts that emitted events in a
-// ledger, instead of loading every committed contract for a protocol.
-func (m *ProtocolContractsModel) BatchGetByContractIDs(ctx context.Context, contractIDs [][]byte) (map[string][]ProtocolContracts, error) {
+// ledger, instead of loading every committed contract for a protocol. q must be the
+// persist transaction when rows it staged must be visible: a batched persist
+// classifies contracts at the batch head, and later ledgers of the same uncommitted
+// batch resolve membership through this lookup.
+func (m *ProtocolContractsModel) BatchGetByContractIDs(ctx context.Context, q db.Querier, contractIDs [][]byte) (map[string][]ProtocolContracts, error) {
 	if len(contractIDs) == 0 {
 		return nil, nil
 	}
@@ -157,7 +160,7 @@ func (m *ProtocolContractsModel) BatchGetByContractIDs(ctx context.Context, cont
 	`
 
 	start := time.Now()
-	rows, err := m.DB.Query(ctx, query, contractIDs)
+	rows, err := q.Query(ctx, query, contractIDs)
 	duration := time.Since(start).Seconds()
 	m.Metrics.QueryDuration.WithLabelValues("BatchGetByContractIDs", "protocol_contracts").Observe(duration)
 	m.Metrics.QueriesTotal.WithLabelValues("BatchGetByContractIDs", "protocol_contracts").Inc()
diff --git a/internal/data/protocol_contracts_test.go b/internal/data/protocol_contracts_test.go
index 53863cb09..c605f4af3 100644
--- a/internal/data/protocol_contracts_test.go
+++ b/internal/data/protocol_contracts_test.go
@@ -246,7 +246,7 @@ func TestProtocolContractsBatchGetByContractIDs(t *testing.T) {
 
 	t.Run("empty input returns nil", func(t *testing.T) {
 		cleanUpDB()
-		result, qErr := model.BatchGetByContractIDs(ctx, nil)
+		result, qErr := model.BatchGetByContractIDs(ctx, dbConnectionPool, nil)
 		require.NoError(t, qErr)
 		assert.Nil(t, result)
 	})
@@ -275,7 +275,7 @@ func TestProtocolContractsBatchGetByContractIDs(t *testing.T) {
 			require.NoError(t, vErr)
 			return v.([]byte)
 		}
-		result, qErr := model.BatchGetByContractIDs(ctx, [][]byte{toBytes(cA), toBytes(cB), toBytes(cC), toBytes(cMissing)})
+		result, qErr := model.BatchGetByContractIDs(ctx, dbConnectionPool, [][]byte{toBytes(cA), toBytes(cB), toBytes(cC), toBytes(cMissing)})
 		require.NoError(t, qErr)
 
 		require.Len(t, result, 2)
diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index fcbb0a379..26e18e41b 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -350,10 +350,14 @@ func (m *ingestService) stageCoordinatedWrites(
 		// this ledger. One bounded query serves every protocol; ledgers with no
 		// contract events skip it entirely. The buffered overlay (below) covers
 		// contracts deployed or upgraded this ledger, which are not yet committed.
+		// The lookup runs on the coordinating transaction, not the pool: in a
+		// batched persist, contracts classified at the batch head are staged on
+		// this still-uncommitted transaction, and a mid-batch ledger's membership
+		// must include them or its events would be silently dropped.
 		var committedByProtocol map[string][]data.ProtocolContracts
 		if eventContractIDs := distinctEventContractIDs(contractEvents); len(eventContractIDs) > 0 {
 			var lookupErr error
-			committedByProtocol, lookupErr = m.models.ProtocolContracts.BatchGetByContractIDs(ctx, eventContractIDs)
+			committedByProtocol, lookupErr = m.models.ProtocolContracts.BatchGetByContractIDs(ctx, dbTx, eventContractIDs)
 			if lookupErr != nil {
 				return fmt.Errorf("resolving protocol contracts for ledger %d: %w", ledgerSeq, lookupErr)
 			}
diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go
index d0c0e8118..c4e3fb184 100644
--- a/internal/services/ingest_test.go
+++ b/internal/services/ingest_test.go
@@ -2542,7 +2542,7 @@ func Test_persistLedgerData_ProtocolCASGating(t *testing.T) {
 
 		// Inject a failing contract-id lookup over the otherwise-real models.
 		contractsMock := data.NewProtocolContractsModelMock(t)
-		contractsMock.On("BatchGetByContractIDs", mock.Anything, mock.Anything).
+		contractsMock.On("BatchGetByContractIDs", mock.Anything, mock.Anything, mock.Anything).
 			Return(nil, fmt.Errorf("db boom")).Once()
 		models.ProtocolContracts = contractsMock
 
@@ -3177,7 +3177,7 @@ func Test_ingestService_ingestLiveLedgers_StageErrorStopsPipeline(t *testing.T)
 // succeeds, so a pre-commit failure leaves the database exactly as it was and the ledger fully
 // retryable.
 func Test_persistLedgerData_Batch(t *testing.T) {
-	setupTest := func(t *testing.T) (context.Context, *ingestService, *pgxpool.Pool) {
+	setupTest := func(t *testing.T, processors ...ProtocolProcessor) (context.Context, *ingestService, *pgxpool.Pool) {
 		t.Helper()
 		dbt := dbtest.Open(t)
 		t.Cleanup(func() { dbt.Close() })
@@ -3211,6 +3211,7 @@ func Test_persistLedgerData_Batch(t *testing.T) {
 			Network:               network.TestNetworkPassphrase,
 			NetworkPassphrase:     network.TestNetworkPassphrase,
 			Archive:               &HistoryArchiveMock{},
+			ProtocolProcessors:    processors,
 		})
 		require.NoError(t, err)
 		return ctx, svc, pool
@@ -3278,6 +3279,49 @@ func Test_persistLedgerData_Batch(t *testing.T) {
 			`SELECT value FROM ingest_store WHERE key = $1`, data.LatestLedgerCursorName).Scan(&cursor))
 		assert.Equal(t, "99", cursor, "the cursor must stay below the whole batch")
 	})
+
+	t.Run("a mid-batch ledger's membership sees contracts classified at the batch head", func(t *testing.T) {
+		var wRaw, cRaw [32]byte
+		wRaw[0], cRaw[0] = 0xA7, 0xC7
+		wasmHash := types.HashBytea(hex.EncodeToString(wRaw[:]))
+		contractHex := types.HashBytea(hex.EncodeToString(cRaw[:]))
+
+		processor := &testProtocolProcessor{id: "testproto"}
+		ctx, svc, pool := setupTest(t, processor)
+		processor.ingestStore = svc.models.IngestStore
+		setupDBCursors(t, ctx, pool, 99, 99)
+		setupProtocolCursors(t, ctx, pool, 99, 99)
+		require.NoError(t, svc.snapshotProtocolCursors(ctx))
+
+		// protocol_wasms.protocol_id is an FK into protocols.
+		_, err := pool.Exec(ctx, `INSERT INTO protocols (id) VALUES ('testproto')`)
+		require.NoError(t, err)
+
+		// The batch head uploads the wasm and deploys contract C, classified to
+		// testproto by the head's plan.
+		head := batchItem(100)
+		head.buffer.PushProtocolWasm(data.ProtocolWasms{WasmHash: wasmHash})
+		head.buffer.PushProtocolContracts(data.ProtocolContracts{ContractID: contractHex, WasmHash: wasmHash})
+		head.plan = &ClassificationPlan{Matches: map[types.HashBytea]string{wasmHash: "testproto"}}
+
+		// The mid-batch ledger carries only an event from C: nothing re-buffers
+		// the contract, so its membership can come only from the rows the head
+		// staged on the still-uncommitted coordinating transaction.
+		mid := batchItem(101)
+		contractID := xdr.ContractId(cRaw)
+		mid.buffer.PushContractEvents(
+			indexer.ContractEventKey{TxIdx: 0, OpIdx: 0},
+			[]xdr.ContractEvent{{Type: xdr.ContractEventTypeContract, ContractId: &contractID}},
+		)
+
+		require.NoError(t, svc.persistLedgerData(ctx, []persistItem{head, mid}))
+
+		// ProcessLedger ran last for the mid-batch ledger; its membership must
+		// include the head-classified contract.
+		assert.Equal(t, uint32(101), processor.processedLedger)
+		require.Len(t, processor.lastContracts, 1)
+		assert.Equal(t, contractHex, processor.lastContracts[0].ContractID)
+	})
 }
 
 func Test_persistLedgerData_SiblingFailureRollsBackEverything(t *testing.T) {

From db30f9cad8e135829b26163eb0313bb37b44750b Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Thu, 20 Aug 2026 15:32:24 -0400
Subject: [PATCH 10/11] refactor(ingest): return only error from the link-table
 COPYs; pin memo semantics
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

BatchCopyAccounts on transactions and operations returned a link-row
count every caller discarded — the count is already observed into the
BatchSize metric inside the method — so both now return plain error,
matching the balance models' BatchCopy shape.

contractDataMemo's contract gets a unit test: get() always yields a
rangeable non-nil map, and the extraction walk runs at most once no
matter how many retry attempts share the memo.

The startup-reconciliation comment now says a crash orphans at most the
persist batch past the cursor; batching had outgrown "the single
ledger".
---
 internal/data/operations.go           | 18 +++++++++---------
 internal/data/operations_test.go      |  4 ++--
 internal/data/transactions.go         | 18 +++++++++---------
 internal/data/transactions_test.go    |  4 ++--
 internal/services/ingest.go           |  6 ++----
 internal/services/ingest_live.go      |  8 ++++----
 internal/services/ingest_live_test.go | 25 +++++++++++++++++++++++++
 7 files changed, 53 insertions(+), 30 deletions(-)

diff --git a/internal/data/operations.go b/internal/data/operations.go
index c0b017059..16ffcd110 100644
--- a/internal/data/operations.go
+++ b/internal/data/operations.go
@@ -417,9 +417,9 @@ func (m *OperationModel) BatchCopy(
 }
 
 // BatchCopyAccounts inserts the operations_accounts links using pgx's binary
-// COPY protocol, and returns the number of link rows written. operations
-// supplies each operation's ledger_created_at, the link table's partition
-// column; the two arguments come from the same buffer and are read only.
+// COPY protocol. operations supplies each operation's ledger_created_at, the
+// link table's partition column; the two arguments come from the same buffer
+// and are read only.
 //
 // IMPORTANT: like BatchCopy, this FAILS on duplicates — COPY has no conflict
 // handling.
@@ -428,9 +428,9 @@ func (m *OperationModel) BatchCopyAccounts(
 	pgxTx pgx.Tx,
 	operations []*types.Operation,
 	stellarAddressesByOpID map[int64]map[string]struct{},
-) (int, error) {
+) error {
 	if len(stellarAddressesByOpID) == 0 {
-		return 0, nil
+		return nil
 	}
 
 	start := time.Now()
@@ -450,14 +450,14 @@ func (m *OperationModel) BatchCopyAccounts(
 			// A silent miss would COPY a zero timestamp — a year-0001 chunk in
 			// the hypertable — and means the caller's operations and
 			// participants disagree about which IDs exist.
-			return 0, fmt.Errorf("no operation supplies ledger_created_at for operation_id %d", opID)
+			return fmt.Errorf("no operation supplies ledger_created_at for operation_id %d", opID)
 		}
 		ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
 		opIDPgtype := pgtype.Int8{Int64: opID, Valid: true}
 		for addr := range addresses {
 			addrBytes, addrErr := types.AddressBytea(addr).Value()
 			if addrErr != nil {
-				return 0, fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
+				return fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
 			}
 			oaRows = append(oaRows, []any{
 				ledgerCreatedAtPgtype,
@@ -479,8 +479,8 @@ func (m *OperationModel) BatchCopyAccounts(
 	m.Metrics.QueriesTotal.WithLabelValues("BatchCopyAccounts", "operations_accounts").Inc()
 	if err != nil {
 		m.Metrics.QueryErrors.WithLabelValues("BatchCopyAccounts", "operations_accounts", utils.GetDBErrorType(err)).Inc()
-		return 0, fmt.Errorf("pgx CopyFrom operations_accounts: %w", err)
+		return fmt.Errorf("pgx CopyFrom operations_accounts: %w", err)
 	}
 
-	return len(oaRows), nil
+	return nil
 }
diff --git a/internal/data/operations_test.go b/internal/data/operations_test.go
index 2d49d1981..9ed3cf11a 100644
--- a/internal/data/operations_test.go
+++ b/internal/data/operations_test.go
@@ -149,7 +149,7 @@ func Test_OperationModel_BatchCopy(t *testing.T) {
 
 			gotCount, err := m.BatchCopy(ctx, pgxTx, tc.operations)
 			if err == nil {
-				_, err = m.BatchCopyAccounts(ctx, pgxTx, tc.operations, tc.stellarAddressesByOpID)
+				err = m.BatchCopyAccounts(ctx, pgxTx, tc.operations, tc.stellarAddressesByOpID)
 			}
 
 			if tc.wantErrContains != "" {
@@ -747,7 +747,7 @@ func BenchmarkOperationModel_BatchCopy(b *testing.B) {
 					pgxTx.Rollback(ctx)
 					b.Fatalf("BatchCopy failed: %v", err)
 				}
-				_, err = m.BatchCopyAccounts(ctx, pgxTx, ops, addressesByOpID)
+				err = m.BatchCopyAccounts(ctx, pgxTx, ops, addressesByOpID)
 				if err != nil {
 					pgxTx.Rollback(ctx)
 					b.Fatalf("BatchCopyAccounts failed: %v", err)
diff --git a/internal/data/transactions.go b/internal/data/transactions.go
index 6780fb6ba..362a862c4 100644
--- a/internal/data/transactions.go
+++ b/internal/data/transactions.go
@@ -269,9 +269,9 @@ func (m *TransactionModel) BatchCopy(
 }
 
 // BatchCopyAccounts inserts the transactions_accounts links using pgx's binary
-// COPY protocol, and returns the number of link rows written. txs supplies each
-// transaction's ledger_created_at, the link table's partition column; the two
-// arguments come from the same buffer and are read only.
+// COPY protocol. txs supplies each transaction's ledger_created_at, the link
+// table's partition column; the two arguments come from the same buffer and
+// are read only.
 //
 // IMPORTANT: like BatchCopy, this FAILS on duplicates — COPY has no conflict
 // handling.
@@ -280,9 +280,9 @@ func (m *TransactionModel) BatchCopyAccounts(
 	pgxTx pgx.Tx,
 	txs []*types.Transaction,
 	stellarAddressesByToID map[int64]map[string]struct{},
-) (int, error) {
+) error {
 	if len(stellarAddressesByToID) == 0 {
-		return 0, nil
+		return nil
 	}
 
 	start := time.Now()
@@ -302,14 +302,14 @@ func (m *TransactionModel) BatchCopyAccounts(
 			// A silent miss would COPY a zero timestamp — a year-0001 chunk in
 			// the hypertable — and means the caller's transactions and
 			// participants disagree about which ToIDs exist.
-			return 0, fmt.Errorf("no transaction supplies ledger_created_at for to_id %d", toID)
+			return fmt.Errorf("no transaction supplies ledger_created_at for to_id %d", toID)
 		}
 		ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
 		toIDPgtype := pgtype.Int8{Int64: toID, Valid: true}
 		for addr := range addresses {
 			addrBytes, addrErr := types.AddressBytea(addr).Value()
 			if addrErr != nil {
-				return 0, fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
+				return fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
 			}
 			taRows = append(taRows, []any{
 				ledgerCreatedAtPgtype,
@@ -331,8 +331,8 @@ func (m *TransactionModel) BatchCopyAccounts(
 	m.Metrics.QueriesTotal.WithLabelValues("BatchCopyAccounts", "transactions_accounts").Inc()
 	if err != nil {
 		m.Metrics.QueryErrors.WithLabelValues("BatchCopyAccounts", "transactions_accounts", utils.GetDBErrorType(err)).Inc()
-		return 0, fmt.Errorf("pgx CopyFrom transactions_accounts: %w", err)
+		return fmt.Errorf("pgx CopyFrom transactions_accounts: %w", err)
 	}
 
-	return len(taRows), nil
+	return nil
 }
diff --git a/internal/data/transactions_test.go b/internal/data/transactions_test.go
index 2863cba1a..1c205c7ce 100644
--- a/internal/data/transactions_test.go
+++ b/internal/data/transactions_test.go
@@ -147,7 +147,7 @@ func Test_TransactionModel_BatchCopy(t *testing.T) {
 
 			gotCount, err := m.BatchCopy(ctx, pgxTx, tc.txs)
 			if err == nil {
-				_, err = m.BatchCopyAccounts(ctx, pgxTx, tc.txs, tc.stellarAddressesByToID)
+				err = m.BatchCopyAccounts(ctx, pgxTx, tc.txs, tc.stellarAddressesByToID)
 			}
 
 			if tc.wantErrContains != "" {
@@ -471,7 +471,7 @@ func BenchmarkTransactionModel_BatchCopy(b *testing.B) {
 					pgxTx.Rollback(ctx)
 					b.Fatalf("BatchCopy failed: %v", err)
 				}
-				_, err = m.BatchCopyAccounts(ctx, pgxTx, txs, addressesByToID)
+				err = m.BatchCopyAccounts(ctx, pgxTx, txs, addressesByToID)
 				if err != nil {
 					pgxTx.Rollback(ctx)
 					b.Fatalf("BatchCopyAccounts failed: %v", err)
diff --git a/internal/services/ingest.go b/internal/services/ingest.go
index 81547a65c..aa80ffb10 100644
--- a/internal/services/ingest.go
+++ b/internal/services/ingest.go
@@ -312,8 +312,7 @@ func (m *ingestService) insertTransactionsAccounts(ctx context.Context, pgxTx pg
 	if len(stellarAddressesByToID) == 0 {
 		return nil
 	}
-	_, err := m.models.Transactions.BatchCopyAccounts(ctx, pgxTx, txs, stellarAddressesByToID)
-	if err != nil {
+	if err := m.models.Transactions.BatchCopyAccounts(ctx, pgxTx, txs, stellarAddressesByToID); err != nil {
 		return fmt.Errorf("batch inserting transactions accounts: %w", err)
 	}
 	return nil
@@ -336,8 +335,7 @@ func (m *ingestService) insertOperationsAccounts(ctx context.Context, pgxTx pgx.
 	if len(stellarAddressesByOpID) == 0 {
 		return nil
 	}
-	_, err := m.models.Operations.BatchCopyAccounts(ctx, pgxTx, ops, stellarAddressesByOpID)
-	if err != nil {
+	if err := m.models.Operations.BatchCopyAccounts(ctx, pgxTx, ops, stellarAddressesByOpID); err != nil {
 		return fmt.Errorf("batch inserting operations accounts: %w", err)
 	}
 	return nil
diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index 26e18e41b..26f5d2a4a 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -563,11 +563,11 @@ func (m *ingestService) startLiveIngestion(ctx context.Context) error {
 		m.appMetrics.Ingestion.LatestLedger.Set(float64(startLedger))
 		m.appMetrics.Ingestion.OldestLedger.Set(float64(startLedger))
 	} else {
-		// Remove any bulk rows a crashed run left above the cursor before that
-		// ledger is re-ingested: persistLedgerData commits the sibling COPY
+		// Remove any bulk rows a crashed run left above the cursor before those
+		// ledgers are 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
+		// cursor, so a crash between those commits orphans (at most) the persist
+		// batch past the cursor. Fatal on failure — ingesting over the orphans
 		// would collide on the bulk tables' primary keys anyway.
 		if err := m.models.IngestStore.DeleteRowsAboveLedger(ctx, latestIngestedLedger); err != nil {
 			return fmt.Errorf("reconciling bulk rows above cursor ledger %d: %w", latestIngestedLedger, err)
diff --git a/internal/services/ingest_live_test.go b/internal/services/ingest_live_test.go
index eafa6fabd..23647af04 100644
--- a/internal/services/ingest_live_test.go
+++ b/internal/services/ingest_live_test.go
@@ -146,3 +146,28 @@ func Test_isPermanentPersistError(t *testing.T) {
 		})
 	}
 }
+
+// Test_contractDataMemo_get covers the memo's two guarantees. The result is
+// always non-nil: a RequiresContractData processor ranges over the map
+// unconditionally, so even a ledger with zero transactions has to yield an
+// empty map rather than nil. And the extraction walk runs at most once no
+// matter how often get() is called, which is what makes one memo safe to
+// share across every persistLedgerDataWithRetry attempt for a ledger.
+func Test_contractDataMemo_get(t *testing.T) {
+	memo := newContractDataMemo(nil, 100)
+
+	first, err := memo.get()
+	require.NoError(t, err)
+	require.NotNil(t, first, "a zero-transaction ledger still needs a rangeable map")
+	assert.Empty(t, first)
+
+	// Marking the first result is what makes the second call's map
+	// identifiable: a second extraction builds a fresh map and could not carry
+	// the marker, so finding it proves the walk did not run again.
+	const marker = "extracted-once"
+	first[marker] = nil
+
+	second, err := memo.get()
+	require.NoError(t, err)
+	assert.Contains(t, second, marker, "get should serve the memoized map instead of extracting again")
+}

From 893685d55295221df00db3f20139158c0cdf30b6 Mon Sep 17 00:00:00 2001
From: Aditya Vyas <aditya.vyas@stellar.org>
Date: Fri, 21 Aug 2026 17:01:24 -0400
Subject: [PATCH 11/11] fix(ingest): route protocol history through the
 state_changes sibling
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Protocol history rows are state_changes rows, so PersistHistory on the
coordinating transaction had two of our own transactions inserting into the
same hypertable concurrently. At a chunk boundary TimescaleDB serializes
chunk creation and the waiter blocks on the other transaction's end — but
the coordinator only commits after every sibling goroutine returns, so if
the sibling is the waiter the pipeline hangs forever, invisible to
Postgres's deadlock detector.

All state_changes writes now go through the one state_changes sibling
transaction, serialized by a mutex (pgx.Tx is not concurrency-safe). The
CAS stays on the coordinator: siblings commit strictly first, so a
committed cursor still implies committed history rows, and rows stranded
above the cursor by a crash are already removed by DeleteRowsAboveLedger.
---
 internal/services/blend/processor.go    |  2 +-
 internal/services/ingest_live.go        | 60 ++++++++++++++++++++++---
 internal/services/protocol_processor.go | 10 +++--
 internal/services/sep41/processor.go    |  2 +-
 4 files changed, 63 insertions(+), 11 deletions(-)

diff --git a/internal/services/blend/processor.go b/internal/services/blend/processor.go
index 38c75417c..0de878fa4 100644
--- a/internal/services/blend/processor.go
+++ b/internal/services/blend/processor.go
@@ -905,7 +905,7 @@ func (p *processor) Reset() {
 	p.needsReset = false
 }
 
-// PersistHistory writes staged state changes inside the CAS transaction.
+// PersistHistory writes staged state changes on the provided transaction.
 func (p *processor) PersistHistory(ctx context.Context, dbTx pgx.Tx) error {
 	p.needsReset = true
 	if len(p.stagedStateChanges) == 0 {
diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index 26f5d2a4a..82b46e034 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -91,6 +92,24 @@ type persistItem struct {
 	buffer       *indexer.IndexerBuffer
 }
 
+// protocolHistorySink is where stageCoordinatedWrites sends the protocol
+// processors' history rows: the state_changes sibling transaction, serialized
+// by the same mutex as the sibling's own COPYs (pgx.Tx is not safe for
+// concurrent use). History rows are state_changes rows, and no table may be
+// written by two concurrent transactions — see the chunk-boundary deadlock
+// note at the sibling definitions in persistLedgerData.
+type protocolHistorySink struct {
+	dbTx pgx.Tx
+	mu   *sync.Mutex
+}
+
+// persist runs one processor's PersistHistory on the sink under its mutex.
+func (h protocolHistorySink) persist(ctx context.Context, processor ProtocolProcessor) error {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return processor.PersistHistory(ctx, h.dbTx) //nolint:wrapcheck // the call site wraps with protocol and ledger context
+}
+
 // batchLabel names a batch in errors and logs: "ledger N" for a single
 // ledger, "ledgers N-M" for a coalesced batch.
 func batchLabel(items []persistItem) string {
@@ -107,7 +126,8 @@ func batchLabel(items []persistItem) string {
 // another) stream concurrently on sibling connections, each in its own
 // transaction covering every ledger in the batch, while the coordinating
 // transaction stages everything else (contracts, classification, protocol
-// state, SAC balances, cursor) ledger by ledger in order — the per-protocol
+// current state, SAC balances, cursor — protocol history rows ride the
+// state_changes sibling) ledger by ledger in order — the per-protocol
 // CAS chain advances N-1 → N inside the transaction, and the guarded
 // cursor's final value is the batch's last ledger. All the slow work
 // happens uncommitted and invisible; only after every stream and the
@@ -140,6 +160,15 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 		}
 	}()
 
+	// The protocol processors' history rows are state_changes rows too, and two
+	// transactions inserting into the same hypertable can deadlock undetectably
+	// at a chunk boundary (TimescaleDB serializes chunk creation while the
+	// coordinating goroutine is blocked in Go, invisible to Postgres's deadlock
+	// detector). So every state_changes write — the ledger's own rows on the
+	// sibling goroutine and protocol history on the coordinating goroutine —
+	// goes through the one state_changes sibling transaction, serialized by
+	// stateChangesMu because pgx.Tx is not safe for concurrent use.
+	var stateChangesMu sync.Mutex
 	siblings := []struct {
 		name string
 		run  func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error
@@ -157,6 +186,8 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 			return m.insertOperationsAccounts(ctx, dbTx, it.buffer.GetOperations(), it.buffer.GetOperationsParticipants())
 		}},
 		{"state_changes", func(ctx context.Context, dbTx pgx.Tx, it *persistItem) error {
+			stateChangesMu.Lock()
+			defer stateChangesMu.Unlock()
 			return m.insertStateChanges(ctx, dbTx, it.buffer.GetStateChanges())
 		}},
 		// Each balance family rides the transaction that stages its FK parents,
@@ -212,10 +243,19 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 	}
 
 	// Stream the sibling writes and stage the coordinated writes concurrently.
-	// Every sibling owns a disjoint set of tables with no FKs among them or to
-	// any other transaction's tables, and the goroutines only read the
+	// No table is ever written by two transactions: every sibling owns a
+	// disjoint set of tables with no FKs among them or to any other
+	// transaction's tables, and the coordinating goroutine's one write outside
+	// its own set — protocol history, which is state_changes rows — goes to the
+	// state_changes sibling under stateChangesMu. The goroutines only read the
 	// quiescent buffers, so the transactions never contend. Within each
 	// transaction the batch's ledgers run in order.
+	var stateChangesTx pgx.Tx
+	for i, s := range siblings {
+		if s.name == "state_changes" {
+			stateChangesTx = siblingTxs[i]
+		}
+	}
 	g, gctx := errgroup.WithContext(ctx)
 	for i, s := range siblings {
 		g.Go(func() error {
@@ -227,10 +267,11 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 			return nil
 		})
 	}
+	history := protocolHistorySink{dbTx: stateChangesTx, mu: &stateChangesMu}
 	g.Go(func() error {
 		for j := range items {
 			it := &items[j]
-			if stageErr := m.stageCoordinatedWrites(gctx, coordTx, it.seq, it.meta, it.plan, it.contractData, it.buffer); stageErr != nil {
+			if stageErr := m.stageCoordinatedWrites(gctx, coordTx, history, it.seq, it.meta, it.plan, it.contractData, it.buffer); stageErr != nil {
 				return fmt.Errorf("staging coordinated writes for ledger %d: %w", it.seq, stageErr)
 			}
 		}
@@ -268,10 +309,17 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt
 // stageCoordinatedWrites runs every per-ledger write the siblings don't own
 // on the coordinating transaction: SAC contract tokens, protocol
 // classification and wasm/contract rows, CAS-gated protocol state, the SAC
-// balance changes, and finally the guarded cursor.
+// balance changes, and finally the guarded cursor. The one exception is
+// protocol history — those are state_changes rows, so they go through
+// history (the state_changes sibling) rather than this transaction. Their
+// CAS stays here: the siblings commit strictly before this transaction, so a
+// committed cursor still implies committed history rows, and a crash in
+// between leaves only rows above the cursor, which DeleteRowsAboveLedger
+// removes at startup like any other state_changes orphan.
 func (m *ingestService) stageCoordinatedWrites(
 	ctx context.Context,
 	dbTx pgx.Tx,
+	history protocolHistorySink,
 	ledgerSeq uint32,
 	ledgerMeta xdr.LedgerCloseMeta,
 	plan *ClassificationPlan,
@@ -445,7 +493,7 @@ func (m *ingestService) stageCoordinatedWrites(
 
 			if historySwapped {
 				persistStart := time.Now()
-				persistErr := processor.PersistHistory(ctx, dbTx)
+				persistErr := history.persist(ctx, processor)
 				m.appMetrics.Ingestion.ProtocolStateProcessingDuration.WithLabelValues(protocolID, "persist_history").Observe(time.Since(persistStart).Seconds())
 				if persistErr != nil {
 					return fmt.Errorf("persisting history for %s at ledger %d: %w", protocolID, ledgerSeq, persistErr)
diff --git a/internal/services/protocol_processor.go b/internal/services/protocol_processor.go
index 93ac543a6..c98bd2192 100644
--- a/internal/services/protocol_processor.go
+++ b/internal/services/protocol_processor.go
@@ -82,9 +82,13 @@ type ProtocolProcessor interface {
 	Reset()
 
 	// PersistHistory writes the history rows accumulated by ProcessLedger since the
-	// last Reset, using the provided transaction. Called inside the CAS-guarded
-	// transaction only when the cursor advances, so writes commit atomically with
-	// the cursor update and any failure rolls back both.
+	// last Reset, using the provided transaction. Called only when the CAS cursor
+	// advances. The migration engine passes the CAS-guarded transaction itself, so
+	// rows commit atomically with the cursor; live ingestion passes its
+	// state_changes sibling transaction, which commits strictly before the
+	// cursor-carrying transaction — a committed cursor still implies committed
+	// rows, and rows stranded above the cursor by a crash are removed by startup
+	// reconciliation.
 	PersistHistory(ctx context.Context, dbTx pgx.Tx) error
 	// PersistCurrentState writes the protocol's current state accumulated by
 	// ProcessLedger since the last Reset, using the provided transaction. Called
diff --git a/internal/services/sep41/processor.go b/internal/services/sep41/processor.go
index 5b673c9d7..4ffe96272 100644
--- a/internal/services/sep41/processor.go
+++ b/internal/services/sep41/processor.go
@@ -318,7 +318,7 @@ func (p *processor) indexContracts(contracts []data.ProtocolContracts) {
 	}
 }
 
-// PersistHistory writes staged state changes inside the CAS transaction.
+// PersistHistory writes staged state changes on the provided transaction.
 func (p *processor) PersistHistory(ctx context.Context, dbTx pgx.Tx) error {
 	p.needsReset = true
 	if len(p.stagedStateChanges) == 0 {