perf(indexer): single-owner lock-free IndexerBuffer + per-tx result folding#648
perf(indexer): single-owner lock-free IndexerBuffer + per-tx result folding#648aditya1702 wants to merge 3 commits into
Conversation
Each IndexerBuffer is owned by a single goroutine in production (one per tx worker, merged serially after group.Wait()), so the RWMutex protected nothing. Remove it, and swap the thread-safe deckarep set (NewSet -> NewThreadUnsafeSet) whose internal mutex was pure overhead on every participant Add. Also drop the 7 defensive maps.Clone() in the Get* methods — callers only iterate the result and the buffer is Clear()ed after each flush — matching the getters that already returned the internal map directly. Concurrent buffer tests are rewritten as single-owner tests, and the stale "large XDR fields" / thread-safety godocs are corrected.
Each ledger transaction is now processed into a lightweight TransactionResult
by a parallel worker; the results are folded into a single ledger buffer
serially (IngestTransactionResult). This replaces allocating a full
IndexerBuffer per transaction (~15 maps each) plus the buffer-to-buffer Merge
pass — Merge and mergeWithTombstone are removed entirely.
Push{Transaction,Operation,StateChange} now take pointers, so the parallel
phase builds results with zero struct copies; the serial fold reuses the
existing lock-free Push* dedup, leaving highest-order-wins/tombstone semantics
unchanged. Also adds buffer benchmarks (PushTransaction is now 0 allocs/op).
|
Codex usage limits have been reached for code reviews. Please check with the admins of this repo to increase the limits by adding credits. |
There was a problem hiding this comment.
Pull request overview
This PR optimizes the internal/indexer ingestion hot path by making IndexerBuffer single-owner/lock-free and replacing per-transaction buffers + merge with a per-tx TransactionResult that is folded serially into one ledger buffer.
Changes:
- Refactors transaction processing to build per-tx
TransactionResultin parallel workers and fold results into the ledger buffer viaIngestTransactionResult. - Removes
IndexerBufferlocking/defensive cloning, switches participant sets to thread-unsafe variants, and drops buffer-to-buffer merge logic. - Updates tests to use pointer-based
Push*APIs and adds benchmarks for folding/push paths.
Reviewed changes
Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| internal/services/ingest_test.go | Updates ingestion tests to pass pointers into Push* APIs. |
| internal/indexer/indexer.go | Builds per-tx TransactionResult in workers and folds results serially into the ledger buffer. |
| internal/indexer/indexer_buffer.go | Removes locks/clones/merge; introduces TransactionResult + IngestTransactionResult; updates Push* to accept pointers. |
| internal/indexer/indexer_buffer_test.go | Adjusts tests for pointer APIs and result folding; removes merge/concurrency tests. |
| internal/indexer/indexer_buffer_bench_test.go | Adds benchmarks focused on result folding and dedup paths. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| for opID, participants := range r.OpParticipants { | ||
| operation := r.Operations[opID] | ||
| for _, participant := range participants { | ||
| b.PushOperation(participant, operation, r.Transaction) | ||
| } | ||
| } |
Replace the hand-built TransactionResult fixtures with benchmarks driven by the real pubnet ledger fixtures under testdata/ (the same corpus the equivalence test uses). BenchmarkFoldRealLedger isolates the serial fold; BenchmarkProcessRealLedger measures the full per-ledger pass. loadLedgerFixture now takes testing.TB so both tests and benchmarks can share it.
| TxParticipants: txParticipants.ToSlice(), | ||
| Operations: make(map[int64]*types.Operation, len(opsParticipants)), | ||
| OpParticipants: make(map[int64][]string, len(opsParticipants)), | ||
| ProtocolWasmBytecodes: make(map[string][]byte), |
There was a problem hiding this comment.
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.
| @@ -68,7 +62,6 @@ type ContractEventKey struct { | |||
| } | |||
|
|
|||
| type IndexerBuffer struct { | |||
There was a problem hiding this comment.
Now that the RWMutex is gone and correctness rests on the single-owner "serial fold after group.Wait()" convention, consider adding a test that runs ProcessLedgerTransactions on a multi-tx ledger under the race detector (-race). The convention is correct today, but it's enforced only by a comment — a race test would catch any future change that accidentally folds or persists a buffer from a background goroutine at review time, instead of it surfacing as a fatal error: concurrent map writes in production.
| // 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]() |
There was a problem hiding this comment.
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.
What
Two changes to
internal/indexerbuffering (one commit each):IndexerBuffer— drop theRWMutex, switch participant sets to the thread-unsafe variant, and remove the 7 defensivemaps.Clone()getters.TransactionResult; results are folded into one ledger buffer serially viaIngestTransactionResult.Merge/mergeWithTombstoneare removed.Why
Each
IndexerBufferis only ever touched by a single goroutine in production — the indexer creates one per tx worker and merges them serially after the pool finishes. So the locks, thread-safe sets, and clones were guarding a race that cannot happen. Separately, allocating N full buffers per ledger (~15 maps each) and then merging them meant extra allocations plus a redundant second dedup pass on the hot ingestion path.Benchmarks
Run against the real pubnet ledger fixtures under
internal/indexer/testdata/—go test -bench=RealLedger -benchmem ./internal/indexer(Apple M2 Pro).FoldRealLedgerisolates the serial fold this PR touches (per-tx results pre-built, untimed);ProcessRealLedgeris the full per-ledger indexing pass, shown for context.51000294(largest)6309298857500008The fold is ~10–15% of a full ledger's indexing cost (the rest is XDR decode + the per-op processors), and it no longer allocates a buffer per transaction or runs a second merge pass.
Notes for reviewers
Push*dedup.Push{Transaction,Operation,StateChange}now take pointers, so the parallel phase builds results with zero struct copies.make checkand the full-raceunit suite pass.Closes #647