feat(protocols): ContractDataChanges framework seam behind a RequiresContractData capability gate

internal/indexer/indexer.go

@@ -10,6 +10,7 @@ import (
 	"github.com/alitto/pond/v2"
 	set "github.com/deckarep/golang-set/v2"
 	"github.com/stellar/go-stellar-sdk/ingest"
+	"github.com/stellar/go-stellar-sdk/strkey"
 	"github.com/stellar/go-stellar-sdk/support/log"
 	"github.com/stellar/go-stellar-sdk/xdr"
 
@@ -462,19 +463,176 @@ func ExtractContractEventsForLedger(ledgerMeta xdr.LedgerCloseMeta) (map[Contrac
 	return out, nil
 }
 
+// ExtractContractDataChangesForLedger returns every ContractData ledger-entry
+// change from a ledger's successful transactions, grouped by the owning
+// contract's C-address strkey — but only after a cheap footprint gate: if no
+// transaction's read-write footprint contains a ContractData key owned by a
+// tracked contract, the ledger is skipped outright (empty map). Soroban
+// guarantees writes ⊆ the declared read-write footprint (host storage is
+// footprint-seeded and a write outside it traps; RestoreFootprint restores
+// exactly the read-write keys; protocol-23 auto-restores are indices into it),
+// so the gate can never skip a ledger that actually changes a tracked
+// contract's entries.
+//
+// Both the gate and the extraction walk the already-decoded LedgerCloseMeta
+// directly — GetChanges reads only the transaction meta, result, and ledger
+// version, so no LedgerTransactionReader (which re-hashes every envelope just
+// to pair envelopes with metas) is ever built. When the gate passes, the FULL
+// ledger's ContractData changes are returned, not just the tracked subset:
+// the map is shared across protocols and each processor filters by its own
+// membership.
+func ExtractContractDataChangesForLedger(ledgerMeta xdr.LedgerCloseMeta, trackedContracts map[xdr.ContractId]struct{}) (map[string][]ingest.Change, error) {
+	if !ledgerTouchesTrackedContractData(ledgerMeta, trackedContracts) {
+		return map[string][]ingest.Change{}, nil
+	}
+
+	ledgerSeq := ledgerMeta.LedgerSequence()
+	out := make(map[string][]ingest.Change)
+	for i := 0; i < ledgerMeta.CountTransactions(); i++ {
+		resultPair := ledgerMeta.TransactionResultPair(i)
+		if !resultPair.Result.Successful() {
+			continue
+		}
+		// A minimal LedgerTransaction: GetChanges touches only these fields
+		// (verified by the fixture equivalence test against the reader-based
+		// reference) — the envelope is deliberately absent. Result pairs and
+		// TxApplyProcessing share application order, the same index alignment
+		// ExtractContractEventsForLedger relies on.
+		tx := ingest.LedgerTransaction{
+			Index:         uint32(i + 1),
+			Result:        resultPair,
+			UnsafeMeta:    ledgerMeta.TxApplyProcessing(i),
+			LedgerVersion: ledgerMeta.ProtocolVersion(),
+			Ledger:        ledgerMeta,
+		}
+		if err := collectContractDataChanges(&tx, ledgerSeq, out); err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+// ledgerTouchesTrackedContractData reports whether any transaction in the
+// ledger declares a read-write footprint ContractData key owned by a tracked
+// contract. It walks envelopes in whatever order the ledger stores them —
+// fine for a ledger-level boolean, which is exactly why the gate is
+// per-ledger rather than per-transaction: envelopes are in tx-set order while
+// metas are in application order, and matching the two is the expensive
+// pairing this function exists to avoid.
+func ledgerTouchesTrackedContractData(ledgerMeta xdr.LedgerCloseMeta, trackedContracts map[xdr.ContractId]struct{}) bool {
+	if len(trackedContracts) == 0 {
+		return false
+	}
+	for _, env := range ledgerMeta.TransactionEnvelopes() {
+		var ext xdr.TransactionExt
+		switch env.Type {
+		case xdr.EnvelopeTypeEnvelopeTypeTx:
+			ext = env.V1.Tx.Ext
+		case xdr.EnvelopeTypeEnvelopeTypeTxFeeBump:
+			ext = env.FeeBump.Tx.InnerTx.V1.Tx.Ext
+		default:
+			// V0 envelopes predate Soroban: no footprint, no ContractData.
+			continue
+		}
+		sorobanData, ok := ext.GetSorobanData()
+		if !ok {
+			continue
+		}
+		for _, key := range sorobanData.Resources.Footprint.ReadWrite {
+			if key.Type != xdr.LedgerEntryTypeContractData {
+				continue
+			}
+			contractID, ok := key.ContractData.Contract.GetContractId()
+			if !ok {
+				continue
+			}
+			if _, tracked := trackedContracts[contractID]; tracked {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// ExtractContractDataChangesFromTransactions extracts the same
+// contract-grouped ContractData changes as ExtractContractDataChangesForLedger
+// over already-materialized transactions, ungated — live ingestion's main
+// pipeline has the transactions in hand and processes one ledger per close,
+// so a footprint gate buys nothing there. ledgerSeq is used only for error
+// context.
+func ExtractContractDataChangesFromTransactions(transactions []ingest.LedgerTransaction, ledgerSeq uint32) (map[string][]ingest.Change, error) {
+	out := make(map[string][]ingest.Change)
+	for i := range transactions {
+		tx := transactions[i]
+		if !tx.Result.Successful() {
+			continue
+		}
+		if err := collectContractDataChanges(&tx, ledgerSeq, out); err != nil {
+			return nil, err
+		}
+	}
+	return out, nil
+}
+
+// collectContractDataChanges appends tx's ContractData changes into out,
+// grouped by the owning contract's C-address strkey.
+//
+// Within a contract, changes preserve transaction application order, so
+// last-write-wins folding per entry key is deterministic. Ledger-level
+// archival evictions are NOT surfaced (GetChanges only walks fee/tx/op meta);
+// per-tx entry removals appear with Post == nil.
+func collectContractDataChanges(tx *ingest.LedgerTransaction, ledgerSeq uint32, out map[string][]ingest.Change) error {
+	changes, chErr := tx.GetChanges()
+	if chErr != nil {
+		return fmt.Errorf("getting changes for ledger %d tx %d: %w", ledgerSeq, tx.Index, chErr)
+	}
+	for _, change := range changes {
+		if change.Type != xdr.LedgerEntryTypeContractData {
+			continue
+		}
+		entry := change.Post
+		if entry == nil {
+			entry = change.Pre
+		}
+		if entry == nil {
+			continue
+		}
+		contractData, ok := entry.Data.GetContractData()
+		if !ok {
+			continue
+		}
+		contractIDBytes, ok := contractData.Contract.GetContractId()
+		if !ok {
+			continue
+		}
+		addr, encErr := strkey.Encode(strkey.VersionByteContract, contractIDBytes[:])
+		if encErr != nil {
+			// Callers rely on this function returning every ContractData
+			// change; silently dropping one would corrupt downstream state.
+			return fmt.Errorf("encoding contract id for ledger %d tx %d: %w", ledgerSeq, tx.Index, encErr)
+		}
+		out[addr] = append(out[addr], change)
+	}
+	return nil
+}
+
 // ProcessLedger extracts transactions from a ledger and indexes them.
-// Returns the participant count for optional metrics recording.
-func ProcessLedger(ctx context.Context, networkPassphrase string, ledgerMeta xdr.LedgerCloseMeta, ledgerIndexer *Indexer, buffer *IndexerBuffer) (int, error) {
+// Returns the participant count for optional metrics recording, plus the
+// materialized transactions so callers with further per-transaction work
+// (live ingestion's ContractData extraction) can reuse them instead of
+// paying for a second LedgerTransactionReader build — its constructor
+// re-hashes every transaction envelope.
+func ProcessLedger(ctx context.Context, networkPassphrase string, ledgerMeta xdr.LedgerCloseMeta, ledgerIndexer *Indexer, buffer *IndexerBuffer) (int, []ingest.LedgerTransaction, error) {
 	ledgerSeq := ledgerMeta.LedgerSequence()
 	transactions, err := GetLedgerTransactions(ctx, networkPassphrase, ledgerMeta)
 	if err != nil {
-		return 0, fmt.Errorf("getting transactions for ledger %d: %w", ledgerSeq, err)
+		return 0, nil, fmt.Errorf("getting transactions for ledger %d: %w", ledgerSeq, err)
 	}
 
 	participantCount, err := ledgerIndexer.ProcessLedgerTransactions(ctx, transactions, buffer)
 	if err != nil {
-		return 0, fmt.Errorf("processing transactions for ledger %d: %w", ledgerSeq, err)
+		return 0, nil, fmt.Errorf("processing transactions for ledger %d: %w", ledgerSeq, err)
 	}
 
-	return participantCount, nil
+	return participantCount, transactions, nil
 }

internal/indexer/indexer_test.go

@@ -16,6 +16,7 @@ import (
 	set "github.com/deckarep/golang-set/v2"
 	"github.com/stellar/go-stellar-sdk/ingest"
 	"github.com/stellar/go-stellar-sdk/network"
+	"github.com/stellar/go-stellar-sdk/strkey"
 	"github.com/stellar/go-stellar-sdk/xdr"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -1355,3 +1356,179 @@ func TestExtractContractEventsForLedger_CorpusCoversWrapperVersions(t *testing.T
 	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")
 }
+
+// contractDataChangeProjection is the payload-relevant subset of an
+// ingest.Change used to compare the reader-based and meta-based extraction
+// paths: the two build different LedgerTransaction carriers (the meta-based
+// one has no envelope), so whole-Change equality would compare carrier
+// internals no consumer reads.
+type contractDataChangeProjection struct {
+	Type           xdr.LedgerEntryType
+	Reason         ingest.LedgerEntryChangeReason
+	OperationIndex uint32
+	Pre            *xdr.LedgerEntry
+	Post           *xdr.LedgerEntry
+}
+
+func projectContractDataChanges(m map[string][]ingest.Change) map[string][]contractDataChangeProjection {
+	out := make(map[string][]contractDataChangeProjection, len(m))
+	for k, changes := range m {
+		ps := make([]contractDataChangeProjection, 0, len(changes))
+		for _, c := range changes {
+			ps = append(ps, contractDataChangeProjection{
+				Type:           c.Type,
+				Reason:         c.Reason,
+				OperationIndex: c.OperationIndex,
+				Pre:            c.Pre,
+				Post:           c.Post,
+			})
+		}
+		out[k] = ps
+	}
+	return out
+}
+
+// extractContractDataChangesViaReader is the reader-based reference
+// implementation: materialize transactions through the
+// LedgerTransactionReader (envelope↔meta pairing via hashing) and extract
+// from those. Kept test-only as the merge gate for the production meta-based
+// path, mirroring extractContractEventsViaReader.
+func extractContractDataChangesViaReader(ctx context.Context, networkPassphrase string, ledgerMeta xdr.LedgerCloseMeta) ([]ingest.LedgerTransaction, map[string][]ingest.Change, error) {
+	transactions, err := GetLedgerTransactions(ctx, networkPassphrase, ledgerMeta)
+	if err != nil {
+		return nil, nil, err
+	}
+	changes, err := ExtractContractDataChangesFromTransactions(transactions, ledgerMeta.LedgerSequence())
+	return transactions, changes, err
+}
+
+// TestExtractContractDataChangesForLedger_Fixtures checks the ContractData
+// extraction paths against the same real-ledger fixture corpus used for
+// contract events: every returned group key must be a valid C-address, every
+// change must be a ContractData change grouped under its own owning contract,
+// every change must originate from a successful transaction, and the whole
+// corpus must yield at least one change (otherwise the corpus wouldn't
+// exercise the extractor at all).
+//
+// It also pins the two load-bearing properties of the production
+// (footprint-gated, meta-based) ExtractContractDataChangesForLedger:
+//   - Equivalence: gated extraction with every changed contract tracked
+//     returns exactly the reader-based reference output.
+//   - Gate soundness: for EVERY contract that has changes, tracking just that
+//     contract must trigger the gate — i.e. on real ledgers, a ContractData
+//     change always has its key in some transaction's read-write footprint
+//     (the Soroban writes ⊆ footprint guarantee the gate relies on).
+func TestExtractContractDataChangesForLedger_Fixtures(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")
+
+	totalChanges := 0
+	for _, path := range paths {
+		t.Run(filepath.Base(path), func(t *testing.T) {
+			lcm := loadLedgerFixture(t, path)
+
+			transactions, got, extractErr := extractContractDataChangesViaReader(ctx, network.PublicNetworkPassphrase, lcm)
+			require.NoError(t, extractErr)
+			wantProjection := projectContractDataChanges(got)
+
+			// Tracked set = every contract with changes this ledger.
+			trackedAll := map[xdr.ContractId]struct{}{}
+			for groupKey := range got {
+				raw, decodeErr := strkey.Decode(strkey.VersionByteContract, groupKey)
+				require.NoError(t, decodeErr)
+				trackedAll[xdr.ContractId(raw)] = struct{}{}
+			}
+
+			// Equivalence: the production gated+meta-based path returns the
+			// reference output when every changed contract is tracked.
+			gated, gatedErr := ExtractContractDataChangesForLedger(lcm, trackedAll)
+			require.NoError(t, gatedErr)
+			assert.Equal(t, wantProjection, projectContractDataChanges(gated),
+				"meta-based extraction must match the reader-based reference")
+
+			// Gate soundness: each changed contract, tracked alone, must
+			// trigger the gate via some transaction's read-write footprint.
+			for contractID := range trackedAll {
+				alone, aloneErr := ExtractContractDataChangesForLedger(lcm, map[xdr.ContractId]struct{}{contractID: {}})
+				require.NoError(t, aloneErr)
+				assert.Equal(t, wantProjection, projectContractDataChanges(alone),
+					"tracking contract %x alone must trigger the gate (footprint ⊇ writes)", contractID)
+			}
+
+			// Gate negative: an untracked-only set skips the ledger outright.
+			empty, emptyErr := ExtractContractDataChangesForLedger(lcm, map[xdr.ContractId]struct{}{{0xde, 0xad, 0xbe, 0xef}: {}})
+			require.NoError(t, emptyErr)
+			assert.Empty(t, empty, "a ledger touching no tracked contract must be skipped")
+			none, noneErr := ExtractContractDataChangesForLedger(lcm, nil)
+			require.NoError(t, noneErr)
+			assert.Empty(t, none, "an empty tracked set must skip every ledger")
+
+			// Independently derive the expected ContractData change count from
+			// the successful transactions, without calling the function under
+			// test, so the "only successful txs" assertion is meaningful rather
+			// than tautological. A handful of historical (pre-CAP-0046-11)
+			// ContractData entries are keyed by a classic account address rather
+			// than a contract address; those have no owning contract to group
+			// under, so, like the extractor, they're excluded here too.
+			wantCount := 0
+			for i := range transactions {
+				tx := transactions[i]
+				if !tx.Result.Successful() {
+					continue
+				}
+				changes, chErr := tx.GetChanges()
+				require.NoError(t, chErr)
+				for _, change := range changes {
+					if change.Type != xdr.LedgerEntryTypeContractData {
+						continue
+					}
+					entry := change.Post
+					if entry == nil {
+						entry = change.Pre
+					}
+					require.NotNil(t, entry)
+					contractData, ok := entry.Data.GetContractData()
+					require.True(t, ok)
+					if _, ok := contractData.Contract.GetContractId(); ok {
+						wantCount++
+					}
+				}
+			}
+
+			gotCount := 0
+			for groupKey, changes := range got {
+				_, decodeErr := strkey.Decode(strkey.VersionByteContract, groupKey)
+				assert.NoError(t, decodeErr, "group key %q must be a valid C-address strkey", groupKey)
+
+				for _, change := range changes {
+					assert.Equal(t, xdr.LedgerEntryTypeContractData, change.Type, "change must be a ContractData change")
+
+					entry := change.Post
+					if entry == nil {
+						entry = change.Pre
+					}
+					require.NotNil(t, entry, "change must have a Pre or Post entry")
+
+					contractData, ok := entry.Data.GetContractData()
+					require.True(t, ok, "entry must decode as ContractData")
+
+					contractIDBytes, ok := contractData.Contract.GetContractId()
+					require.True(t, ok, "ContractData.Contract must be a contract address")
+
+					wantAddr, encErr := strkey.Encode(strkey.VersionByteContract, contractIDBytes[:])
+					require.NoError(t, encErr)
+					assert.Equal(t, wantAddr, groupKey, "change must be grouped under its owning contract")
+				}
+				gotCount += len(changes)
+			}
+
+			assert.Equal(t, wantCount, gotCount, "extractor must return exactly the ContractData changes from successful transactions")
+			totalChanges += gotCount
+		})
+	}
+
+	require.Greater(t, totalChanges, 0, "fixture corpus must produce at least one ContractData change")
+}

internal/loadtest/runner.go

@@ -208,7 +208,7 @@ func runIngestionLoop(
 		ingestStart := time.Now()
 		processStart := time.Now()
 		buffer := indexer.NewIndexerBuffer()
-		_, err = indexer.ProcessLedger(ctx, cfg.NetworkPassphrase, ledgerMeta, ledgerIndexer, buffer)
+		_, _, err = indexer.ProcessLedger(ctx, cfg.NetworkPassphrase, ledgerMeta, ledgerIndexer, buffer)
 		if err != nil {
 			return fmt.Errorf("processing ledger %d: %w", currentLedger, err)
 		}