diff --git a/internal/indexer/indexer.go b/internal/indexer/indexer.go index d25767955..54da8d1f7 100644 --- a/internal/indexer/indexer.go +++ b/internal/indexer/indexer.go @@ -22,9 +22,8 @@ import ( ) type IndexerBufferInterface interface { - PushTransaction(participant string, transaction types.Transaction) - PushOperation(participant string, operation types.Operation, transaction types.Transaction) - PushStateChange(transaction types.Transaction, operation types.Operation, stateChange types.StateChange) + // IngestTransactionResult folds a worker's per-transaction result into the buffer. + IngestTransactionResult(result *TransactionResult) GetTransactionsParticipants() map[int64]set.Set[string] GetOperationsParticipants() map[int64]set.Set[string] GetNumberOfTransactions() int @@ -35,24 +34,15 @@ type IndexerBufferInterface interface { GetTrustlineChanges() map[TrustlineChangeKey]types.TrustlineChange GetAccountChanges() map[string]types.AccountChange GetSACBalanceChanges() map[SACBalanceChangeKey]types.SACBalanceChange - PushTrustlineChange(trustlineChange types.TrustlineChange) - PushAccountChange(accountChange types.AccountChange) - PushSACBalanceChange(sacBalanceChange types.SACBalanceChange) - PushSACContract(c *data.Contract) - PushProtocolWasm(wasm data.ProtocolWasms) - PushProtocolWasmBytecode(wasmHash string, bytecode []byte) - PushProtocolContracts(contract data.ProtocolContracts) - PushContractEvents(key ContractEventKey, events []xdr.ContractEvent) GetUniqueTrustlineAssets() []data.TrustlineAsset GetSACContracts() map[string]*data.Contract GetProtocolWasms() map[string]data.ProtocolWasms - // GetProtocolWasmBytecodes returns a shallow clone of the wasmHash → bytecode - // map. The []byte values alias buffer-owned storage and MUST be treated as - // read-only; bytecode is content-addressed and immutable. + // GetProtocolWasmBytecodes returns the wasmHash → bytecode map. The []byte values alias + // buffer-owned storage and MUST be treated as read-only; bytecode is content-addressed + // and immutable. GetProtocolWasmBytecodes() map[string][]byte GetProtocolContracts() map[string]data.ProtocolContracts GetContractEvents() map[ContractEventKey][]xdr.ContractEvent - Merge(other IndexerBufferInterface) Clear() } @@ -126,13 +116,14 @@ func NewIndexer(networkPassphrase string, pool pond.Pool, ingestionMetrics *metr } // ProcessLedgerTransactions processes all transactions in a ledger in parallel. -// It collects transaction data (participants, operations, state changes) and populates the buffer in a single pass. +// Each worker builds an independent TransactionResult (no shared buffer, no locks); the results +// are then folded into the single ledger buffer serially. This avoids allocating a full +// IndexerBuffer per transaction and the subsequent buffer-to-buffer merge. // Returns the total participant count for metrics. func (i *Indexer) ProcessLedgerTransactions(ctx context.Context, transactions []ingest.LedgerTransaction, ledgerBuffer IndexerBufferInterface) (int, error) { group := i.pool.NewGroupContext(ctx) - txnBuffers := make([]*IndexerBuffer, len(transactions)) - participantCounts := make([]int, len(transactions)) + results := make([]*TransactionResult, len(transactions)) var errs []error errMu := sync.Mutex{} @@ -140,16 +131,14 @@ func (i *Indexer) ProcessLedgerTransactions(ctx context.Context, transactions [] index := idx tx := tx group.Submit(func() { - buffer := NewIndexerBuffer() - count, err := i.processTransaction(ctx, tx, buffer) + result, err := i.processTransaction(ctx, tx) if err != nil { errMu.Lock() errs = append(errs, fmt.Errorf("processing transaction at ledger=%d tx=%d: %w", tx.Ledger.LedgerSequence(), tx.Index, err)) errMu.Unlock() return } - txnBuffers[index] = buffer - participantCounts[index] = count + results[index] = result }) } @@ -160,44 +149,50 @@ func (i *Indexer) ProcessLedgerTransactions(ctx context.Context, transactions [] return 0, fmt.Errorf("processing transactions: %w", errors.Join(errs...)) } - // Merge buffers and count participants + // Fold per-transaction results into the ledger buffer serially (single-owner, no locks). totalParticipants := 0 - for idx, buffer := range txnBuffers { - ledgerBuffer.Merge(buffer) - totalParticipants += participantCounts[idx] + for _, result := range results { + if result == nil { + continue + } + ledgerBuffer.IngestTransactionResult(result) + totalParticipants += result.ParticipantCount } return totalParticipants, nil } -// processTransaction processes a single transaction - collects data and populates buffer. -// Returns participant count for metrics. -func (i *Indexer) processTransaction(ctx context.Context, tx ingest.LedgerTransaction, buffer *IndexerBuffer) (int, error) { +// processTransaction processes a single transaction and returns its result bundle for the caller +// to fold into the ledger buffer. It performs no buffer writes itself, so workers can run fully in +// parallel with no shared state. +func (i *Indexer) processTransaction(ctx context.Context, tx ingest.LedgerTransaction) (*TransactionResult, error) { // Get transaction participants txParticipants, err := i.participantsProcessor.GetTransactionParticipants(tx) if err != nil { - return 0, fmt.Errorf("getting transaction participants: %w", err) + return nil, fmt.Errorf("getting transaction participants: %w", err) } // Get operations participants opsParticipants, err := i.participantsProcessor.GetOperationsParticipants(tx) if err != nil { - return 0, fmt.Errorf("getting operations participants: %w", err) + return nil, fmt.Errorf("getting operations participants: %w", err) } // Get state changes stateChanges, err := i.getTransactionStateChanges(ctx, tx, opsParticipants) if err != nil { - return 0, fmt.Errorf("getting transaction state changes: %w", err) + return nil, fmt.Errorf("getting transaction state changes: %w", err) } // Convert transaction data dataTx, err := processors.ConvertTransaction(&tx) if err != nil { - return 0, fmt.Errorf("creating data transaction: %w", err) + return nil, fmt.Errorf("creating data transaction: %w", err) } - // Count all unique participants for metrics + // Count all unique participants for metrics. This transient set unions the participants + // processor's (thread-safe) sets, so it must be thread-safe too — deckarep's Union + // type-asserts its argument to the receiver's variant. allParticipants := set.NewSet[string]() allParticipants = allParticipants.Union(txParticipants) for _, opParticipants := range opsParticipants { @@ -207,79 +202,71 @@ func (i *Indexer) processTransaction(ctx context.Context, tx ingest.LedgerTransa allParticipants.Add(string(stateChange.AccountID)) } - // Insert transaction participants - for participant := range txParticipants.Iter() { - buffer.PushTransaction(participant, *dataTx) + result := &TransactionResult{ + Transaction: dataTx, + TxParticipants: txParticipants.ToSlice(), + Operations: make(map[int64]*types.Operation, len(opsParticipants)), + OpParticipants: make(map[int64][]string, len(opsParticipants)), + ProtocolWasmBytecodes: make(map[string][]byte), + ContractEvents: make(map[ContractEventKey][]xdr.ContractEvent), + ParticipantCount: allParticipants.Cardinality(), } // Get operation results for extracting result codes opResults, _ := tx.Result.OperationResults() - // Insert operations participants - operationsMap := make(map[int64]*types.Operation) + // Build operations and their participants for opID, opParticipants := range opsParticipants { dataOp, opErr := processors.ConvertOperation(&tx, &opParticipants.OpWrapper.Operation, opID, opParticipants.OpWrapper.Index, opResults) if opErr != nil { - return 0, fmt.Errorf("creating data operation: %w", opErr) - } - operationsMap[opID] = dataOp - for participant := range opParticipants.Participants.Iter() { - buffer.PushOperation(participant, *dataOp, *dataTx) + return nil, fmt.Errorf("creating data operation: %w", opErr) } + result.Operations[opID] = dataOp + result.OpParticipants[opID] = opParticipants.Participants.ToSlice() } // Process trustline, account, and SAC balance changes from ledger changes for _, opParticipants := range opsParticipants { trustlineChanges, tlErr := i.trustlinesProcessor.ProcessOperation(ctx, opParticipants.OpWrapper) if tlErr != nil { - return 0, fmt.Errorf("processing trustline changes: %w", tlErr) - } - for _, tlChange := range trustlineChanges { - buffer.PushTrustlineChange(tlChange) + return nil, fmt.Errorf("processing trustline changes: %w", tlErr) } + result.TrustlineChanges = append(result.TrustlineChanges, trustlineChanges...) accountChanges, accErr := i.accountsProcessor.ProcessOperation(ctx, opParticipants.OpWrapper) if accErr != nil { - return 0, fmt.Errorf("processing account changes: %w", accErr) - } - for _, accChange := range accountChanges { - buffer.PushAccountChange(accChange) + return nil, fmt.Errorf("processing account changes: %w", accErr) } + result.AccountChanges = append(result.AccountChanges, accountChanges...) sacBalanceChanges, sacErr := i.sacBalancesProcessor.ProcessOperation(ctx, opParticipants.OpWrapper) if sacErr != nil { - return 0, fmt.Errorf("processing SAC balance changes: %w", sacErr) - } - for _, sacChange := range sacBalanceChanges { - buffer.PushSACBalanceChange(sacChange) + return nil, fmt.Errorf("processing SAC balance changes: %w", sacErr) } + result.SACBalanceChanges = append(result.SACBalanceChanges, sacBalanceChanges...) sacContracts, sacInstanceErr := i.sacInstancesProcessor.ProcessOperation(ctx, opParticipants.OpWrapper) if sacInstanceErr != nil { - return 0, fmt.Errorf("processing SAC instances: %w", sacInstanceErr) - } - for _, c := range sacContracts { - buffer.PushSACContract(c) + return nil, fmt.Errorf("processing SAC instances: %w", sacInstanceErr) } + result.SACContracts = append(result.SACContracts, sacContracts...) protocolWasms, pwErr := i.protocolWasmsProcessor.ProcessOperation(ctx, opParticipants.OpWrapper) if pwErr != nil { - return 0, fmt.Errorf("processing protocol wasms: %w", pwErr) + return nil, fmt.Errorf("processing protocol wasms: %w", pwErr) } for _, wasm := range protocolWasms { - buffer.PushProtocolWasm(wasm.Record) + result.ProtocolWasms = append(result.ProtocolWasms, wasm.Record) if len(wasm.Bytecode) > 0 { - buffer.PushProtocolWasmBytecode(string(wasm.Record.WasmHash), wasm.Bytecode) + result.ProtocolWasmBytecodes[string(wasm.Record.WasmHash)] = wasm.Bytecode } } protocolContracts, pcErr := i.protocolContractsProcessor.ProcessOperation(ctx, opParticipants.OpWrapper) if pcErr != nil { - return 0, fmt.Errorf("processing protocol contracts: %w", pcErr) - } - for _, contract := range protocolContracts { - buffer.PushProtocolContracts(contract) + return nil, fmt.Errorf("processing protocol contracts: %w", pcErr) } + result.ProtocolContracts = append(result.ProtocolContracts, protocolContracts...) } // Fold transaction fee-phase changes (fee debits + Soroban fee refunds) into @@ -288,19 +275,16 @@ func (i *Indexer) processTransaction(ctx context.Context, tx ingest.LedgerTransa // those phases — e.g. a fee-bump fee source (issue #637). feeAccountChanges, feeErr := i.accountsProcessor.ProcessTransactionFees(ctx, tx) if feeErr != nil { - return 0, fmt.Errorf("processing transaction fee account changes: %w", feeErr) - } - for _, accChange := range feeAccountChanges { - buffer.PushAccountChange(accChange) + return nil, fmt.Errorf("processing transaction fee account changes: %w", feeErr) } - - // Stash contract events into the buffer so protocol processors can consume - // them without re-decoding LedgerCloseMeta. Only successful transactions are - // indexed — protocol processors (e.g. SEP-41) only care about successful - // invocations, and emitting events from failed txs would force every consumer - // to re-filter. SACEventsProcessor still calls GetContractEventsForOperation - // itself (see TODO at processors/contracts/sac.go); that consolidation is a - // separate cleanup. + result.AccountChanges = append(result.AccountChanges, feeAccountChanges...) + + // Stash contract events so protocol processors can consume them without re-decoding + // LedgerCloseMeta. Only successful transactions are indexed — protocol processors + // (e.g. SEP-41) only care about successful invocations, and emitting events from failed + // txs would force every consumer to re-filter. SACEventsProcessor still calls + // GetContractEventsForOperation itself (see TODO at processors/contracts/sac.go); that + // consolidation is a separate cleanup. if tx.Result.Successful() { for _, opParticipants := range opsParticipants { opWrapper := opParticipants.OpWrapper @@ -309,37 +293,32 @@ func (i *Indexer) processTransaction(ctx context.Context, tx ingest.LedgerTransa } events, evErr := tx.GetContractEventsForOperation(opWrapper.Index) if evErr != nil { - return 0, fmt.Errorf("extracting contract events for op %d: %w", opWrapper.Index, evErr) + return nil, fmt.Errorf("extracting contract events for op %d: %w", opWrapper.Index, evErr) } if len(events) == 0 { continue } - buffer.PushContractEvents(ContractEventKey{TxIdx: tx.Index, OpIdx: opWrapper.Index}, events) + result.ContractEvents[ContractEventKey{TxIdx: tx.Index, OpIdx: opWrapper.Index}] = events } } - // Insert state changes + // Collect state changes, dropping empties and those whose operation is missing. The fold + // resolves each change's operation from result.Operations by OperationID. for _, stateChange := range stateChanges { // Skip empty state changes (no account to associate with) if stateChange.AccountID == "" { continue } - - // Get the correct operation for this state change - var operation types.Operation - if stateChange.OperationID != 0 { - correctOp := operationsMap[stateChange.OperationID] - if correctOp == nil { - log.Ctx(ctx).Errorf("operation ID %d not found in operations map for state change (to_id=%d, category=%s)", stateChange.OperationID, stateChange.ToID, stateChange.StateChangeCategory) - continue - } - operation = *correctOp + // Fee state changes (OperationID == 0) have no associated operation; the rest must + // resolve to a known operation. + if stateChange.OperationID != 0 && result.Operations[stateChange.OperationID] == nil { + log.Ctx(ctx).Errorf("operation ID %d not found in operations map for state change (to_id=%d, category=%s)", stateChange.OperationID, stateChange.ToID, stateChange.StateChangeCategory) + continue } - // For fee state changes (OperationID == 0), operation remains zero value - buffer.PushStateChange(*dataTx, operation, stateChange) + result.StateChanges = append(result.StateChanges, stateChange) } - return allParticipants.Cardinality(), nil + return result, nil } // getTransactionStateChanges processes operations of a transaction and calculates all state changes diff --git a/internal/indexer/indexer_buffer.go b/internal/indexer/indexer_buffer.go index 2a94479eb..fb084e82a 100644 --- a/internal/indexer/indexer_buffer.go +++ b/internal/indexer/indexer_buffer.go @@ -5,9 +5,7 @@ package indexer import ( "fmt" - "maps" "strings" - "sync" set "github.com/deckarep/golang-set/v2" "github.com/google/uuid" @@ -18,8 +16,8 @@ import ( "github.com/stellar/wallet-backend/internal/indexer/types" ) -// IndexerBuffer is a thread-safe, memory-efficient buffer for collecting blockchain data -// during ledger ingestion. It uses a two-level storage architecture: +// IndexerBuffer is a memory-efficient buffer for collecting blockchain data during ledger +// ingestion. It uses a two-level storage architecture: // // ARCHITECTURE: // 1. Canonical Storage Layer: @@ -28,23 +26,19 @@ import ( // - This layer owns the actual data and ensures only ONE copy exists in memory // // 2. Transaction/Operation to Participants Mapping Layer: -// - participantsByTxHash: Maps each transaction hash to a SET of participant IDs +// - participantsByToID: Maps each transaction ToID to a SET of participant IDs // - participantsByOpID: Maps each operation ID to a SET of participant IDs // - Efficiently tracks which participants interacted with each tx/op // // MEMORY OPTIMIZATION: -// Transaction structs contain large XDR fields (10-50+ KB each). When multiple participants -// interact with the same transaction, they all point to the SAME canonical pointer instead -// of storing duplicate copies. +// When multiple participants interact with the same transaction or operation, they all point +// to the SAME canonical pointer instead of storing duplicate copies. // -// PERFORMANCE: -// - Push operations: O(1) via set.Add() with automatic deduplication -// - No manual duplicate checking: Sets handle uniqueness automatically -// - MergeBuffer: O(n) with zero temporary map allocations -// -// THREAD SAFETY: -// All public methods use RWMutex for concurrent read/exclusive write access. -// Callers can safely use multiple buffers in parallel goroutines. +// OWNERSHIP: +// Not safe for concurrent use. Each buffer instance is owned by a single goroutine: +// ProcessLedgerTransactions builds a per-transaction TransactionResult in parallel workers and +// folds each into one ledger buffer serially (IngestTransactionResult). The participant sets are +// thread-unsafe for the same reason. type TrustlineChangeKey struct { AccountID string @@ -68,7 +62,6 @@ type ContractEventKey struct { } type IndexerBuffer struct { - mu sync.RWMutex txByHash map[string]*types.Transaction participantsByToID map[int64]set.Set[string] opByID map[int64]*types.Operation @@ -92,7 +85,7 @@ type IndexerBuffer struct { } // NewIndexerBuffer creates a new IndexerBuffer with initialized data structures. -// All maps and sets are pre-allocated to avoid nil pointer issues during concurrent access. +// All maps are pre-allocated to avoid nil map access. func NewIndexerBuffer() *IndexerBuffer { return &IndexerBuffer{ txByHash: make(map[string]*types.Transaction), @@ -118,24 +111,17 @@ 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 // which participants interacted with it. Multiple participants can reference the same transaction. -// Thread-safe: acquires write lock. -func (b *IndexerBuffer) PushTransaction(participant string, transaction types.Transaction) { - b.mu.Lock() - defer b.mu.Unlock() - - b.pushTransactionUnsafe(participant, &transaction) +func (b *IndexerBuffer) PushTransaction(participant string, transaction *types.Transaction) { + b.recordTransaction(participant, transaction) } -// pushTransactionUnsafe is the internal implementation that assumes the caller -// already holds the write lock. This method implements the following pattern: +// recordTransaction is the shared internal helper that stores a transaction pointer and +// records the participant: // // 1. Check if transaction already exists in txByHash // 2. If not, store the transaction pointer -// 3. Add participant to the global participants set -// 4. Add participant to this transaction's participant set in participantsByToID -// -// Caller must hold write lock. -func (b *IndexerBuffer) pushTransactionUnsafe(participant string, transaction *types.Transaction) { +// 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 @@ -144,7 +130,7 @@ func (b *IndexerBuffer) pushTransactionUnsafe(participant string, transaction *t // Track this participant by ToID toID := transaction.ToID if _, exists := b.participantsByToID[toID]; !exists { - b.participantsByToID[toID] = set.NewSet[string]() + b.participantsByToID[toID] = set.NewThreadUnsafeSet[string]() } // Add participant - O(1) with automatic deduplication @@ -152,29 +138,17 @@ func (b *IndexerBuffer) pushTransactionUnsafe(participant string, transaction *t } // GetNumberOfTransactions returns the count of unique transactions in the buffer. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetNumberOfTransactions() int { - b.mu.RLock() - defer b.mu.RUnlock() - return len(b.txByHash) } // GetNumberOfOperations returns the count of unique operations in the buffer. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetNumberOfOperations() int { - b.mu.RLock() - defer b.mu.RUnlock() - return len(b.opByID) } // GetTransactions returns all unique transactions. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetTransactions() []*types.Transaction { - b.mu.RLock() - defer b.mu.RUnlock() - txs := make([]*types.Transaction, 0, len(b.txByHash)) for _, txPtr := range b.txByHash { txs = append(txs, txPtr) @@ -183,12 +157,9 @@ func (b *IndexerBuffer) GetTransactions() []*types.Transaction { return txs } -// GetTransactionsParticipants returns a map of transaction ToIDs to its participants. +// GetTransactionsParticipants returns the map of transaction ToIDs to their participants. func (b *IndexerBuffer) GetTransactionsParticipants() map[int64]set.Set[string] { - b.mu.RLock() - defer b.mu.RUnlock() - - return maps.Clone(b.participantsByToID) + return b.participantsByToID } // pushWithTombstone deduplicates change into m, keeping the highest-ordered change per key. @@ -229,31 +200,6 @@ func pushWithTombstone[K comparable, V any]( m[key] = change } -// mergeWithTombstone folds src into dst using the same tombstone-aware dedup as pushWithTombstone. -// It first merges src's tombstones into dst (keeping the higher order per key and evicting any dst -// entry the tombstone cancels), then replays src's surviving changes. Merging tombstones is -// required because a create->remove cancelled entirely within src leaves only a tombstone there, -// which must carry over so a lower-order change in dst cannot resurrect the key. -func mergeWithTombstone[K comparable, V any]( - dst, src map[K]V, - dstTombstones, srcTombstones map[K]int64, - order func(V) int64, - isNoopRemove func(existing, incoming V) bool, -) { - for key, tomb := range srcTombstones { - if existing, ok := dstTombstones[key]; !ok || tomb > existing { - dstTombstones[key] = tomb - } - if entry, ok := dst[key]; ok && order(entry) <= dstTombstones[key] { - delete(dst, key) - } - } - - for key, change := range src { - pushWithTombstone(dst, dstTombstones, key, change, order, isNoopRemove) - } -} - func accountOrder(c types.AccountChange) int64 { return c.SortKey } func accountIsNoopRemove(existing, incoming types.AccountChange) bool { @@ -273,11 +219,7 @@ func sacBalanceIsNoopRemove(existing, incoming types.SACBalanceChange) bool { } // PushTrustlineChange adds a trustline change to the buffer and tracks unique assets. -// Thread-safe: acquires write lock. func (b *IndexerBuffer) PushTrustlineChange(trustlineChange types.TrustlineChange) { - b.mu.Lock() - defer b.mu.Unlock() - code, issuer, err := ParseAssetString(trustlineChange.Asset) if err != nil { return // Skip invalid assets @@ -301,42 +243,28 @@ func (b *IndexerBuffer) PushTrustlineChange(trustlineChange types.TrustlineChang } // GetTrustlineChanges returns all trustline changes stored in the buffer. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetTrustlineChanges() map[TrustlineChangeKey]types.TrustlineChange { - b.mu.RLock() - defer b.mu.RUnlock() - return b.trustlineChangesByTrustlineKey } // PushAccountChange adds an account change to the buffer with deduplication. // Keeps the change with highest SortKey per account. A CREATE→REMOVE within the same ledger nets // to nothing and is tombstoned so a later lower-key change cannot resurrect it (see -// pushWithTombstone). Thread-safe: acquires write lock. +// pushWithTombstone). func (b *IndexerBuffer) PushAccountChange(accountChange types.AccountChange) { - b.mu.Lock() - defer b.mu.Unlock() - pushWithTombstone(b.accountChangesByAccountID, b.accountTombstones, accountChange.AccountID, accountChange, accountOrder, accountIsNoopRemove) } // GetAccountChanges returns all account changes stored in the buffer. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetAccountChanges() map[string]types.AccountChange { - b.mu.RLock() - defer b.mu.RUnlock() - return b.accountChangesByAccountID } // PushSACBalanceChange adds a SAC balance change to the buffer with deduplication. // Keeps the change with highest OperationID per (AccountID, ContractID). An ADD→REMOVE within the // same ledger nets to nothing and is tombstoned so a later lower-key change cannot resurrect it -// (see pushWithTombstone). Thread-safe: acquires write lock. +// (see pushWithTombstone). func (b *IndexerBuffer) PushSACBalanceChange(sacBalanceChange types.SACBalanceChange) { - b.mu.Lock() - defer b.mu.Unlock() - key := SACBalanceChangeKey{ AccountID: sacBalanceChange.AccountID, ContractID: sacBalanceChange.ContractID, @@ -345,31 +273,19 @@ func (b *IndexerBuffer) PushSACBalanceChange(sacBalanceChange types.SACBalanceCh } // GetSACBalanceChanges returns all SAC balance changes stored in the buffer. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetSACBalanceChanges() map[SACBalanceChangeKey]types.SACBalanceChange { - b.mu.RLock() - defer b.mu.RUnlock() - return b.sacBalanceChangesByKey } // PushOperation adds an operation and its parent transaction, associating both with a participant. // Uses canonical pointer pattern for both operations and transactions to avoid memory duplication. -// Thread-safe: acquires write lock. -func (b *IndexerBuffer) PushOperation(participant string, operation types.Operation, transaction types.Transaction) { - b.mu.Lock() - defer b.mu.Unlock() - - b.pushOperationUnsafe(participant, &operation) - b.pushTransactionUnsafe(participant, &transaction) +func (b *IndexerBuffer) PushOperation(participant string, operation *types.Operation, transaction *types.Transaction) { + b.recordOperation(participant, operation) + b.recordTransaction(participant, transaction) } // GetOperations returns all unique operations from the canonical storage. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetOperations() []*types.Operation { - b.mu.RLock() - defer b.mu.RUnlock() - ops := make([]*types.Operation, 0, len(b.opByID)) for _, opPtr := range b.opByID { ops = append(ops, opPtr) @@ -377,18 +293,15 @@ func (b *IndexerBuffer) GetOperations() []*types.Operation { return ops } -// GetOperationsParticipants returns a map of operation IDs to its participants. +// GetOperationsParticipants returns the map of operation IDs to their participants. func (b *IndexerBuffer) GetOperationsParticipants() map[int64]set.Set[string] { - b.mu.RLock() - defer b.mu.RUnlock() - - return maps.Clone(b.participantsByOpID) + return b.participantsByOpID } -// pushOperationUnsafe is the internal implementation for operation storage. -// Stores one copy of each operation (by ID) and tracks which participants interacted with it. -// Caller must hold write lock. -func (b *IndexerBuffer) pushOperationUnsafe(participant string, operation *types.Operation) { +// recordOperation is the shared internal helper that stores an operation pointer and records +// the participant. Stores one copy of each operation (by ID) and tracks which participants +// interacted with it. +func (b *IndexerBuffer) recordOperation(participant string, operation *types.Operation) { opID := operation.ID if _, exists := b.opByID[opID]; !exists { b.opByID[opID] = operation @@ -396,150 +309,105 @@ func (b *IndexerBuffer) pushOperationUnsafe(participant string, operation *types // Track this participant globally if _, exists := b.participantsByOpID[opID]; !exists { - b.participantsByOpID[opID] = set.NewSet[string]() + b.participantsByOpID[opID] = set.NewThreadUnsafeSet[string]() } b.participantsByOpID[opID].Add(participant) } // PushStateChange adds a state change along with its associated transaction and operation. -// Thread-safe: acquires write lock. -func (b *IndexerBuffer) PushStateChange(transaction types.Transaction, operation types.Operation, stateChange types.StateChange) { - b.mu.Lock() - defer b.mu.Unlock() - +// operation may be nil for fee state changes, which have no associated operation. +func (b *IndexerBuffer) PushStateChange(transaction *types.Transaction, operation *types.Operation, stateChange types.StateChange) { b.stateChanges = append(b.stateChanges, stateChange) - b.pushTransactionUnsafe(string(stateChange.AccountID), &transaction) + b.recordTransaction(string(stateChange.AccountID), transaction) // Fee changes dont have an operation ID associated with them - if stateChange.OperationID != 0 { - b.pushOperationUnsafe(string(stateChange.AccountID), &operation) + if stateChange.OperationID != 0 && operation != nil { + b.recordOperation(string(stateChange.AccountID), operation) } } -// GetStateChanges returns a copy of all state changes stored in the buffer. -// Thread-safe: uses read lock. +// GetStateChanges returns all state changes stored in the buffer. func (b *IndexerBuffer) GetStateChanges() []types.StateChange { - b.mu.RLock() - defer b.mu.RUnlock() - return b.stateChanges } -// Merge merges another IndexerBuffer into this buffer. This is used to combine -// per-ledger or per-transaction buffers into a single buffer for batch DB insertion. -// -// MERGE STRATEGY: -// 1. Union global participant sets (O(m) set operation) -// 2. Copy storage maps (txByHash, opByID) using maps.Copy -// 3. For each transaction hash in other.participantsByTxHash: -// - Merge other's participant set into our participant set for that tx hash -// - Creates new set if tx doesn't exist in our mapping yet -// -// 4. For each operation ID in other.participantsByOpID: -// - Merge other's participant set into our participant set for that op ID -// - Creates new set if op doesn't exist in our mapping yet -// -// 5. Append other's state changes to ours +// TransactionResult is the per-transaction output produced by a parallel worker in +// ProcessLedgerTransactions. Workers build these independently (no shared buffer, no locks); the +// serial fold (IngestTransactionResult) then replays them into one ledger buffer. This avoids +// allocating a full IndexerBuffer per transaction and the subsequent buffer-to-buffer merge. // -// MEMORY EFFICIENCY: -// Zero temporary allocations - uses direct map/set manipulation. -// -// Thread-safe: acquires write lock on this buffer, read lock on other buffer. -func (b *IndexerBuffer) Merge(other IndexerBufferInterface) { - b.mu.Lock() - defer b.mu.Unlock() - - // Type assert to get concrete buffer for efficient merging - otherBuffer, ok := other.(*IndexerBuffer) - if !ok { - return - } - - otherBuffer.mu.RLock() - defer otherBuffer.mu.RUnlock() - - // Merge transactions (canonical storage) - this establishes our canonical pointers - maps.Copy(b.txByHash, otherBuffer.txByHash) - for toID, otherParticipants := range otherBuffer.participantsByToID { - if existing, exists := b.participantsByToID[toID]; exists { - // Merge into existing set - iterate and add (Union creates new set) - for participant := range otherParticipants.Iter() { - existing.Add(participant) - } - } else { - // Clone the set instead of creating empty + iterating - b.participantsByToID[toID] = otherParticipants.Clone() +// Operations is keyed by operation ID and is shared by OpParticipants (participant tracking) and +// StateChanges (state-change → operation association). StateChanges is already filtered by the +// worker: entries with an empty AccountID or an OperationID with no matching operation are dropped. +type TransactionResult struct { + Transaction *types.Transaction + TxParticipants []string + Operations map[int64]*types.Operation + OpParticipants map[int64][]string + StateChanges []types.StateChange + TrustlineChanges []types.TrustlineChange + AccountChanges []types.AccountChange + SACBalanceChanges []types.SACBalanceChange + SACContracts []*data.Contract + ProtocolWasms []data.ProtocolWasms + ProtocolWasmBytecodes map[string][]byte + ProtocolContracts []data.ProtocolContracts + ContractEvents map[ContractEventKey][]xdr.ContractEvent + ParticipantCount int +} + +// IngestTransactionResult folds a single transaction's result into the buffer, applying the same +// per-key deduplication as the individual Push* methods. It is called serially by +// ProcessLedgerTransactions after the parallel workers finish, so no locking is required. +func (b *IndexerBuffer) IngestTransactionResult(r *TransactionResult) { + for _, participant := range r.TxParticipants { + b.PushTransaction(participant, r.Transaction) + } + + for opID, participants := range r.OpParticipants { + operation := r.Operations[opID] + for _, participant := range participants { + b.PushOperation(participant, operation, r.Transaction) } } - // Merge operations (canonical storage) - maps.Copy(b.opByID, otherBuffer.opByID) - for opID, otherParticipants := range otherBuffer.participantsByOpID { - if existing, exists := b.participantsByOpID[opID]; exists { - // Merge into existing set - iterate and add (Union creates new set) - for participant := range otherParticipants.Iter() { - existing.Add(participant) - } - } else { - // Clone the set instead of creating empty + iterating - b.participantsByOpID[opID] = otherParticipants.Clone() - } + for _, trustlineChange := range r.TrustlineChanges { + b.PushTrustlineChange(trustlineChange) } - - // Merge state changes - b.stateChanges = append(b.stateChanges, otherBuffer.stateChanges...) - - // Merge trustline, account, and SAC balance changes with the same tombstone-aware dedup as - // the Push* methods (highest order wins; a same-ledger create/add→remove nets to nothing and - // tombstones the key so a lower-order change cannot resurrect it). - mergeWithTombstone(b.trustlineChangesByTrustlineKey, otherBuffer.trustlineChangesByTrustlineKey, b.trustlineTombstones, otherBuffer.trustlineTombstones, trustlineOrder, trustlineIsNoopRemove) - mergeWithTombstone(b.accountChangesByAccountID, otherBuffer.accountChangesByAccountID, b.accountTombstones, otherBuffer.accountTombstones, accountOrder, accountIsNoopRemove) - mergeWithTombstone(b.sacBalanceChangesByKey, otherBuffer.sacBalanceChangesByKey, b.sacTombstones, otherBuffer.sacTombstones, sacBalanceOrder, sacBalanceIsNoopRemove) - - // Merge unique trustline assets - maps.Copy(b.uniqueTrustlineAssets, otherBuffer.uniqueTrustlineAssets) - - // Merge SAC contracts (first-write wins for deduplication) - for id, contract := range otherBuffer.sacContractsByID { - if _, exists := b.sacContractsByID[id]; !exists { - b.sacContractsByID[id] = contract - } + for _, accountChange := range r.AccountChanges { + b.PushAccountChange(accountChange) } - - // Merge protocol WASMs (first-write wins for deduplication) - for hash, wasm := range otherBuffer.protocolWasmsByHash { - if _, exists := b.protocolWasmsByHash[hash]; !exists { - b.protocolWasmsByHash[hash] = wasm - } + for _, sacBalanceChange := range r.SACBalanceChanges { + b.PushSACBalanceChange(sacBalanceChange) + } + for _, contract := range r.SACContracts { + b.PushSACContract(contract) + } + for _, wasm := range r.ProtocolWasms { + b.PushProtocolWasm(wasm) + } + for wasmHash, bytecode := range r.ProtocolWasmBytecodes { + b.PushProtocolWasmBytecode(wasmHash, bytecode) + } + for _, contract := range r.ProtocolContracts { + b.PushProtocolContracts(contract) } - // Merge wasm bytecodes (first-write wins; bytecode for a given hash is content-addressed and immutable) - for hash, code := range otherBuffer.wasmBytecodesByHash { - if _, exists := b.wasmBytecodesByHash[hash]; !exists { - b.wasmBytecodesByHash[hash] = code + for _, stateChange := range r.StateChanges { + var operation *types.Operation + if stateChange.OperationID != 0 { + operation = r.Operations[stateChange.OperationID] } + b.PushStateChange(r.Transaction, operation, stateChange) } - // Merge protocol contracts (last-write-wins: otherBuffer has later ledger data) - maps.Copy(b.protocolContractsByID, otherBuffer.protocolContractsByID) - - // Merge contract events (first-write wins: each (txIdx, opIdx) key is - // produced by exactly one goroutine in ProcessLedgerTransactions, so - // collisions don't occur in normal operation. First-write keeps merges - // idempotent if a caller ever merges twice.) - for key, events := range otherBuffer.contractEventsByKey { - if _, exists := b.contractEventsByKey[key]; !exists { - b.contractEventsByKey[key] = events - } + for key, events := range r.ContractEvents { + b.PushContractEvents(key, events) } } // Clear resets the buffer to its initial empty state while preserving allocated capacity. // Use this to reuse the buffer after flushing data to the database during backfill. -// Thread-safe: acquires write lock. func (b *IndexerBuffer) Clear() { - b.mu.Lock() - defer b.mu.Unlock() - // Clear maps (keep allocated backing arrays) clear(b.txByHash) clear(b.participantsByToID) @@ -567,11 +435,7 @@ func (b *IndexerBuffer) Clear() { } // GetUniqueTrustlineAssets returns all unique trustline assets with pre-computed IDs. -// Thread-safe: uses read lock. func (b *IndexerBuffer) GetUniqueTrustlineAssets() []data.TrustlineAsset { - b.mu.RLock() - defer b.mu.RUnlock() - assets := make([]data.TrustlineAsset, 0, len(b.uniqueTrustlineAssets)) for _, asset := range b.uniqueTrustlineAssets { assets = append(assets, asset) @@ -580,33 +444,21 @@ func (b *IndexerBuffer) GetUniqueTrustlineAssets() []data.TrustlineAsset { } // PushSACContract adds a SAC contract with extracted metadata to the buffer. -// Thread-safe: acquires write lock. func (b *IndexerBuffer) PushSACContract(c *data.Contract) { - b.mu.Lock() - defer b.mu.Unlock() - if _, exists := b.sacContractsByID[c.ContractID]; !exists { b.sacContractsByID[c.ContractID] = c } } -// GetSACContracts returns a map of SAC contract IDs to their metadata. -// Thread-safe: uses read lock. +// GetSACContracts returns the map of SAC contract IDs to their metadata. func (b *IndexerBuffer) GetSACContracts() map[string]*data.Contract { - b.mu.RLock() - defer b.mu.RUnlock() - - return maps.Clone(b.sacContractsByID) + return b.sacContractsByID } // PushProtocolWasm adds a protocol WASM record to the buffer (deduplicated by // hash; first-write wins). The record's ProtocolID is left for the // classification dispatcher to populate at persistence time. -// Thread-safe: acquires write lock. func (b *IndexerBuffer) PushProtocolWasm(wasm data.ProtocolWasms) { - b.mu.Lock() - defer b.mu.Unlock() - key := string(wasm.WasmHash) if _, exists := b.protocolWasmsByHash[key]; !exists { b.protocolWasmsByHash[key] = wasm @@ -617,54 +469,33 @@ func (b *IndexerBuffer) PushProtocolWasm(wasm data.ProtocolWasms) { // classification dispatcher in PersistLedgerData to extract specs and run // per-protocol validators. Bytecode is content-addressed by hash, so // first-write wins is safe. -// Thread-safe: acquires write lock. func (b *IndexerBuffer) PushProtocolWasmBytecode(wasmHash string, bytecode []byte) { - b.mu.Lock() - defer b.mu.Unlock() - if _, exists := b.wasmBytecodesByHash[wasmHash]; !exists { b.wasmBytecodesByHash[wasmHash] = bytecode } } -// GetProtocolWasms returns a clone of the protocol WASMs map. -// Thread-safe: uses read lock. +// GetProtocolWasms returns the protocol WASMs map. func (b *IndexerBuffer) GetProtocolWasms() map[string]data.ProtocolWasms { - b.mu.RLock() - defer b.mu.RUnlock() - - return maps.Clone(b.protocolWasmsByHash) + return b.protocolWasmsByHash } -// GetProtocolWasmBytecodes returns a clone of the wasmHash → bytecode map. -// The map is a shallow copy: the returned []byte values alias the buffer's -// internal storage and MUST be treated as read-only by callers. Bytecode is -// content-addressed by wasmHash and immutable by construction; mutating a -// returned slice would corrupt the buffer's encapsulated state. -// Thread-safe: uses read lock. +// GetProtocolWasmBytecodes returns the wasmHash → bytecode map. The []byte values +// alias the buffer's internal storage and MUST be treated as read-only by callers. +// Bytecode is content-addressed by wasmHash and immutable by construction; mutating +// a returned slice would corrupt the buffer's encapsulated state. func (b *IndexerBuffer) GetProtocolWasmBytecodes() map[string][]byte { - b.mu.RLock() - defer b.mu.RUnlock() - - return maps.Clone(b.wasmBytecodesByHash) + return b.wasmBytecodesByHash } // PushProtocolContracts adds a protocol contract to the buffer with deduplication (last-write-wins). -// Thread-safe: acquires write lock. func (b *IndexerBuffer) PushProtocolContracts(contract data.ProtocolContracts) { - b.mu.Lock() - defer b.mu.Unlock() - b.protocolContractsByID[string(contract.ContractID)] = contract } -// GetProtocolContracts returns a clone of the protocol contracts map. -// Thread-safe: uses read lock. +// GetProtocolContracts returns the protocol contracts map. func (b *IndexerBuffer) GetProtocolContracts() map[string]data.ProtocolContracts { - b.mu.RLock() - defer b.mu.RUnlock() - - return maps.Clone(b.protocolContractsByID) + return b.protocolContractsByID } // PushContractEvents stashes the contract events emitted by a single @@ -673,27 +504,19 @@ func (b *IndexerBuffer) GetProtocolContracts() map[string]data.ProtocolContracts // processors consume from this map instead of re-decoding LedgerCloseMeta. // First-write wins on key collisions (which should not occur under the // indexer's parallel-per-tx split). -// Thread-safe: acquires write lock. func (b *IndexerBuffer) PushContractEvents(key ContractEventKey, events []xdr.ContractEvent) { if len(events) == 0 { return } - b.mu.Lock() - defer b.mu.Unlock() - if _, exists := b.contractEventsByKey[key]; !exists { b.contractEventsByKey[key] = events } } -// GetContractEvents returns a shallow clone of the contract-events map. -// Event slices alias buffer-owned storage and MUST be treated as read-only. -// Thread-safe: uses read lock. +// GetContractEvents returns the contract-events map. Event slices alias +// buffer-owned storage and MUST be treated as read-only. func (b *IndexerBuffer) GetContractEvents() map[ContractEventKey][]xdr.ContractEvent { - b.mu.RLock() - defer b.mu.RUnlock() - - return maps.Clone(b.contractEventsByKey) + return b.contractEventsByKey } // ParseAssetString parses a "CODE:ISSUER" formatted asset string into its components. diff --git a/internal/indexer/indexer_buffer_bench_test.go b/internal/indexer/indexer_buffer_bench_test.go new file mode 100644 index 000000000..dbbe3d8bf --- /dev/null +++ b/internal/indexer/indexer_buffer_bench_test.go @@ -0,0 +1,92 @@ +package indexer + +import ( + "context" + "path/filepath" + "runtime" + "testing" + + "github.com/alitto/pond/v2" + + "github.com/stellar/go-stellar-sdk/network" +) + +// These benchmarks run against the real pubnet ledger fixtures under testdata/ (see +// loadLedgerFixture), so they exercise the actual transaction/operation/change shapes production +// ingests rather than synthetic data. Fixtures are pubnet, hence PublicNetworkPassphrase. + +// benchIndexer builds an Indexer with the real processors. nil metrics is accepted (see +// TestIndexer_ProcessLedgerTransactions_FeePhaseBalances). +func benchIndexer() *Indexer { + return NewIndexer(network.PublicNetworkPassphrase, pond.NewPool(runtime.NumCPU()), nil) +} + +// BenchmarkProcessRealLedger measures a full per-ledger indexing pass: parallel workers build a +// per-tx TransactionResult, which are folded into one ledger buffer. One sub-benchmark per fixture. +func BenchmarkProcessRealLedger(b *testing.B) { + ctx := context.Background() + idx := benchIndexer() + + paths, err := filepath.Glob("testdata/*.xdr.gz") + if err != nil { + b.Fatal(err) + } + for _, path := range paths { + lcm := loadLedgerFixture(b, path) + txs, err := GetLedgerTransactions(ctx, network.PublicNetworkPassphrase, lcm) + if err != nil { + b.Fatal(err) + } + + b.Run(filepath.Base(path), func(b *testing.B) { + buf := NewIndexerBuffer() + b.ReportAllocs() + for b.Loop() { + if _, err := idx.ProcessLedgerTransactions(ctx, txs, buf); err != nil { + b.Fatal(err) + } + buf.Clear() + } + }) + } +} + +// BenchmarkFoldRealLedger isolates the serial fold this change actually touches: it pre-builds each +// real ledger's per-tx results once (untimed) and then measures only IngestTransactionResult + the +// buffer Clear that mirrors backfill reuse. One sub-benchmark per fixture. +func BenchmarkFoldRealLedger(b *testing.B) { + ctx := context.Background() + idx := benchIndexer() + + paths, err := filepath.Glob("testdata/*.xdr.gz") + if err != nil { + b.Fatal(err) + } + for _, path := range paths { + lcm := loadLedgerFixture(b, path) + txs, err := GetLedgerTransactions(ctx, network.PublicNetworkPassphrase, lcm) + if err != nil { + b.Fatal(err) + } + + results := make([]*TransactionResult, 0, len(txs)) + for _, tx := range txs { + result, err := idx.processTransaction(ctx, tx) + if err != nil { + b.Fatal(err) + } + results = append(results, result) + } + + b.Run(filepath.Base(path), func(b *testing.B) { + buf := NewIndexerBuffer() + b.ReportAllocs() + for b.Loop() { + for _, result := range results { + buf.IngestTransactionResult(result) + } + buf.Clear() + } + }) + } +} diff --git a/internal/indexer/indexer_buffer_test.go b/internal/indexer/indexer_buffer_test.go index e22f38311..1025e7328 100644 --- a/internal/indexer/indexer_buffer_test.go +++ b/internal/indexer/indexer_buffer_test.go @@ -1,7 +1,6 @@ package indexer import ( - "sync" "testing" set "github.com/deckarep/golang-set/v2" @@ -29,15 +28,15 @@ func TestIndexerBuffer_PushTransaction(t *testing.T) { tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - indexerBuffer.PushTransaction("alice", tx1) - indexerBuffer.PushTransaction("alice", tx2) - indexerBuffer.PushTransaction("bob", tx2) - indexerBuffer.PushTransaction("bob", tx2) // duplicate is a no-op + indexerBuffer.PushTransaction("alice", &tx1) + indexerBuffer.PushTransaction("alice", &tx2) + indexerBuffer.PushTransaction("bob", &tx2) + indexerBuffer.PushTransaction("bob", &tx2) // duplicate is a no-op // Assert participants by transaction txParticipants := indexerBuffer.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("alice"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("alice", "bob"), txParticipants[tx2.ToID]) + assert.Equal(t, set.NewThreadUnsafeSet("alice"), txParticipants[tx1.ToID]) + assert.Equal(t, set.NewThreadUnsafeSet("alice", "bob"), txParticipants[tx2.ToID]) // Assert GetNumberOfTransactions assert.Equal(t, 2, indexerBuffer.GetNumberOfTransactions()) @@ -45,41 +44,6 @@ func TestIndexerBuffer_PushTransaction(t *testing.T) { // Assert GetAllTransactions assert.ElementsMatch(t, []*types.Transaction{&tx1, &tx2}, indexerBuffer.GetTransactions()) }) - - t.Run("🟢 concurrent pushes", func(t *testing.T) { - indexerBuffer := NewIndexerBuffer() - - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - - wg := sync.WaitGroup{} - wg.Add(4) - go func() { - indexerBuffer.PushTransaction("alice", tx1) - wg.Done() - }() - go func() { - indexerBuffer.PushTransaction("alice", tx2) - wg.Done() - }() - go func() { - indexerBuffer.PushTransaction("bob", tx2) - wg.Done() - }() - go func() { - indexerBuffer.PushTransaction("bob", tx2) // duplicate is a no-op - wg.Done() - }() - wg.Wait() - - // Assert participants by transaction - txParticipants := indexerBuffer.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("alice"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("alice", "bob"), txParticipants[tx2.ToID]) - - // Assert GetNumberOfTransactions - assert.Equal(t, 2, indexerBuffer.GetNumberOfTransactions()) - }) } func TestIndexerBuffer_PushOperation(t *testing.T) { @@ -91,54 +55,20 @@ func TestIndexerBuffer_PushOperation(t *testing.T) { op1 := types.Operation{ID: 1} op2 := types.Operation{ID: 2} - indexerBuffer.PushOperation("alice", op1, tx1) - indexerBuffer.PushOperation("bob", op2, tx2) - indexerBuffer.PushOperation("chuck", op2, tx2) - indexerBuffer.PushOperation("chuck", op2, tx2) // duplicate operation ID is a no-op + indexerBuffer.PushOperation("alice", &op1, &tx1) + indexerBuffer.PushOperation("bob", &op2, &tx2) + indexerBuffer.PushOperation("chuck", &op2, &tx2) + indexerBuffer.PushOperation("chuck", &op2, &tx2) // duplicate operation ID is a no-op // Assert participants by operation opParticipants := indexerBuffer.GetOperationsParticipants() - assert.Equal(t, set.NewSet("alice"), opParticipants[int64(1)]) - assert.Equal(t, set.NewSet("bob", "chuck"), opParticipants[int64(2)]) + assert.Equal(t, set.NewThreadUnsafeSet("alice"), opParticipants[int64(1)]) + assert.Equal(t, set.NewThreadUnsafeSet("bob", "chuck"), opParticipants[int64(2)]) // Assert transactions were also added txParticipants := indexerBuffer.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("alice"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("bob", "chuck"), txParticipants[tx2.ToID]) - }) - - t.Run("🟢 concurrent pushes", func(t *testing.T) { - indexerBuffer := NewIndexerBuffer() - - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - op1 := types.Operation{ID: 1} - op2 := types.Operation{ID: 2} - - wg := sync.WaitGroup{} - wg.Add(4) - go func() { - indexerBuffer.PushOperation("alice", op1, tx1) - wg.Done() - }() - go func() { - indexerBuffer.PushOperation("bob", op2, tx2) - wg.Done() - }() - go func() { - indexerBuffer.PushOperation("chuck", op2, tx2) - wg.Done() - }() - go func() { - indexerBuffer.PushOperation("chuck", op2, tx2) // duplicate operation ID is a no-op - wg.Done() - }() - wg.Wait() - - // Assert participants by operation - opParticipants := indexerBuffer.GetOperationsParticipants() - assert.Equal(t, set.NewSet("alice"), opParticipants[int64(1)]) - assert.Equal(t, set.NewSet("bob", "chuck"), opParticipants[int64(2)]) + assert.Equal(t, set.NewThreadUnsafeSet("alice"), txParticipants[tx1.ToID]) + assert.Equal(t, set.NewThreadUnsafeSet("bob", "chuck"), txParticipants[tx2.ToID]) }) } @@ -153,45 +83,14 @@ func TestIndexerBuffer_PushStateChange(t *testing.T) { sc2 := types.StateChange{ToID: 2, StateChangeID: 1} sc3 := types.StateChange{ToID: 3, StateChangeID: 1} - indexerBuffer.PushStateChange(tx, op, sc1) - indexerBuffer.PushStateChange(tx, op, sc2) - indexerBuffer.PushStateChange(tx, op, sc3) + indexerBuffer.PushStateChange(&tx, &op, sc1) + indexerBuffer.PushStateChange(&tx, &op, sc2) + indexerBuffer.PushStateChange(&tx, &op, sc3) allStateChanges := indexerBuffer.GetStateChanges() assert.Equal(t, []types.StateChange{sc1, sc2, sc3}, allStateChanges) }) - t.Run("🟢 concurrent pushes", func(t *testing.T) { - indexerBuffer := NewIndexerBuffer() - - tx := types.Transaction{Hash: "c76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48763", ToID: 1} - op := types.Operation{ID: 1} - - sc1 := types.StateChange{ToID: 1, StateChangeID: 1} - sc2 := types.StateChange{ToID: 2, StateChangeID: 1} - sc3 := types.StateChange{ToID: 3, StateChangeID: 1} - - wg := sync.WaitGroup{} - wg.Add(2) - - go func() { - defer wg.Done() - indexerBuffer.PushStateChange(tx, op, sc1) - }() - - go func() { - defer wg.Done() - indexerBuffer.PushStateChange(tx, op, sc2) - indexerBuffer.PushStateChange(tx, op, sc3) - }() - - wg.Wait() - - allStateChanges := indexerBuffer.GetStateChanges() - assert.Len(t, allStateChanges, 3) - assert.ElementsMatch(t, []types.StateChange{sc1, sc2, sc3}, allStateChanges) - }) - t.Run("🟢 with operations and transactions", func(t *testing.T) { indexerBuffer := NewIndexerBuffer() @@ -200,8 +99,8 @@ func TestIndexerBuffer_PushStateChange(t *testing.T) { op1 := types.Operation{ID: 3} op2 := types.Operation{ID: 4} op3 := types.Operation{ID: 5} - indexerBuffer.PushOperation("someone", op1, tx1) - indexerBuffer.PushOperation("someone", op2, tx2) + indexerBuffer.PushOperation("someone", &op1, &tx1) + indexerBuffer.PushOperation("someone", &op2, &tx2) sc1 := buildStateChange(3, types.StateChangeReasonCredit, "alice", op1.ID) sc2 := buildStateChange(4, types.StateChangeReasonDebit, "alice", op2.ID) @@ -210,25 +109,25 @@ func TestIndexerBuffer_PushStateChange(t *testing.T) { sc4 := buildStateChange(1, types.StateChangeReasonDebit, "bob", 0) sc5 := buildStateChange(2, types.StateChangeReasonDebit, "bob", 0) - indexerBuffer.PushStateChange(tx1, op1, sc1) - indexerBuffer.PushStateChange(tx2, op2, sc2) - indexerBuffer.PushStateChange(tx2, op3, sc3) // This operation should be added - indexerBuffer.PushStateChange(tx2, types.Operation{}, sc4) // Fee state changes don't have an operation - indexerBuffer.PushStateChange(tx2, types.Operation{}, sc5) // Fee state changes don't have an operation + indexerBuffer.PushStateChange(&tx1, &op1, sc1) + indexerBuffer.PushStateChange(&tx2, &op2, sc2) + indexerBuffer.PushStateChange(&tx2, &op3, sc3) // This operation should be added + indexerBuffer.PushStateChange(&tx2, nil, sc4) // Fee state changes don't have an operation + indexerBuffer.PushStateChange(&tx2, nil, sc5) // Fee state changes don't have an operation allStateChanges := indexerBuffer.GetStateChanges() assert.Equal(t, []types.StateChange{sc1, sc2, sc3, sc4, sc5}, allStateChanges) // Assert transaction participants txParticipants := indexerBuffer.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("someone", "alice"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("someone", "alice", "eve", "bob"), txParticipants[tx2.ToID]) + assert.Equal(t, set.NewThreadUnsafeSet("someone", "alice"), txParticipants[tx1.ToID]) + assert.Equal(t, set.NewThreadUnsafeSet("someone", "alice", "eve", "bob"), txParticipants[tx2.ToID]) // Assert operation participants opParticipants := indexerBuffer.GetOperationsParticipants() - assert.Equal(t, set.NewSet("someone", "alice"), opParticipants[int64(3)]) - assert.Equal(t, set.NewSet("someone", "alice"), opParticipants[int64(4)]) - assert.Equal(t, set.NewSet("eve"), opParticipants[int64(5)]) + assert.Equal(t, set.NewThreadUnsafeSet("someone", "alice"), opParticipants[int64(3)]) + assert.Equal(t, set.NewThreadUnsafeSet("someone", "alice"), opParticipants[int64(4)]) + assert.Equal(t, set.NewThreadUnsafeSet("eve"), opParticipants[int64(5)]) }) } @@ -241,14 +140,14 @@ func TestIndexerBuffer_GetNumberOfTransactions(t *testing.T) { tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - indexerBuffer.PushTransaction("alice", tx1) + indexerBuffer.PushTransaction("alice", &tx1) assert.Equal(t, 1, indexerBuffer.GetNumberOfTransactions()) - indexerBuffer.PushTransaction("bob", tx2) + indexerBuffer.PushTransaction("bob", &tx2) assert.Equal(t, 2, indexerBuffer.GetNumberOfTransactions()) // Duplicate should not increase count - indexerBuffer.PushTransaction("charlie", tx2) + indexerBuffer.PushTransaction("charlie", &tx2) assert.Equal(t, 2, indexerBuffer.GetNumberOfTransactions()) }) } @@ -260,9 +159,9 @@ func TestIndexerBuffer_GetAllTransactions(t *testing.T) { tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1, LedgerNumber: 100} tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2, LedgerNumber: 101} - indexerBuffer.PushTransaction("alice", tx1) - indexerBuffer.PushTransaction("bob", tx2) - indexerBuffer.PushTransaction("charlie", tx2) // duplicate + indexerBuffer.PushTransaction("alice", &tx1) + indexerBuffer.PushTransaction("bob", &tx2) + indexerBuffer.PushTransaction("charlie", &tx2) // duplicate allTxs := indexerBuffer.GetTransactions() require.Len(t, allTxs, 2) @@ -277,13 +176,13 @@ func TestIndexerBuffer_GetAllTransactionsParticipants(t *testing.T) { tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - indexerBuffer.PushTransaction("alice", tx1) - indexerBuffer.PushTransaction("bob", tx1) - indexerBuffer.PushTransaction("alice", tx2) + indexerBuffer.PushTransaction("alice", &tx1) + indexerBuffer.PushTransaction("bob", &tx1) + indexerBuffer.PushTransaction("alice", &tx2) txParticipants := indexerBuffer.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("alice", "bob"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("alice"), txParticipants[tx2.ToID]) + assert.Equal(t, set.NewThreadUnsafeSet("alice", "bob"), txParticipants[tx1.ToID]) + assert.Equal(t, set.NewThreadUnsafeSet("alice"), txParticipants[tx2.ToID]) }) } @@ -295,9 +194,9 @@ func TestIndexerBuffer_GetAllOperations(t *testing.T) { op1 := types.Operation{ID: 1} op2 := types.Operation{ID: 2} - indexerBuffer.PushOperation("alice", op1, tx1) - indexerBuffer.PushOperation("bob", op2, tx1) - indexerBuffer.PushOperation("charlie", op2, tx1) // duplicate + indexerBuffer.PushOperation("alice", &op1, &tx1) + indexerBuffer.PushOperation("bob", &op2, &tx1) + indexerBuffer.PushOperation("charlie", &op2, &tx1) // duplicate allOps := indexerBuffer.GetOperations() require.Len(t, allOps, 2) @@ -313,13 +212,13 @@ func TestIndexerBuffer_GetAllOperationsParticipants(t *testing.T) { op1 := types.Operation{ID: 1} op2 := types.Operation{ID: 2} - indexerBuffer.PushOperation("alice", op1, tx1) - indexerBuffer.PushOperation("bob", op1, tx1) - indexerBuffer.PushOperation("alice", op2, tx1) + indexerBuffer.PushOperation("alice", &op1, &tx1) + indexerBuffer.PushOperation("bob", &op1, &tx1) + indexerBuffer.PushOperation("alice", &op2, &tx1) opParticipants := indexerBuffer.GetOperationsParticipants() - assert.Equal(t, set.NewSet("alice", "bob"), opParticipants[int64(1)]) - assert.Equal(t, set.NewSet("alice"), opParticipants[int64(2)]) + assert.Equal(t, set.NewThreadUnsafeSet("alice", "bob"), opParticipants[int64(1)]) + assert.Equal(t, set.NewThreadUnsafeSet("alice"), opParticipants[int64(2)]) }) } @@ -334,235 +233,62 @@ func TestIndexerBuffer_GetAllStateChanges(t *testing.T) { sc2 := types.StateChange{ToID: 2, StateChangeID: 1, AccountID: "bob"} sc3 := types.StateChange{ToID: 3, StateChangeID: 1, AccountID: "charlie"} - indexerBuffer.PushStateChange(tx, op, sc1) - indexerBuffer.PushStateChange(tx, op, sc2) - indexerBuffer.PushStateChange(tx, op, sc3) + indexerBuffer.PushStateChange(&tx, &op, sc1) + indexerBuffer.PushStateChange(&tx, &op, sc2) + indexerBuffer.PushStateChange(&tx, &op, sc3) allStateChanges := indexerBuffer.GetStateChanges() assert.Equal(t, []types.StateChange{sc1, sc2, sc3}, allStateChanges) }) } -func TestIndexerBuffer_Merge(t *testing.T) { - t.Run("🟢 merge empty buffers", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - buffer1.Merge(buffer2) - assert.Equal(t, 0, buffer1.GetNumberOfTransactions()) - assert.Len(t, buffer1.GetStateChanges(), 0) - }) - - t.Run("🟢 merge transactions only", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - - buffer1.PushTransaction("alice", tx1) - buffer2.PushTransaction("bob", tx2) - - buffer1.Merge(buffer2) - - // Verify transactions - allTxs := buffer1.GetTransactions() - assert.Len(t, allTxs, 2) - assert.ElementsMatch(t, []*types.Transaction{&tx1, &tx2}, allTxs) - - // Verify transaction participants - txParticipants := buffer1.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("alice"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("bob"), txParticipants[tx2.ToID]) - }) - - t.Run("🟢 merge operations only", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - op1 := types.Operation{ID: 1} - op2 := types.Operation{ID: 2} - - buffer1.PushOperation("alice", op1, tx1) - buffer2.PushOperation("bob", op2, tx1) - - buffer1.Merge(buffer2) - - // Verify operations - allOps := buffer1.GetOperations() - assert.Len(t, allOps, 2) - assert.ElementsMatch(t, []*types.Operation{&op1, &op2}, allOps) - - // Verify operation participants - opParticipants := buffer1.GetOperationsParticipants() - assert.Equal(t, set.NewSet("alice"), opParticipants[int64(1)]) - assert.Equal(t, set.NewSet("bob"), opParticipants[int64(2)]) - }) - - t.Run("🟢 merge state changes only", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - tx := types.Transaction{Hash: "c76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48763", ToID: 1} - op := types.Operation{ID: 1} - - sc1 := types.StateChange{ToID: 1, StateChangeID: 1, AccountID: "alice"} - sc2 := types.StateChange{ToID: 2, StateChangeID: 1, AccountID: "bob"} - - buffer1.PushStateChange(tx, op, sc1) - buffer2.PushStateChange(tx, op, sc2) - - buffer1.Merge(buffer2) - - // Verify state changes - allStateChanges := buffer1.GetStateChanges() - assert.Len(t, allStateChanges, 2) - assert.Equal(t, sc1, allStateChanges[0]) - assert.Equal(t, sc2, allStateChanges[1]) - }) - - t.Run("🟢 merge with overlapping data", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - op1 := types.Operation{ID: 1} - - // Buffer1 has tx1 with alice - buffer1.PushTransaction("alice", tx1) - buffer1.PushOperation("alice", op1, tx1) - - // Buffer2 has tx1 with bob (overlapping tx) and tx2 with charlie - buffer2.PushTransaction("bob", tx1) - buffer2.PushTransaction("charlie", tx2) - buffer2.PushOperation("bob", op1, tx1) - - buffer1.Merge(buffer2) - - // Verify transactions - allTxs := buffer1.GetTransactions() - assert.Len(t, allTxs, 2) - - // Verify tx1 has both alice and bob as participants - txParticipants := buffer1.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("alice", "bob"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("charlie"), txParticipants[tx2.ToID]) - - // Verify operation participants merged - opParticipants := buffer1.GetOperationsParticipants() - assert.Equal(t, set.NewSet("alice", "bob"), opParticipants[int64(1)]) - }) - - t.Run("🟢 merge into empty buffer", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - op1 := types.Operation{ID: 1} - sc1 := types.StateChange{ToID: 1, StateChangeID: 1, AccountID: "alice"} - - buffer2.PushTransaction("alice", tx1) - buffer2.PushOperation("bob", op1, tx1) - buffer2.PushStateChange(tx1, op1, sc1) - - buffer1.Merge(buffer2) - - assert.Equal(t, 1, buffer1.GetNumberOfTransactions()) - assert.Len(t, buffer1.GetOperations(), 1) - assert.Len(t, buffer1.GetStateChanges(), 1) - }) - - t.Run("🟢 merge empty buffer into populated", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() +func TestIndexerBuffer_IngestTransactionResult(t *testing.T) { + t.Run("🟢 folds a single result into the buffer", func(t *testing.T) { + buffer := NewIndexerBuffer() - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - buffer1.PushTransaction("alice", tx1) + tx := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} + op := types.Operation{ID: 10} + sc := buildStateChange(1, types.StateChangeReasonCredit, "alice", 10) + + result := &TransactionResult{ + Transaction: &tx, + TxParticipants: []string{"alice"}, + Operations: map[int64]*types.Operation{10: &op}, + OpParticipants: map[int64][]string{10: {"alice", "bob"}}, + StateChanges: []types.StateChange{sc}, + AccountChanges: []types.AccountChange{{AccountID: "alice", SortKey: 1, Operation: types.AccountOpUpdate, Balance: 100}}, + ParticipantCount: 2, + } - buffer1.Merge(buffer2) + buffer.IngestTransactionResult(result) - assert.Equal(t, 1, buffer1.GetNumberOfTransactions()) + assert.Equal(t, 1, buffer.GetNumberOfTransactions()) + assert.Equal(t, 1, buffer.GetNumberOfOperations()) + assert.Equal(t, set.NewThreadUnsafeSet("alice", "bob"), buffer.GetOperationsParticipants()[10]) + assert.Equal(t, set.NewThreadUnsafeSet("alice", "bob"), buffer.GetTransactionsParticipants()[1]) + assert.Len(t, buffer.GetStateChanges(), 1) + assert.Len(t, buffer.GetAccountChanges(), 1) }) - t.Run("🟢 concurrent merges", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - buffer3 := NewIndexerBuffer() - - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - tx3 := types.Transaction{Hash: "b76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48762", ToID: 3} - - buffer1.PushTransaction("alice", tx1) - buffer2.PushTransaction("bob", tx2) - buffer3.PushTransaction("charlie", tx3) - - wg := sync.WaitGroup{} - wg.Add(2) - - go func() { - defer wg.Done() - buffer1.Merge(buffer2) - }() - - go func() { - defer wg.Done() - buffer1.Merge(buffer3) - }() - - wg.Wait() + t.Run("🟢 dedups across multiple folded results (CREATE→REMOVE tombstone)", func(t *testing.T) { + buffer := NewIndexerBuffer() + tx := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - // Verify all data merged correctly - assert.Equal(t, 3, buffer1.GetNumberOfTransactions()) - }) + // One result creates the account, a later result removes it within the same ledger. Folded + // through the same buffer, the CREATE→REMOVE nets to nothing (see pushWithTombstone). + create := &TransactionResult{ + Transaction: &tx, + AccountChanges: []types.AccountChange{{AccountID: "GACCT", SortKey: 1, Operation: types.AccountOpCreate, Balance: 100}}, + } + remove := &TransactionResult{ + Transaction: &tx, + AccountChanges: []types.AccountChange{{AccountID: "GACCT", SortKey: 2, Operation: types.AccountOpRemove}}, + } - t.Run("🟢 merge complete buffers with all data types", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() + buffer.IngestTransactionResult(create) + buffer.IngestTransactionResult(remove) - tx1 := types.Transaction{Hash: "e76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48760", ToID: 1} - tx2 := types.Transaction{Hash: "a76b7b0133690fbfb2de8fa9ca2273cb4f2e29447e0cf0e14a5f82d0daa48761", ToID: 2} - op1 := types.Operation{ID: 1} - op2 := types.Operation{ID: 2} - sc1 := types.StateChange{ToID: 1, StateChangeID: 1, AccountID: "alice", OperationID: 1} - sc2 := types.StateChange{ToID: 2, StateChangeID: 1, AccountID: "bob", OperationID: 2} - - // Buffer1 - buffer1.PushTransaction("alice", tx1) - buffer1.PushOperation("alice", op1, tx1) - buffer1.PushStateChange(tx1, op1, sc1) - - // Buffer2 - buffer2.PushTransaction("bob", tx2) - buffer2.PushOperation("bob", op2, tx2) - buffer2.PushStateChange(tx2, op2, sc2) - - buffer1.Merge(buffer2) - - // Verify transactions - allTxs := buffer1.GetTransactions() - assert.Len(t, allTxs, 2) - - // Verify operations - allOps := buffer1.GetOperations() - assert.Len(t, allOps, 2) - - // Verify state changes - allStateChanges := buffer1.GetStateChanges() - assert.Len(t, allStateChanges, 2) - assert.Equal(t, sc1, allStateChanges[0]) - assert.Equal(t, sc2, allStateChanges[1]) - - // Verify participants mappings - txParticipants := buffer1.GetTransactionsParticipants() - assert.Equal(t, set.NewSet("alice"), txParticipants[tx1.ToID]) - assert.Equal(t, set.NewSet("bob"), txParticipants[tx2.ToID]) - - opParticipants := buffer1.GetOperationsParticipants() - assert.Equal(t, set.NewSet("alice"), opParticipants[int64(1)]) - assert.Equal(t, set.NewSet("bob"), opParticipants[int64(2)]) + assert.Len(t, buffer.GetAccountChanges(), 0) }) } @@ -747,138 +473,6 @@ func TestIndexerBuffer_PushSACBalanceChange(t *testing.T) { }) } -func TestIndexerBuffer_MergeSACBalanceChanges(t *testing.T) { - t.Run("🟢 merge SAC balance changes from two buffers", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - account1 := "CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC" - account2 := "CAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAD2KM" - contract1 := "CCWAMYJME4H5CKG7OLXGC2T4M6FL52XCZ3OQOAV6LL3GLA4RO4WH3ASP" - - buffer1.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: account1, - ContractID: contract1, - Balance: "100", - Operation: types.SACBalanceOpAdd, - OperationID: 1, - }) - - buffer2.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: account2, - ContractID: contract1, - Balance: "200", - Operation: types.SACBalanceOpAdd, - OperationID: 2, - }) - - buffer1.Merge(buffer2) - - changes := buffer1.GetSACBalanceChanges() - assert.Len(t, changes, 2) - }) - - t.Run("🟢 merge keeps higher OperationID during merge", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - accountID := "CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC" - contractID := "CCWAMYJME4H5CKG7OLXGC2T4M6FL52XCZ3OQOAV6LL3GLA4RO4WH3ASP" - - buffer1.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: accountID, - ContractID: contractID, - Balance: "100", - Operation: types.SACBalanceOpAdd, - OperationID: 50, - }) - - buffer2.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: accountID, - ContractID: contractID, - Balance: "200", - Operation: types.SACBalanceOpUpdate, - OperationID: 100, // Higher - }) - - buffer1.Merge(buffer2) - - changes := buffer1.GetSACBalanceChanges() - assert.Len(t, changes, 1) - - key := SACBalanceChangeKey{AccountID: accountID, ContractID: contractID} - assert.Equal(t, "200", changes[key].Balance) - assert.Equal(t, int64(100), changes[key].OperationID) - }) - - t.Run("🟢 merge handles ADD→REMOVE no-op across buffers", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - accountID := "CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC" - contractID := "CCWAMYJME4H5CKG7OLXGC2T4M6FL52XCZ3OQOAV6LL3GLA4RO4WH3ASP" - - // Buffer1 has ADD - buffer1.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: accountID, - ContractID: contractID, - Balance: "100", - Operation: types.SACBalanceOpAdd, - OperationID: 50, - }) - - // Buffer2 has REMOVE (higher opID) - buffer2.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: accountID, - ContractID: contractID, - Balance: "0", - Operation: types.SACBalanceOpRemove, - OperationID: 100, - }) - - buffer1.Merge(buffer2) - - // ADD→REMOVE across merge is a no-op - changes := buffer1.GetSACBalanceChanges() - assert.Len(t, changes, 0) - }) - - t.Run("🟢 merge ignores lower OperationID from other buffer", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - accountID := "CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC" - contractID := "CCWAMYJME4H5CKG7OLXGC2T4M6FL52XCZ3OQOAV6LL3GLA4RO4WH3ASP" - - // Buffer1 has higher opID - buffer1.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: accountID, - ContractID: contractID, - Balance: "200", - Operation: types.SACBalanceOpUpdate, - OperationID: 100, - }) - - // Buffer2 has lower opID - should be ignored - buffer2.PushSACBalanceChange(types.SACBalanceChange{ - AccountID: accountID, - ContractID: contractID, - Balance: "50", - Operation: types.SACBalanceOpAdd, - OperationID: 50, - }) - - buffer1.Merge(buffer2) - - changes := buffer1.GetSACBalanceChanges() - assert.Len(t, changes, 1) - - key := SACBalanceChangeKey{AccountID: accountID, ContractID: contractID} - assert.Equal(t, "200", changes[key].Balance) - assert.Equal(t, int64(100), changes[key].OperationID) - }) -} - func TestIndexerBuffer_ProtocolWasms(t *testing.T) { t.Run("push and get protocol wasms with dedup", func(t *testing.T) { buffer := NewIndexerBuffer() @@ -897,22 +491,6 @@ func TestIndexerBuffer_ProtocolWasms(t *testing.T) { assert.Equal(t, wasm2, wasms[string(wasm2.WasmHash)]) }) - t.Run("merge protocol wasms first-write-wins", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - buffer1.PushProtocolWasm(data.ProtocolWasms{WasmHash: "0100000000000000000000000000000000000000000000000000000000000000"}) - buffer2.PushProtocolWasm(data.ProtocolWasms{WasmHash: "0100000000000000000000000000000000000000000000000000000000000000"}) // duplicate across buffers - buffer2.PushProtocolWasm(data.ProtocolWasms{WasmHash: "0200000000000000000000000000000000000000000000000000000000000000"}) - - buffer1.Merge(buffer2) - - wasms := buffer1.GetProtocolWasms() - assert.Len(t, wasms, 2) - assert.Contains(t, wasms, "0100000000000000000000000000000000000000000000000000000000000000") - assert.Contains(t, wasms, "0200000000000000000000000000000000000000000000000000000000000000") - }) - t.Run("clear protocol wasms", func(t *testing.T) { buffer := NewIndexerBuffer() buffer.PushProtocolWasm(data.ProtocolWasms{WasmHash: "0100000000000000000000000000000000000000000000000000000000000000"}) @@ -945,23 +523,6 @@ func TestIndexerBuffer_ProtocolContracts(t *testing.T) { assert.Equal(t, c2, contracts[string(c2.ContractID)]) }) - t.Run("merge protocol contracts last-write-wins", func(t *testing.T) { - buffer1 := NewIndexerBuffer() - buffer2 := NewIndexerBuffer() - - buffer1.PushProtocolContracts(data.ProtocolContracts{ContractID: "0100000000000000000000000000000000000000000000000000000000000000", WasmHash: "0a00000000000000000000000000000000000000000000000000000000000000"}) - buffer2.PushProtocolContracts(data.ProtocolContracts{ContractID: "0100000000000000000000000000000000000000000000000000000000000000", WasmHash: "6300000000000000000000000000000000000000000000000000000000000000"}) // upgrade - buffer2.PushProtocolContracts(data.ProtocolContracts{ContractID: "0200000000000000000000000000000000000000000000000000000000000000", WasmHash: "1400000000000000000000000000000000000000000000000000000000000000"}) - - buffer1.Merge(buffer2) - - contracts := buffer1.GetProtocolContracts() - assert.Len(t, contracts, 2) - // Last-write-wins: buffer2's value (later data) should overwrite buffer1's - assert.Equal(t, types.HashBytea("6300000000000000000000000000000000000000000000000000000000000000"), contracts["0100000000000000000000000000000000000000000000000000000000000000"].WasmHash) - assert.Equal(t, types.HashBytea("1400000000000000000000000000000000000000000000000000000000000000"), contracts["0200000000000000000000000000000000000000000000000000000000000000"].WasmHash) - }) - t.Run("clear protocol contracts", func(t *testing.T) { buffer := NewIndexerBuffer() buffer.PushProtocolContracts(data.ProtocolContracts{ContractID: "0100000000000000000000000000000000000000000000000000000000000000", WasmHash: "0a00000000000000000000000000000000000000000000000000000000000000"}) @@ -1154,104 +715,6 @@ func TestIndexerBuffer_PushAccountChange(t *testing.T) { }) } -func TestIndexerBuffer_MergeAccountChanges(t *testing.T) { - // Each tx is processed into its own buffer, then merged. Because the dedup key is a - // total order with no ties, the winner is the same no matter what order buffers merge — - // that is what makes parallel-then-merge safe. - accountChange := func(sortKey, balance int64) types.AccountChange { - return types.AccountChange{ - AccountID: accountChangeAddr, - SortKey: sortKey, - LedgerNumber: 12345, - Operation: types.AccountOpUpdate, - Balance: balance, - } - } - - // mergeBalance pushes a and b into separate buffers, merges them in the given order into - // a fresh buffer, and returns the surviving balance for accountChangeAddr. - mergeBalance := func(a, b types.AccountChange) int64 { - bufA := NewIndexerBuffer() - bufA.PushAccountChange(a) - bufB := NewIndexerBuffer() - bufB.PushAccountChange(b) - - merged := NewIndexerBuffer() - merged.Merge(bufA) - merged.Merge(bufB) - return merged.GetAccountChanges()[accountChangeAddr].Balance - } - - t.Run("🟢 operation beats fee across buffers", func(t *testing.T) { - op := accountChange(accountRank(rankOp, 2, 1), 500) - fee := accountChange(accountRank(rankFee, 9, 0), 999) - assert.Equal(t, int64(500), mergeBalance(op, fee)) - assert.Equal(t, int64(500), mergeBalance(fee, op)) - }) - - t.Run("🟢 refund beats operation across buffers", func(t *testing.T) { - op := accountChange(accountRank(rankOp, 8, 1), 500) - refund := accountChange(accountRank(rankRefund, 3, 0), 550) - assert.Equal(t, int64(550), mergeBalance(op, refund)) - assert.Equal(t, int64(550), mergeBalance(refund, op)) - }) - - t.Run("🟢 later-tx fee wins independent of merge order", func(t *testing.T) { - early := accountChange(accountRank(rankFee, 3, 0), 100) - late := accountChange(accountRank(rankFee, 7, 0), 200) - assert.Equal(t, int64(200), mergeBalance(early, late)) - assert.Equal(t, int64(200), mergeBalance(late, early)) - }) - - t.Run("🟢 later-tx refund wins independent of merge order", func(t *testing.T) { - early := accountChange(accountRank(rankRefund, 3, 0), 100) - late := accountChange(accountRank(rankRefund, 7, 0), 200) - assert.Equal(t, int64(200), mergeBalance(early, late)) - assert.Equal(t, int64(200), mergeBalance(late, early)) - }) - - // cancelledBuffer returns a buffer in which the account was created then removed within its - // own operations — its map entry is gone, leaving only a tombstone that Merge must propagate. - cancelledBuffer := func() *IndexerBuffer { - buf := NewIndexerBuffer() - buf.PushAccountChange(types.AccountChange{AccountID: accountChangeAddr, SortKey: accountRank(rankOp, 1, 1), Operation: types.AccountOpCreate, Balance: 100}) - buf.PushAccountChange(types.AccountChange{AccountID: accountChangeAddr, SortKey: accountRank(rankOp, 1, 2), Operation: types.AccountOpRemove}) - return buf - } - - t.Run("🟢 tombstone propagates across merge and blocks lower-key change", func(t *testing.T) { - lower := NewIndexerBuffer() - lower.PushAccountChange(accountChange(accountRank(rankFee, 1, 0), 999)) - - ab := NewIndexerBuffer() - ab.Merge(cancelledBuffer()) - ab.Merge(lower) - assert.Len(t, ab.GetAccountChanges(), 0) - - ba := NewIndexerBuffer() - ba.Merge(lower) - ba.Merge(cancelledBuffer()) - assert.Len(t, ba.GetAccountChanges(), 0) - }) - - t.Run("🟢 higher-key change across merge re-creates a tombstoned account", func(t *testing.T) { - recreate := NewIndexerBuffer() - recreate.PushAccountChange(types.AccountChange{AccountID: accountChangeAddr, SortKey: accountRank(rankOp, 5, 1), Operation: types.AccountOpCreate, Balance: 700}) - - ab := NewIndexerBuffer() - ab.Merge(cancelledBuffer()) - ab.Merge(recreate) - require.Len(t, ab.GetAccountChanges(), 1) - assert.Equal(t, int64(700), ab.GetAccountChanges()[accountChangeAddr].Balance) - - ba := NewIndexerBuffer() - ba.Merge(recreate) - ba.Merge(cancelledBuffer()) - require.Len(t, ba.GetAccountChanges(), 1) - assert.Equal(t, int64(700), ba.GetAccountChanges()[accountChangeAddr].Balance) - }) -} - func TestIndexerBuffer_TrustlineTombstone(t *testing.T) { const asset = "USDC:GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN" trustline := func(opID, balance int64, op types.TrustlineOpType) types.TrustlineChange { @@ -1287,17 +750,4 @@ func TestIndexerBuffer_TrustlineTombstone(t *testing.T) { assert.Equal(t, int64(700), c.Balance) } }) - - t.Run("🟢 tombstone propagates across merge and blocks lower-key change", func(t *testing.T) { - cancelled := NewIndexerBuffer() - cancelled.PushTrustlineChange(trustline(100, 1000, types.TrustlineOpAdd)) - cancelled.PushTrustlineChange(trustline(200, 0, types.TrustlineOpRemove)) - lower := NewIndexerBuffer() - lower.PushTrustlineChange(trustline(50, 500, types.TrustlineOpUpdate)) - - ab := NewIndexerBuffer() - ab.Merge(cancelled) - ab.Merge(lower) - assert.Len(t, ab.GetTrustlineChanges(), 0) - }) } diff --git a/internal/indexer/indexer_test.go b/internal/indexer/indexer_test.go index 610dc5027..b6ebf3335 100644 --- a/internal/indexer/indexer_test.go +++ b/internal/indexer/indexer_test.go @@ -1103,7 +1103,7 @@ func extractContractEventsViaReader(ctx context.Context, networkPassphrase strin // To add fixtures: build ingest.NewLedgerBackend (datastore + PublicNetworkPassphrase), // then per sequence call GetLedger, MarshalBinary, gzip, and write // testdata/ledger-.xdr.gz. Keep both wrapper versions represented. -func loadLedgerFixture(t *testing.T, path string) xdr.LedgerCloseMeta { +func loadLedgerFixture(t testing.TB, path string) xdr.LedgerCloseMeta { t.Helper() f, err := os.Open(path) require.NoError(t, err) diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go index b721f0996..9ec8afa97 100644 --- a/internal/services/ingest_test.go +++ b/internal/services/ingest_test.go @@ -952,12 +952,12 @@ func Test_ingestService_flushBatchBufferWithRetry(t *testing.T) { sc1 := createTestStateChange(1, testAddr1, 200) sc2 := createTestStateChange(2, testAddr2, 201) - buf.PushTransaction(testAddr1, tx1) - buf.PushTransaction(testAddr2, tx2) - buf.PushOperation(testAddr1, op1, tx1) - buf.PushOperation(testAddr2, op2, tx2) - buf.PushStateChange(tx1, op1, sc1) - buf.PushStateChange(tx2, op2, sc2) + buf.PushTransaction(testAddr1, &tx1) + buf.PushTransaction(testAddr2, &tx2) + buf.PushOperation(testAddr1, &op1, &tx1) + buf.PushOperation(testAddr2, &op2, &tx2) + buf.PushStateChange(&tx1, &op1, sc1) + buf.PushStateChange(&tx2, &op2, sc2) return buf }, updateCursorTo: nil, @@ -973,7 +973,7 @@ func Test_ingestService_flushBatchBufferWithRetry(t *testing.T) { setupBuffer: func() *indexer.IndexerBuffer { buf := indexer.NewIndexerBuffer() tx1 := createTestTransaction(flushTxHash3, 3) - buf.PushTransaction(testAddr1, tx1) + buf.PushTransaction(testAddr1, &tx1) return buf }, updateCursorTo: ptrUint32(50), @@ -989,7 +989,7 @@ func Test_ingestService_flushBatchBufferWithRetry(t *testing.T) { setupBuffer: func() *indexer.IndexerBuffer { buf := indexer.NewIndexerBuffer() tx1 := createTestTransaction(flushTxHash4, 4) - buf.PushTransaction(testAddr1, tx1) + buf.PushTransaction(testAddr1, &tx1) return buf }, updateCursorTo: ptrUint32(150),