Give each write family its own persist sibling

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,
 		},
 		{

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 {

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)
+	}
 }

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)
 	}

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())

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,79 @@ 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. 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{},
+) error {
+	if len(stellarAddressesByOpID) == 0 {
+		return nil
+	}
+
+	start := time.Now()
+
+	// Build OpID -> LedgerCreatedAt lookup from operations
+	ledgerCreatedAtByOpID := make(map[int64]time.Time, len(operations))
+	for _, op := range operations {
+		ledgerCreatedAtByOpID[op.ID] = op.LedgerCreatedAt
+	}
 
-		m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "operations_accounts").Inc()
+	// 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, 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 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 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 fmt.Errorf("pgx CopyFrom operations_accounts: %w", err)
+	}
 
-	return len(operations), nil
+	return nil
 }

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 {

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()