perf(indexer): extract contract events directly from ledger meta (#639)

* perf(indexer): extract contract events from meta directly, skipping envelope hashing

ExtractContractEventsForLedger reads contract events straight from the decoded
LedgerCloseMeta instead of building a LedgerTransactionReader (which re-hashes
every transaction envelope). Byte-for-byte equivalent to the reader path; a
differential test on real pubnet fixtures is the merge gate.

Sliced as-is from PR #622 for independent review.

* test(indexer): broaden ExtractContractEventsForLedger equivalence corpus

Validate the meta-only contract-event walk against the reader across older
history, and drop the now-unused params:

- Add wrapper-V1 (Protocol 20-22) fixtures carrying real Soroban events; the
  prior corpus was entirely wrapper-V2. The pubnet data lake re-encodes all
  history to TransactionMetaV4, so the LedgerCloseMeta wrapper version (which
  CountTransactions/TransactionResultPair/TxApplyProcessing switch on) is the
  real "older history" axis, not the meta version.
- Slim the wrapper-V2 corpus to one fixture per distinct transaction shape.
- Add synthetic tests for the V3 SorobanMeta read, fee-bump inner-result
  unwrap, and footprint-op skip (shapes the lake cannot provide as real data).
- Add a guard asserting the corpus keeps both wrapper versions with events.
- Drop the unused ctx/networkPassphrase params from ExtractContractEventsForLedger.

Author: aditya1702

internal/indexer/indexer.go

@@ -370,39 +370,48 @@ func GetLedgerTransactions(ctx context.Context, networkPassphrase string, ledger
 	return transactions, nil
 }
 
-// ExtractContractEventsForLedger walks a ledger's transactions and returns
-// the (txIdx, opIdx) → []ContractEvent map that the full indexer pipeline
-// would have pushed into the buffer. It performs no participant tracking,
-// no state-change processing, and no DB I/O — it's the minimal subset that
-// protocol-migrate needs to drive ProtocolProcessor.ProcessLedger without
-// running the full indexer.
+// ExtractContractEventsForLedger walks a ledger's transactions directly from the
+// decoded LedgerCloseMeta and returns the (txIdx, opIdx) → []ContractEvent map
+// that the full indexer pipeline would have pushed into the buffer. For each
+// transaction index i it reads the result pair, filters operations by their
+// result Tr type, and reads events from TxApplyProcessing(i) — without building
+// a LedgerTransactionReader, which would re-hash every transaction envelope just
+// to pair envelopes with metas we never read here. It is therefore a pure
+// function of the decoded ledger and needs neither a context nor the network
+// passphrase.
+//
+// The output is identical to the reader-based path; that equivalence is the
+// merge gate (see extractContractEventsViaReader and
+// TestExtractContractEventsForLedger_EquivalenceOnRealLedgers).
 //
 // Only events from successful transactions are returned, matching the live
 // indexer's filter in processTransaction.
-func ExtractContractEventsForLedger(ctx context.Context, networkPassphrase string, ledgerMeta xdr.LedgerCloseMeta) (map[ContractEventKey][]xdr.ContractEvent, error) {
-	transactions, err := GetLedgerTransactions(ctx, networkPassphrase, ledgerMeta)
-	if err != nil {
-		return nil, fmt.Errorf("getting transactions for ledger %d: %w", ledgerMeta.LedgerSequence(), err)
-	}
-
+func ExtractContractEventsForLedger(ledgerMeta xdr.LedgerCloseMeta) (map[ContractEventKey][]xdr.ContractEvent, error) {
 	out := make(map[ContractEventKey][]xdr.ContractEvent)
-	for _, tx := range transactions {
-		if !tx.Result.Successful() {
+	for i := 0; i < ledgerMeta.CountTransactions(); i++ {
+		result := ledgerMeta.TransactionResultPair(i).Result
+		if !result.Successful() {
+			continue
+		}
+		opResults, ok := result.OperationResults()
+		if !ok {
 			continue
 		}
-		for opIdx, op := range tx.Envelope.Operations() {
-			if op.Body.Type != xdr.OperationTypeInvokeHostFunction {
+		meta := ledgerMeta.TxApplyProcessing(i)
+		for opIdx, opr := range opResults {
+			tr, trOK := opr.GetTr()
+			if !trOK || tr.Type != xdr.OperationTypeInvokeHostFunction {
 				continue
 			}
-			events, evErr := tx.GetContractEventsForOperation(uint32(opIdx))
+			events, evErr := meta.GetContractEventsForOperation(uint32(opIdx))
 			if evErr != nil {
 				return nil, fmt.Errorf("extracting contract events for ledger %d tx %d op %d: %w",
-					ledgerMeta.LedgerSequence(), tx.Index, opIdx, evErr)
+					ledgerMeta.LedgerSequence(), i+1, opIdx, evErr)
 			}
 			if len(events) == 0 {
 				continue
 			}
-			out[ContractEventKey{TxIdx: tx.Index, OpIdx: uint32(opIdx)}] = events
+			out[ContractEventKey{TxIdx: uint32(i + 1), OpIdx: uint32(opIdx)}] = events
 		}
 	}
 	return out, nil

internal/indexer/indexer_test.go

@@ -2,8 +2,13 @@
 package indexer
 
 import (
+	"compress/gzip"
 	"context"
 	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
 	"runtime"
 	"testing"
 
@@ -975,3 +980,287 @@ func TestIndexer_GetLedgerTransactions(t *testing.T) {
 		})
 	}
 }
+
+// extractContractEventsViaReader is the reader-based reference implementation of
+// ExtractContractEventsForLedger, kept as the oracle for the differential
+// equivalence test. It constructs a LedgerTransactionReader (which re-hashes
+// every transaction envelope) and reads events through the resulting
+// LedgerTransaction values. The production function must produce output equal to
+// this oracle for every committed ledger fixture.
+func extractContractEventsViaReader(ctx context.Context, networkPassphrase string, ledgerMeta xdr.LedgerCloseMeta) (map[ContractEventKey][]xdr.ContractEvent, error) {
+	transactions, err := GetLedgerTransactions(ctx, networkPassphrase, ledgerMeta)
+	if err != nil {
+		return nil, fmt.Errorf("getting transactions for ledger %d: %w", ledgerMeta.LedgerSequence(), err)
+	}
+
+	out := make(map[ContractEventKey][]xdr.ContractEvent)
+	for _, tx := range transactions {
+		if !tx.Result.Successful() {
+			continue
+		}
+		for opIdx, op := range tx.Envelope.Operations() {
+			if op.Body.Type != xdr.OperationTypeInvokeHostFunction {
+				continue
+			}
+			events, evErr := tx.GetContractEventsForOperation(uint32(opIdx))
+			if evErr != nil {
+				return nil, fmt.Errorf("extracting contract events for ledger %d tx %d op %d: %w",
+					ledgerMeta.LedgerSequence(), tx.Index, opIdx, evErr)
+			}
+			if len(events) == 0 {
+				continue
+			}
+			out[ContractEventKey{TxIdx: tx.Index, OpIdx: uint32(opIdx)}] = events
+		}
+	}
+	return out, nil
+}
+
+// loadLedgerFixture reads a gzip-compressed XDR LedgerCloseMeta from testdata/.
+//
+// Fixtures are pubnet ledgers from the public data lake (bucket
+// aws-public-blockchain/v1.1/stellar/ledgers/pubnet, us-east-2). The lake re-encodes
+// all history to TransactionMetaV4, so every fixture — like everything production
+// reads — is V4, even ledgers that closed under older protocols. What does vary is
+// the LedgerCloseMeta wrapper version (V1 for Protocol 20-22, V2 for 23+), which the
+// walk switches on, so the corpus keeps both, with a mix of tx shapes (plain and
+// fee-bumped invocations, failed txs, multi-event ledgers). The V3 meta branch isn't
+// in the lake and is covered by synthetic tests.
+//
+// To add fixtures: build ingest.NewLedgerBackend (datastore + PublicNetworkPassphrase),
+// then per sequence call GetLedger, MarshalBinary, gzip, and write
+// testdata/ledger-<seq>.xdr.gz. Keep both wrapper versions represented.
+func loadLedgerFixture(t *testing.T, path string) xdr.LedgerCloseMeta {
+	t.Helper()
+	f, err := os.Open(path)
+	require.NoError(t, err)
+	defer f.Close()
+
+	gz, err := gzip.NewReader(f)
+	require.NoError(t, err)
+	defer gz.Close()
+
+	raw, err := io.ReadAll(gz)
+	require.NoError(t, err)
+
+	var lcm xdr.LedgerCloseMeta
+	require.NoError(t, lcm.UnmarshalBinary(raw))
+	return lcm
+}
+
+func TestExtractContractEventsForLedger_EquivalenceOnRealLedgers(t *testing.T) {
+	ctx := context.Background()
+
+	paths, err := filepath.Glob("testdata/*.xdr.gz")
+	require.NoError(t, err)
+	require.NotEmpty(t, paths, "no ledger fixtures under testdata/ — regenerate per the loadLedgerFixture recipe")
+
+	for _, path := range paths {
+		t.Run(filepath.Base(path), func(t *testing.T) {
+			lcm := loadLedgerFixture(t, path)
+
+			want, err := extractContractEventsViaReader(ctx, network.PublicNetworkPassphrase, lcm)
+			require.NoError(t, err)
+
+			got, err := ExtractContractEventsForLedger(lcm)
+			require.NoError(t, err)
+
+			require.Equal(t, want, got)
+		})
+	}
+}
+
+// newSyntheticLedgerCloseMeta builds a minimal single-transaction V0
+// LedgerCloseMeta carrying a successful result with the given operation results
+// and apply meta. It omits the TxSet/envelopes: the meta-only extractor never
+// reads them, so this isolates op-result + apply-meta behavior that real
+// fixtures can't produce on demand.
+func newSyntheticLedgerCloseMeta(seq uint32, opResults []xdr.OperationResult, applyMeta xdr.TransactionMeta) xdr.LedgerCloseMeta {
+	results := opResults
+	return newSyntheticLedgerCloseMetaWithResult(seq, xdr.TransactionResult{
+		Result: xdr.TransactionResultResult{
+			Code:    xdr.TransactionResultCodeTxSuccess,
+			Results: &results,
+		},
+	}, applyMeta)
+}
+
+// newSyntheticLedgerCloseMetaWithResult is newSyntheticLedgerCloseMeta with a
+// full TransactionResult, for modelling fee-bump (inner-result) shapes.
+func newSyntheticLedgerCloseMetaWithResult(seq uint32, result xdr.TransactionResult, applyMeta xdr.TransactionMeta) xdr.LedgerCloseMeta {
+	return xdr.LedgerCloseMeta{
+		V: 0,
+		V0: &xdr.LedgerCloseMetaV0{
+			LedgerHeader: xdr.LedgerHeaderHistoryEntry{
+				Header: xdr.LedgerHeader{LedgerSeq: xdr.Uint32(seq)},
+			},
+			TxProcessing: []xdr.TransactionResultMeta{
+				{
+					Result:            xdr.TransactionResultPair{Result: result},
+					TxApplyProcessing: applyMeta,
+				},
+			},
+		},
+	}
+}
+
+// TestExtractContractEventsForLedger_V4OperationsShorterThanResults proves the
+// result-Tr.Type filter prevents indexing past the end of TransactionMetaV4.Operations.
+// The V4 apply meta has ONE operation meta entry (index 0), but the tx has TWO
+// operation results: [0] InvokeHostFunction, [1] Payment. A walk that indexed
+// Operations by every result index would panic on Operations[1]; the filter
+// skips the non-InvokeHostFunction result before it is ever used to index.
+func TestExtractContractEventsForLedger_V4OperationsShorterThanResults(t *testing.T) {
+	applyMeta := xdr.TransactionMeta{
+		V: 4,
+		V4: &xdr.TransactionMetaV4{
+			Operations: []xdr.OperationMetaV2{
+				{Events: []xdr.ContractEvent{{Type: xdr.ContractEventTypeContract}}}, // index 0 only
+			},
+		},
+	}
+	opResults := []xdr.OperationResult{
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypeInvokeHostFunction}},
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypePayment}},
+	}
+
+	lcm := newSyntheticLedgerCloseMeta(100, opResults, applyMeta)
+
+	out, err := ExtractContractEventsForLedger(lcm)
+	require.NoError(t, err)
+	require.Len(t, out, 1)
+	events, ok := out[ContractEventKey{TxIdx: 1, OpIdx: 0}]
+	require.True(t, ok, "expected events at tx 1 op 0")
+	require.Len(t, events, 1)
+}
+
+// TestExtractContractEventsForLedger_UnknownMetaVersionErrors confirms the one
+// intentional behavior change: an unsupported TransactionMeta version surfaces
+// as a propagated error (fail loud) rather than being silently dropped.
+func TestExtractContractEventsForLedger_UnknownMetaVersionErrors(t *testing.T) {
+	opResults := []xdr.OperationResult{
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypeInvokeHostFunction}},
+	}
+	applyMeta := xdr.TransactionMeta{V: 99} // unsupported; all arm pointers nil
+
+	lcm := newSyntheticLedgerCloseMeta(101, opResults, applyMeta)
+
+	_, err := ExtractContractEventsForLedger(lcm)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "unsupported TransactionMeta version")
+}
+
+// In V3 meta, Soroban events live at the transaction level (SorobanMeta.Events)
+// and the op index is ignored. This checks the walk surfaces them at op 0.
+//
+// It's our only V3 coverage: the data lake re-encodes all history to V4, so a real
+// V3 fixture can't be fetched.
+func TestExtractContractEventsForLedger_V3SorobanMetaEvents(t *testing.T) {
+	ev1 := xdr.ContractEvent{Type: xdr.ContractEventTypeContract}
+	ev2 := xdr.ContractEvent{Type: xdr.ContractEventTypeSystem}
+	applyMeta := xdr.TransactionMeta{
+		V: 3,
+		V3: &xdr.TransactionMetaV3{
+			SorobanMeta: &xdr.SorobanTransactionMeta{Events: []xdr.ContractEvent{ev1, ev2}},
+		},
+	}
+	opResults := []xdr.OperationResult{
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypeInvokeHostFunction}},
+	}
+	lcm := newSyntheticLedgerCloseMeta(103, opResults, applyMeta)
+
+	out, err := ExtractContractEventsForLedger(lcm)
+	require.NoError(t, err)
+	require.Len(t, out, 1)
+	got, ok := out[ContractEventKey{TxIdx: 1, OpIdx: 0}]
+	require.True(t, ok, "expected events at tx 1 op 0")
+	require.Equal(t, []xdr.ContractEvent{ev1, ev2}, got)
+}
+
+// A fee-bump's operation results live in its inner transaction. This checks the
+// walk unwraps to those inner results instead of reading the empty outer result.
+// Real fee-bumped invocations are also in the fixture corpus.
+func TestExtractContractEventsForLedger_FeeBumpUnwrapsInnerResults(t *testing.T) {
+	ev := xdr.ContractEvent{Type: xdr.ContractEventTypeContract}
+	applyMeta := xdr.TransactionMeta{
+		V: 3,
+		V3: &xdr.TransactionMetaV3{
+			SorobanMeta: &xdr.SorobanTransactionMeta{Events: []xdr.ContractEvent{ev}},
+		},
+	}
+	innerOpResults := []xdr.OperationResult{
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypeInvokeHostFunction}},
+	}
+	feeBump := xdr.TransactionResult{
+		Result: xdr.TransactionResultResult{
+			Code: xdr.TransactionResultCodeTxFeeBumpInnerSuccess,
+			InnerResultPair: &xdr.InnerTransactionResultPair{
+				Result: xdr.InnerTransactionResult{
+					Result: xdr.InnerTransactionResultResult{
+						Code:    xdr.TransactionResultCodeTxSuccess,
+						Results: &innerOpResults,
+					},
+				},
+			},
+		},
+	}
+	lcm := newSyntheticLedgerCloseMetaWithResult(104, feeBump, applyMeta)
+
+	out, err := ExtractContractEventsForLedger(lcm)
+	require.NoError(t, err)
+	require.Len(t, out, 1, "fee-bump inner op results must be unwrapped")
+	got, ok := out[ContractEventKey{TxIdx: 1, OpIdx: 0}]
+	require.True(t, ok)
+	require.Equal(t, []xdr.ContractEvent{ev}, got)
+}
+
+// RestoreFootprint and ExtendFootprintTtl ops carry no contract events, so the
+// walk should skip them and return events only for the InvokeHostFunction op.
+// Done synthetically since these ops weren't found in the scanned pubnet ledgers.
+func TestExtractContractEventsForLedger_FootprintOpsSkipped(t *testing.T) {
+	applyMeta := xdr.TransactionMeta{
+		V: 4,
+		V4: &xdr.TransactionMetaV4{
+			Operations: []xdr.OperationMetaV2{
+				{Events: []xdr.ContractEvent{{Type: xdr.ContractEventTypeContract}}}, // index 0 (invoke)
+			},
+		},
+	}
+	opResults := []xdr.OperationResult{
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypeInvokeHostFunction}},
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypeExtendFootprintTtl}},
+		{Code: xdr.OperationResultCodeOpInner, Tr: &xdr.OperationResultTr{Type: xdr.OperationTypeRestoreFootprint}},
+	}
+	lcm := newSyntheticLedgerCloseMeta(105, opResults, applyMeta)
+
+	out, err := ExtractContractEventsForLedger(lcm)
+	require.NoError(t, err)
+	require.Len(t, out, 1, "only the InvokeHostFunction op yields events; footprint ops are skipped")
+	_, ok := out[ContractEventKey{TxIdx: 1, OpIdx: 0}]
+	require.True(t, ok)
+}
+
+// Guards that the corpus keeps real fixtures for both LedgerCloseMeta wrapper
+// versions the lake serves with events: V1 (Protocol 20-22) and V2 (Protocol 23+).
+// The walk switches on this version, so dropping either set would quietly lose
+// coverage of an older history window.
+//
+// All lake fixtures are TransactionMetaV4; the V3 branch is covered by the
+// synthetic tests above.
+func TestExtractContractEventsForLedger_CorpusCoversWrapperVersions(t *testing.T) {
+	paths, err := filepath.Glob("testdata/*.xdr.gz")
+	require.NoError(t, err)
+	require.NotEmpty(t, paths)
+
+	wrapperWithEvents := map[int32]bool{} // LedgerCloseMeta wrapper version -> a ledger of it yielded events
+	for _, path := range paths {
+		lcm := loadLedgerFixture(t, path)
+		events, evErr := ExtractContractEventsForLedger(lcm)
+		require.NoError(t, evErr)
+		if len(events) > 0 {
+			wrapperWithEvents[lcm.V] = true
+		}
+	}
+	require.True(t, wrapperWithEvents[1], "corpus must include a wrapper-V1 (Protocol 20-22) ledger with extracted contract events")
+	require.True(t, wrapperWithEvents[2], "corpus must include a wrapper-V2 (Protocol 23+) ledger with extracted contract events")
+}

internal/indexer/testdata/.gitattributes

@@ -0,0 +1 @@
+*.xdr.gz binary

internal/indexer/testdata/ledger-51000294.xdr.gz

@@ -261,7 +261,7 @@ func (s *protocolMigrateEngine) processAllProtocols(ctx context.Context, protoco
 		// Extract contract events once per ledger; all trackers below share the
 		// same map. This is the migration-side analogue of the live-ingest path
 		// where buffer.GetContractEvents() is computed once per ledger.
-		ledgerEvents, eventsErr := indexer.ExtractContractEventsForLedger(ctx, s.networkPassphrase, ledgerMeta)
+		ledgerEvents, eventsErr := indexer.ExtractContractEventsForLedger(ledgerMeta)
 		if eventsErr != nil {
 			return handedOffProtocolIDs(trackers), fmt.Errorf("extracting contract events for ledger %d: %w", seq, eventsErr)
 		}