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, }, { 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) + } } 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/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/operations.go b/internal/data/operations.go index e01b7fbf4..16ffcd110 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,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 } diff --git a/internal/data/operations_test.go b/internal/data/operations_test.go index 282133f2e..9ed3cf11a 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/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/data/transactions.go b/internal/data/transactions.go index a8ce68374..362a862c4 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,79 @@ 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. 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{}, +) error { + if len(stellarAddressesByToID) == 0 { + return nil + } + + start := time.Now() + + // Build ToID -> LedgerCreatedAt lookup from transactions + ledgerCreatedAtByToID := make(map[int64]time.Time, len(txs)) + for _, tx := range txs { + ledgerCreatedAtByToID[tx.ToID] = tx.LedgerCreatedAt + } - m.Metrics.QueriesTotal.WithLabelValues("BatchCopy", "transactions_accounts").Inc() + // 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, 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 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 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 fmt.Errorf("pgx CopyFrom transactions_accounts: %w", err) + } - return len(txs), nil + return nil } diff --git a/internal/data/transactions_test.go b/internal/data/transactions_test.go index e70b745f3..1c205c7ce 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/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()) diff --git a/internal/db/db.go b/internal/db/db.go index 99233c0a9..a013190f8 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -17,7 +17,12 @@ import ( const ( DefaultMaxConnIdleTime time.Duration = 10 * time.Second - DefaultMaxConns int32 = 10 + // 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/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) { 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.go b/internal/services/ingest.go index b7f907a5c..aa80ffb10 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,52 @@ 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 + } + if err := m.models.Transactions.BatchCopyAccounts(ctx, pgxTx, txs, stellarAddressesByToID); 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 + } + if err := m.models.Operations.BatchCopyAccounts(ctx, pgxTx, ops, stellarAddressesByOpID); 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..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 { @@ -101,21 +120,25 @@ 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, -// 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. +// set. The five bulk COPY families — transactions, transactions_accounts, +// 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 (contracts, classification, 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 +// 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 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. // // 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 @@ -137,19 +160,57 @@ 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 }{ {"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 { + stateChangesMu.Lock() + defer stateChangesMu.Unlock() return m.insertStateChanges(ctx, dbTx, it.buffer.GetStateChanges()) }}, + // 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(), + it.buffer.GetLiquidityPoolShareChanges(), + 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 { @@ -173,17 +234,28 @@ 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. - // 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. + // Stream the sibling writes and stage the coordinated writes concurrently. + // 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 { @@ -195,18 +267,19 @@ 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) } } return nil }) if err = g.Wait(); err != nil { - // Nothing has committed: the deferred rollbacks discard all four - // 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) } @@ -233,28 +306,27 @@ func (m *ingestService) persistLedgerData(ctx context.Context, items []persistIt return nil } -// stageCoordinatedWrites runs every per-ledger write except the three 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: SAC contract tokens, protocol +// classification and wasm/contract rows, CAS-gated protocol state, the SAC +// 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, 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) @@ -266,7 +338,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 @@ -311,7 +383,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 @@ -326,10 +398,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) } @@ -417,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) @@ -440,18 +516,17 @@ func (m *ingestService) stageCoordinatedWrites( } } - // 5. Process token changes (trustline add/remove/update, native balance, SAC balance) - if txErr = m.tokenIngestionService.ProcessTokenChanges(ctx, dbTx, - buffer.GetTrustlineChanges(), - buffer.GetAccountChanges(), + // 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(), - buffer.GetLiquidityPoolShareChanges(), - buffer.GetLiquidityPoolChanges(), ); 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. @@ -536,11 +611,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) @@ -682,12 +757,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_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") +} diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go index 791d4b31c..c4e3fb184 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,101 @@ 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("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", + mock.Anything, // ctx + mock.Anything, // dbTx + mock.Anything, // accountChangesByAccountID + 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 { @@ -1577,12 +1673,20 @@ func Test_persistLedgerDataWithRetry(t *testing.T) { // Mock AccountTokenService to succeed mockTokenIngestionService := NewTokenIngestionServiceMock(t) - mockTokenIngestionService.On("ProcessTokenChanges", + mockTokenIngestionService.On("ProcessTrustlineChanges", mock.Anything, // ctx mock.Anything, // dbTx mock.Anything, // trustlineChangesByTrustlineKey - mock.Anything, // accountChangesByAccountID + ).Return(nil) + mockTokenIngestionService.On("ProcessSACBalanceChanges", + mock.Anything, // ctx + mock.Anything, // dbTx 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) @@ -1655,15 +1759,23 @@ func Test_persistLedgerDataWithRetry(t *testing.T) { // Mock AccountTokenService to return error (simulating DB failure) mockTokenIngestionService := NewTokenIngestionServiceMock(t) - mockTokenIngestionService.On("ProcessTokenChanges", + mockTokenIngestionService.On("ProcessSACBalanceChanges", + mock.Anything, // ctx + mock.Anything, // dbTx + 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 mock.Anything, // accountChangesByAccountID - mock.Anything, // sacBalanceChangesByKey mock.Anything, // lpShareChangesByKey mock.Anything, // lpChangesByPoolID - ).Return(fmt.Errorf("db connection failed")) + ).Return(nil).Maybe() svc, err := NewIngestService(IngestServiceConfig{ IngestionMode: IngestionModeLive, @@ -1733,24 +1845,28 @@ func Test_persistLedgerDataWithRetry(t *testing.T) { // Mock AccountTokenService to fail once then succeed mockTokenIngestionService := NewTokenIngestionServiceMock(t) - mockTokenIngestionService.On("ProcessTokenChanges", + mockTokenIngestionService.On("ProcessSACBalanceChanges", 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("ProcessSACBalanceChanges", + mock.Anything, // ctx + mock.Anything, // dbTx + 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 mock.Anything, // accountChangesByAccountID - mock.Anything, // sacBalanceChangesByKey mock.Anything, // lpShareChangesByKey mock.Anything, // lpChangesByPoolID - ).Return(nil).Once() + ).Return(nil).Maybe() svc, err := NewIngestService(IngestServiceConfig{ IngestionMode: IngestionModeLive, @@ -1939,12 +2055,20 @@ func Test_persistLedgerData_ProtocolCASGating(t *testing.T) { require.NoError(t, err) mockTokenIngestionService := NewTokenIngestionServiceMock(t) - mockTokenIngestionService.On("ProcessTokenChanges", + mockTokenIngestionService.On("ProcessTrustlineChanges", mock.Anything, // ctx mock.Anything, // dbTx mock.Anything, // trustlineChangesByTrustlineKey - mock.Anything, // accountChangesByAccountID + ).Return(nil).Maybe() + mockTokenIngestionService.On("ProcessSACBalanceChanges", + mock.Anything, // ctx + mock.Anything, // dbTx 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() @@ -2418,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 @@ -2668,10 +2792,15 @@ func Test_persistLedgerData_ClassificationPlan(t *testing.T) { require.NoError(t, err) mockTokenIngestionService := NewTokenIngestionServiceMock(t) - mockTokenIngestionService.On("ProcessTokenChanges", - 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, + ).Return(nil).Maybe() svc, err := NewIngestService(IngestServiceConfig{ IngestionMode: IngestionModeLive, @@ -2838,12 +2967,20 @@ func Test_ingestService_ingestLiveLedgers_LagReadDoesNotBlockConsumer(t *testing require.NoError(t, err) mockTokenIngestionService := NewTokenIngestionServiceMock(t) - mockTokenIngestionService.On("ProcessTokenChanges", + mockTokenIngestionService.On("ProcessTrustlineChanges", mock.Anything, // ctx mock.Anything, // dbTx mock.Anything, // trustlineChangesByTrustlineKey - mock.Anything, // accountChangesByAccountID + ).Return(nil).Maybe() + mockTokenIngestionService.On("ProcessSACBalanceChanges", + mock.Anything, // ctx + mock.Anything, // dbTx 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() @@ -2973,8 +3110,14 @@ 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("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, ).Return(nil).Maybe() cursorReached := func(target uint32) bool { @@ -3034,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() }) @@ -3048,8 +3191,14 @@ 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("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, ).Return(nil).Maybe() svc, err := NewIngestService(IngestServiceConfig{ @@ -3062,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 @@ -3129,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) { @@ -3148,8 +3341,14 @@ 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("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, ).Return(nil).Maybe() svc, err := NewIngestService(IngestServiceConfig{ diff --git a/internal/services/mocks.go b/internal/services/mocks.go index 6bea76a3e..3612e1a4d 100644 --- a/internal/services/mocks.go +++ b/internal/services/mocks.go @@ -126,8 +126,18 @@ 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) 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) +} + +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/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 { diff --git a/internal/services/token_ingestion.go b/internal/services/token_ingestion.go index 548881bcd..3c929d541 100644 --- a/internal/services/token_ingestion.go +++ b/internal/services/token_ingestion.go @@ -15,10 +15,26 @@ import ( ) // TokenIngestionService provides write access to account token storage during live ingestion. +// 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 { - // 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 + // 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 } // Verify interface compliance at compile time @@ -56,22 +72,24 @@ 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 - } +// 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) +} - if err := s.processTrustlineChanges(ctx, dbTx, trustlineChangesByTrustlineKey); err != nil { - return err - } +// 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 +// 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.processSACBalanceChanges(ctx, dbTx, sacBalanceChangesByKey); err != nil { - return err - } if err := s.processLiquidityPoolChanges(ctx, dbTx, lpChangesByPoolID); err != nil { return err } diff --git a/internal/services/token_ingestion_test.go b/internal/services/token_ingestion_test.go index 35a886958..429c4cca0 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,13 @@ 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.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) })