Skip to content

perf(indexer): single-owner lock-free IndexerBuffer + per-tx result folding #648

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aditya1702 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

perf(indexer): single-owner lock-free IndexerBuffer + per-tx result folding #648

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0c35199
perf(indexer): make IndexerBuffer single-owner and lock-free
aditya1702 Jul 1, 2026
d3bf854
perf(indexer): fold per-tx result bundles instead of per-worker buffers
aditya1702 Jul 1, 2026
5bd0463
test(indexer): benchmark against real testdata ledgers, not synthetic
aditya1702 Jul 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 74 additions & 95 deletions internal/indexer/indexer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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()
}

Expand Down Expand Up @@ -126,30 +116,29 @@ 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{}

for idx, tx := range 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
})
}

Expand All @@ -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]()

@aristidesstaffieri aristidesstaffieri Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

allParticipants uses the thread-safe set.NewSet to compute the ParticipantCount metric, taking a lock on every Add/Union for a local that never escapes this goroutine. (This allocation predates the PR, but since it now routes the count through result.ParticipantCount and rewrites this block, flagging while it's in view.) It's only forced thread-safe because deckarep's Union type-asserts its argument to the receiver's variant — a plain map[string]struct{}, or summing the source sets' cardinalities, drops the per-tx locking.

allParticipants = allParticipants.Union(txParticipants)
for _, opParticipants := range opsParticipants {
Expand All @@ -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),

@aristidesstaffieri aristidesstaffieri Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ProtocolWasmBytecodes and ContractEvents are allocated for every TransactionResult, but almost no transaction produces either — 2N throwaway maps per ledger on the hot path this PR is optimizing. Leaving them nil and make()-ing lazily on first insert avoids the allocations; the range loops over them in IngestTransactionResult are already no-ops over nil maps.

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
Expand All @@ -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
Expand All @@ -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
Expand Down
Loading
Loading