Skip to content

Optimize indexer buffer design #558

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 19 commits into
base: remove-optimized-catchup
Choose a base branch
from
Open

Optimize indexer buffer design #558

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
19 commits
Select commit Hold shift + click to select a range
ee2e4ce
Eliminate duplicate GetTransactions() allocation per ledger
aditya1702 Mar 31, 2026
166f022
Remove unnecessary maps.Clone() from IndexerBuffer getters
aditya1702 Mar 31, 2026
86f0758
Reuse IndexerBuffer across ledgers in live ingestion
aditya1702 Mar 31, 2026
b32f050
Add BatchPushChanges to reduce lock acquisitions per operation
aditya1702 Mar 31, 2026
7551fc6
Bound ledger indexer pool to NumCPU
aditya1702 Mar 31, 2026
6287754
Use sync.Pool for per-transaction IndexerBuffer reuse
aditya1702 Mar 31, 2026
956ae32
Remove unused PushAccountChange and PushSACContract public methods
aditya1702 Mar 31, 2026
cb3f0fd
Remove unused PushSACBalanceChange public method
aditya1702 Mar 31, 2026
9007c34
Remove sync.RWMutex from IndexerBuffer and collapse unsafe methods
aditya1702 Mar 31, 2026
1518463
Change Push method signatures to accept pointers instead of values
aditya1702 Mar 31, 2026
5ba31b7
Replace golang-set/v2 with plain ParticipantSet for participant tracking
aditya1702 Mar 31, 2026
899373c
Optimize Merge() to steal sets and fix Union allocation in processTra…
aditya1702 Mar 31, 2026
a137638
Add benchmarks for IndexerBuffer operations
aditya1702 Mar 31, 2026
43f053e
Update indexer_buffer_test.go
aditya1702 Mar 31, 2026
420f259
Eliminate per-transaction buffers: use single shared buffer with mutex
aditya1702 Mar 31, 2026
f5be129
make check
aditya1702 Mar 31, 2026
08345dd
Pre-allocate benchmark data to isolate buffer performance
aditya1702 Mar 31, 2026
d3b6c89
Update indexer_buffer_bench_test.go
aditya1702 Mar 31, 2026
743ae26
Batch transaction buffer writes to reduce locks
aditya1702 Apr 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
3 changes: 1 addition & 2 deletions internal/data/operations.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import (
"strings"
"time"

set "github.com/deckarep/golang-set/v2"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/jackc/pgx/v5/pgxpool"
Expand Down Expand Up @@ -334,7 +333,7 @@ func (m *OperationModel) BatchCopy(
ctx context.Context,
pgxTx pgx.Tx,
operations []*types.Operation,
stellarAddressesByOpID map[int64]set.Set[string],
stellarAddressesByOpID map[int64]types.ParticipantSet,
) (int, error) {
if len(operations) == 0 {
return 0, nil
Expand Down
15 changes: 7 additions & 8 deletions internal/data/operations_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import (
"testing"
"time"

set "github.com/deckarep/golang-set/v2"
"github.com/jackc/pgx/v5"
"github.com/prometheus/client_golang/prometheus"
"github.com/stellar/go-stellar-sdk/keypair"
Expand All @@ -21,9 +20,9 @@ import (

// generateTestOperations creates n test operations for benchmarking.
// It returns a map of operation IDs to addresses.
func generateTestOperations(n int, startID int64) ([]*types.Operation, map[int64]set.Set[string]) {
func generateTestOperations(n int, startID int64) ([]*types.Operation, map[int64]types.ParticipantSet) {
ops := make([]*types.Operation, n)
addressesByOpID := make(map[int64]set.Set[string])
addressesByOpID := make(map[int64]types.ParticipantSet)
now := time.Now()

for i := 0; i < n; i++ {
Expand All @@ -37,7 +36,7 @@ func generateTestOperations(n int, startID int64) ([]*types.Operation, map[int64
LedgerNumber: uint32(i + 1),
LedgerCreatedAt: now,
}
addressesByOpID[opID] = set.NewSet(address)
addressesByOpID[opID] = types.NewParticipantSet(address)
}

return ops, addressesByOpID
Expand Down Expand Up @@ -101,26 +100,26 @@ func Test_OperationModel_BatchCopy(t *testing.T) {
testCases := []struct {
name string
operations []*types.Operation
stellarAddressesByOpID map[int64]set.Set[string]
stellarAddressesByOpID map[int64]types.ParticipantSet
wantCount int
wantErrContains string
}{
{
name: "🟢successful_insert_multiple",
operations: []*types.Operation{&op1, &op2},
stellarAddressesByOpID: map[int64]set.Set[string]{op1.ID: set.NewSet(kp1.Address()), op2.ID: set.NewSet(kp2.Address())},
stellarAddressesByOpID: map[int64]types.ParticipantSet{op1.ID: types.NewParticipantSet(kp1.Address()), op2.ID: types.NewParticipantSet(kp2.Address())},
wantCount: 2,
},
{
name: "🟢empty_input",
operations: []*types.Operation{},
stellarAddressesByOpID: map[int64]set.Set[string]{},
stellarAddressesByOpID: map[int64]types.ParticipantSet{},
wantCount: 0,
},
{
name: "🟢no_participants",
operations: []*types.Operation{&op1},
stellarAddressesByOpID: map[int64]set.Set[string]{},
stellarAddressesByOpID: map[int64]types.ParticipantSet{},
wantCount: 1,
},
}
Expand Down
3 changes: 1 addition & 2 deletions internal/data/transactions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import (
"strings"
"time"

set "github.com/deckarep/golang-set/v2"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
"github.com/jackc/pgx/v5/pgxpool"
Expand Down Expand Up @@ -245,7 +244,7 @@ func (m *TransactionModel) BatchCopy(
ctx context.Context,
pgxTx pgx.Tx,
txs []*types.Transaction,
stellarAddressesByToID map[int64]set.Set[string],
stellarAddressesByToID map[int64]types.ParticipantSet,
) (int, error) {
if len(txs) == 0 {
return 0, nil
Expand Down
17 changes: 8 additions & 9 deletions internal/data/transactions_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@ import (
"testing"
"time"

set "github.com/deckarep/golang-set/v2"
"github.com/jackc/pgx/v5"
"github.com/prometheus/client_golang/prometheus"
"github.com/stellar/go-stellar-sdk/keypair"
Expand All @@ -22,9 +21,9 @@ import (

// generateTestTransactions creates n test transactions for benchmarking.
// Uses toid.New to generate realistic ToIDs based on ledger sequence and transaction index.
func generateTestTransactions(n int, startLedger int32) ([]*types.Transaction, map[int64]set.Set[string]) {
func generateTestTransactions(n int, startLedger int32) ([]*types.Transaction, map[int64]types.ParticipantSet) {
txs := make([]*types.Transaction, n)
addressesByToID := make(map[int64]set.Set[string])
addressesByToID := make(map[int64]types.ParticipantSet)
now := time.Now()

for i := 0; i < n; i++ {
Expand All @@ -43,7 +42,7 @@ func generateTestTransactions(n int, startLedger int32) ([]*types.Transaction, m
LedgerCreatedAt: now,
IsFeeBump: false,
}
addressesByToID[toID] = set.NewSet(address)
addressesByToID[toID] = types.NewParticipantSet(address)
}

return txs, addressesByToID
Expand Down Expand Up @@ -93,32 +92,32 @@ func Test_TransactionModel_BatchCopy(t *testing.T) {
testCases := []struct {
name string
txs []*types.Transaction
stellarAddressesByToID map[int64]set.Set[string]
stellarAddressesByToID map[int64]types.ParticipantSet
wantCount int
wantErrContains string
}{
{
name: "🟢successful_insert_multiple",
txs: []*types.Transaction{&txCopy1, &txCopy2},
stellarAddressesByToID: map[int64]set.Set[string]{txCopy1.ToID: set.NewSet(kp1.Address()), txCopy2.ToID: set.NewSet(kp2.Address())},
stellarAddressesByToID: map[int64]types.ParticipantSet{txCopy1.ToID: types.NewParticipantSet(kp1.Address()), txCopy2.ToID: types.NewParticipantSet(kp2.Address())},
wantCount: 2,
},
{
name: "🟢empty_input",
txs: []*types.Transaction{},
stellarAddressesByToID: map[int64]set.Set[string]{},
stellarAddressesByToID: map[int64]types.ParticipantSet{},
wantCount: 0,
},
{
name: "🟢single_transaction",
txs: []*types.Transaction{&txCopy3},
stellarAddressesByToID: map[int64]set.Set[string]{txCopy3.ToID: set.NewSet(kp1.Address())},
stellarAddressesByToID: map[int64]types.ParticipantSet{txCopy3.ToID: types.NewParticipantSet(kp1.Address())},
wantCount: 1,
},
{
name: "🟢no_participants",
txs: []*types.Transaction{&txCopy1},
stellarAddressesByToID: map[int64]set.Set[string]{},
stellarAddressesByToID: map[int64]types.ParticipantSet{},
wantCount: 1,
},
}
Expand Down
Loading
Loading