diff --git a/internal/indexer/indexer.go b/internal/indexer/indexer.go
index 6f67a7146..f60568ff0 100644
--- a/internal/indexer/indexer.go
+++ b/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
 }
diff --git a/internal/indexer/indexer_test.go b/internal/indexer/indexer_test.go
index 11ef00f7a..5a2903923 100644
--- a/internal/indexer/indexer_test.go
+++ b/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")
+}
diff --git a/internal/loadtest/runner.go b/internal/loadtest/runner.go
index 09a269a5c..3960df3a6 100644
--- a/internal/loadtest/runner.go
+++ b/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)
 		}
diff --git a/internal/services/ingest.go b/internal/services/ingest.go
index 99605a501..294853949 100644
--- a/internal/services/ingest.go
+++ b/internal/services/ingest.go
@@ -12,6 +12,7 @@ import (
 	set "github.com/deckarep/golang-set/v2"
 	"github.com/jackc/pgx/v5"
 	"github.com/stellar/go-stellar-sdk/historyarchive"
+	"github.com/stellar/go-stellar-sdk/ingest"
 	"github.com/stellar/go-stellar-sdk/ingest/ledgerbackend"
 	"github.com/stellar/go-stellar-sdk/support/log"
 	"github.com/stellar/go-stellar-sdk/xdr"
@@ -203,14 +204,17 @@ func (m *ingestService) Run(ctx context.Context, startLedger uint32, endLedger u
 	}
 }
 
-// processLedger processes a single ledger - gets the transactions and processes them using indexer processors.
-func (m *ingestService) processLedger(ctx context.Context, ledgerMeta xdr.LedgerCloseMeta, buffer *indexer.IndexerBuffer) error {
-	participantCount, err := indexer.ProcessLedger(ctx, m.networkPassphrase, ledgerMeta, m.ledgerIndexer, buffer)
+// processLedger processes a single ledger - gets the transactions and
+// processes them using indexer processors. The materialized transactions are
+// returned so the live path can reuse them for ContractData extraction
+// instead of building a second LedgerTransactionReader for the same ledger.
+func (m *ingestService) processLedger(ctx context.Context, ledgerMeta xdr.LedgerCloseMeta, buffer *indexer.IndexerBuffer) ([]ingest.LedgerTransaction, error) {
+	participantCount, transactions, err := indexer.ProcessLedger(ctx, m.networkPassphrase, ledgerMeta, m.ledgerIndexer, buffer)
 	if err != nil {
-		return fmt.Errorf("processing ledger %d: %w", ledgerMeta.LedgerSequence(), err)
+		return nil, fmt.Errorf("processing ledger %d: %w", ledgerMeta.LedgerSequence(), err)
 	}
 	m.appMetrics.Ingestion.ParticipantsCount.Observe(float64(participantCount))
-	return nil
+	return transactions, nil
 }
 
 // insertIntoDB persists the processed data from the buffer to the database.
diff --git a/internal/services/ingest_backfill.go b/internal/services/ingest_backfill.go
index a632225b0..6d5ca1f9a 100644
--- a/internal/services/ingest_backfill.go
+++ b/internal/services/ingest_backfill.go
@@ -342,7 +342,7 @@ func (m *ingestService) processLedgersInBatch(
 		}
 		endTime = ledgerTime
 
-		if err := m.processLedger(ctx, ledgerMeta, batchBuffer); err != nil {
+		if _, err := m.processLedger(ctx, ledgerMeta, batchBuffer); err != nil {
 			return ledgersProcessed, startTime, endTime, fmt.Errorf("processing ledger %d: %w", ledgerSeq, err)
 		}
 		ledgersProcessed++
diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go
index 3f7e11e31..6bcc2121a 100644
--- a/internal/services/ingest_live.go
+++ b/internal/services/ingest_live.go
@@ -10,6 +10,7 @@ import (
 
 	set "github.com/deckarep/golang-set/v2"
 	"github.com/jackc/pgx/v5"
+	"github.com/stellar/go-stellar-sdk/ingest"
 	"github.com/stellar/go-stellar-sdk/ingest/ledgerbackend"
 	"github.com/stellar/go-stellar-sdk/support/log"
 	"github.com/stellar/go-stellar-sdk/xdr"
@@ -33,13 +34,19 @@ const (
 // It handles: trustline assets, contract tokens, filtered data insertion,
 // token changes, and cursor update.
 func (m *ingestService) PersistLedgerData(ctx context.Context, ledgerSeq uint32, buffer *indexer.IndexerBuffer, cursorName string) (int, int, error) {
-	return m.persistLedgerData(ctx, ledgerSeq, nil, buffer, cursorName)
+	return m.persistLedgerData(ctx, ledgerSeq, nil, nil, buffer, cursorName)
 }
 
+// persistLedgerData takes the ledger's already-materialized transactions
+// (from the processLedger staging pass) so protocol ContractData extraction
+// reuses them instead of rebuilding a LedgerTransactionReader; transactions
+// is nil exactly when ledgerMeta is (the loadtest path), which also skips the
+// protocol section.
 func (m *ingestService) persistLedgerData(
 	ctx context.Context,
 	ledgerSeq uint32,
 	ledgerMeta *xdr.LedgerCloseMeta,
+	transactions []ingest.LedgerTransaction,
 	buffer *indexer.IndexerBuffer,
 	cursorName string,
 ) (int, int, error) {
@@ -70,8 +77,9 @@ func (m *ingestService) persistLedgerData(
 		// protocol side effects (e.g. SEP-41 contract_tokens metadata) happen
 		// inside this same dbTx via the validators' Validate calls. Live
 		// protocol processors then stage ledger state from the classification
-		// result before the generic protocol_wasms / protocol_contracts rows
-		// are persisted below.
+		// result; the generic protocol_contracts rows are persisted after them
+		// so a processor's name-enriched row lands first (the generic insert's
+		// COALESCE preserves it).
 		bufferedWasms := buffer.GetProtocolWasms()
 		bufferedBytecodes := buffer.GetProtocolWasmBytecodes()
 		bufferedContracts := buffer.GetProtocolContracts()
@@ -86,6 +94,25 @@ func (m *ingestService) persistLedgerData(
 			return fmt.Errorf("classifying ledger %d: %w", ledgerSeq, classifyErr)
 		}
 
+		// Persist this ledger's wasm rows BEFORE processors run: a processor
+		// enriching protocol_contracts (e.g. contract names decoded from
+		// instance storage) inserts rows that are FK-filtered against
+		// protocol_wasms, so a contract deployed in the same ledger as its
+		// wasm upload would otherwise be silently dropped.
+		if len(bufferedWasms) > 0 {
+			wasmSlice := make([]data.ProtocolWasms, 0, len(bufferedWasms))
+			for hash, wasm := range bufferedWasms {
+				if pid, ok := classification[types.HashBytea(hash)]; ok {
+					stamped := pid
+					wasm.ProtocolID = &stamped
+				}
+				wasmSlice = append(wasmSlice, wasm)
+			}
+			if txErr = m.models.ProtocolWasms.BatchInsert(ctx, dbTx, wasmSlice); txErr != nil {
+				return fmt.Errorf("inserting protocol wasms for ledger %d: %w", ledgerSeq, txErr)
+			}
+		}
+
 		// 2.6: Per-protocol CAS-gated state production. The compare-and-swap on each
 		// protocol cursor is the authoritative gate — exactly one of live ingestion or
 		// protocol-migrate wins a given ledger. Staging (ProcessLedger) and persistence
@@ -110,6 +137,12 @@ func (m *ingestService) persistLedgerData(
 				}
 			}
 
+			// ContractData extraction is lazy and memoized: it runs at most once
+			// per ledger, and only when a processor that won a CAS requires it —
+			// a protocol still backfilling costs nothing extra.
+			var contractDataChanges map[string][]ingest.Change
+			contractDataExtracted := false
+
 			for protocolID, processor := range m.protocolProcessors {
 				historyCursor := utils.ProtocolHistoryCursorName(protocolID)
 				currentStateCursor := utils.ProtocolCurrentStateCursorName(protocolID)
@@ -128,13 +161,38 @@ func (m *ingestService) persistLedgerData(
 					continue
 				}
 
-				contracts := getEffectiveProtocolContracts(protocolID, committedByProtocol[protocolID], bufferedContracts, classification)
+				committed := committedByProtocol[protocolID]
+				if processor.RequiresContractData() {
+					if !contractDataExtracted {
+						var cdErr error
+						contractDataChanges, cdErr = indexer.ExtractContractDataChangesFromTransactions(transactions, ledgerSeq)
+						if cdErr != nil {
+							return fmt.Errorf("extracting contract data changes for ledger %d: %w", ledgerSeq, cdErr)
+						}
+						contractDataExtracted = true
+					}
+
+					// ContractData-driven processors need the protocol's FULL committed
+					// membership, not just this ledger's event emitters: entries can
+					// change on a contract that emitted no event this ledger, and event
+					// decoding may disambiguate against tracked contracts that appear
+					// only in another contract's topics. Protocols requiring contract
+					// data have bounded membership, so the per-ledger query stays cheap.
+					var fullErr error
+					committed, fullErr = m.models.ProtocolContracts.GetByProtocolID(ctx, protocolID)
+					if fullErr != nil {
+						return fmt.Errorf("resolving full protocol contracts for ledger %d protocol %s: %w", ledgerSeq, protocolID, fullErr)
+					}
+				}
+
+				contracts := getEffectiveProtocolContracts(protocolID, committed, bufferedContracts, classification)
 				input := ProtocolProcessorInput{
-					LedgerSequence:    ledgerSeq,
-					LedgerCloseTime:   ledgerCloseTime,
-					ContractEvents:    contractEvents,
-					ProtocolContracts: contracts,
-					StagingMode:       StagingModeBoth,
+					LedgerSequence:      ledgerSeq,
+					LedgerCloseTime:     ledgerCloseTime,
+					ContractEvents:      contractEvents,
+					ProtocolContracts:   contracts,
+					StagingMode:         StagingModeBoth,
+					ContractDataChanges: contractDataChanges,
 				}
 				// Reset before staging so a retried transaction (ingestProcessedDataWithRetry)
 				// re-stages cleanly; the processor is long-lived and accumulates across
@@ -166,19 +224,6 @@ func (m *ingestService) persistLedgerData(
 			}
 		}
 
-		if len(bufferedWasms) > 0 {
-			wasmSlice := make([]data.ProtocolWasms, 0, len(bufferedWasms))
-			for hash, wasm := range bufferedWasms {
-				if pid, ok := classification[types.HashBytea(hash)]; ok {
-					stamped := pid
-					wasm.ProtocolID = &stamped
-				}
-				wasmSlice = append(wasmSlice, wasm)
-			}
-			if txErr = m.models.ProtocolWasms.BatchInsert(ctx, dbTx, wasmSlice); txErr != nil {
-				return fmt.Errorf("inserting protocol wasms for ledger %d: %w", ledgerSeq, txErr)
-			}
-		}
 		if len(contractSlice) > 0 {
 			if txErr = m.models.ProtocolContracts.BatchInsert(ctx, dbTx, contractSlice); txErr != nil {
 				return fmt.Errorf("inserting protocol contracts for ledger %d: %w", ledgerSeq, txErr)
@@ -348,7 +393,7 @@ func (m *ingestService) ingestLiveLedgers(ctx context.Context, startLedger uint3
 		totalStart := time.Now()
 		processStart := time.Now()
 		buffer := indexer.NewIndexerBuffer()
-		err := m.processLedger(ctx, ledgerMeta, buffer)
+		transactions, err := m.processLedger(ctx, ledgerMeta, buffer)
 		if err != nil {
 			m.appMetrics.Ingestion.ErrorsTotal.WithLabelValues("ingest_live").Inc()
 			return fmt.Errorf("processing ledger %d: %w", currentLedger, err)
@@ -357,7 +402,7 @@ func (m *ingestService) ingestLiveLedgers(ctx context.Context, startLedger uint3
 
 		// All DB operations in a single atomic transaction with retry
 		dbStart := time.Now()
-		numTransactionProcessed, numOperationProcessed, err := m.ingestProcessedDataWithRetry(ctx, currentLedger, ledgerMeta, buffer)
+		numTransactionProcessed, numOperationProcessed, err := m.ingestProcessedDataWithRetry(ctx, currentLedger, ledgerMeta, transactions, buffer)
 		if err != nil {
 			m.appMetrics.Ingestion.ErrorsTotal.WithLabelValues("ingest_live").Inc()
 			return fmt.Errorf("processing ledger %d: %w", currentLedger, err)
@@ -465,6 +510,7 @@ func (m *ingestService) ingestProcessedDataWithRetry(
 	ctx context.Context,
 	currentLedger uint32,
 	ledgerMeta xdr.LedgerCloseMeta,
+	transactions []ingest.LedgerTransaction,
 	buffer *indexer.IndexerBuffer,
 ) (int, int, error) {
 	var lastErr error
@@ -475,7 +521,7 @@ func (m *ingestService) ingestProcessedDataWithRetry(
 		default:
 		}
 
-		numTxs, numOps, err := m.persistLedgerData(ctx, currentLedger, &ledgerMeta, buffer, data.LatestLedgerCursorName)
+		numTxs, numOps, err := m.persistLedgerData(ctx, currentLedger, &ledgerMeta, transactions, buffer, data.LatestLedgerCursorName)
 		if err == nil {
 			return numTxs, numOps, nil
 		}
diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go
index 69b0e221d..04abda38f 100644
--- a/internal/services/ingest_test.go
+++ b/internal/services/ingest_test.go
@@ -2,6 +2,7 @@ package services
 
 import (
 	"context"
+	"encoding/hex"
 	"fmt"
 	"strconv"
 	"testing"
@@ -1605,7 +1606,7 @@ func Test_ingestProcessedDataWithRetry(t *testing.T) {
 
 		// Call ingestProcessedDataWithRetry - should succeed
 		// Note: assetIDMap and contractIDMap are no longer passed - operations use direct DB queries
-		numTx, numOps, err := svc.ingestProcessedDataWithRetry(ctx, 100, xdr.LedgerCloseMeta{}, buffer)
+		numTx, numOps, err := svc.ingestProcessedDataWithRetry(ctx, 100, xdr.LedgerCloseMeta{}, nil, buffer)
 
 		// Verify success
 		require.NoError(t, err)
@@ -1686,7 +1687,7 @@ func Test_ingestProcessedDataWithRetry(t *testing.T) {
 
 		// Call ingestProcessedDataWithRetry - should fail after retries due to DB error
 		// Note: assetIDMap and contractIDMap are no longer passed - operations use direct DB queries
-		_, _, err = svc.ingestProcessedDataWithRetry(ctx, 100, xdr.LedgerCloseMeta{}, buffer)
+		_, _, err = svc.ingestProcessedDataWithRetry(ctx, 100, xdr.LedgerCloseMeta{}, nil, buffer)
 
 		// Verify error propagates with retry failure message
 		require.Error(t, err)
@@ -1776,7 +1777,7 @@ func Test_ingestProcessedDataWithRetry(t *testing.T) {
 
 		// Call ingestProcessedDataWithRetry - should succeed after retry
 		// Note: assetIDMap and contractIDMap are no longer passed - operations use direct DB queries
-		numTx, numOps, err := svc.ingestProcessedDataWithRetry(ctx, 100, xdr.LedgerCloseMeta{}, buffer)
+		numTx, numOps, err := svc.ingestProcessedDataWithRetry(ctx, 100, xdr.LedgerCloseMeta{}, nil, buffer)
 
 		// Verify success after retry
 		require.NoError(t, err)
@@ -1874,6 +1875,8 @@ type testProtocolProcessor struct {
 
 func (p *testProtocolProcessor) ProtocolID() string { return p.id }
 
+func (p *testProtocolProcessor) RequiresContractData() bool { return false }
+
 func (p *testProtocolProcessor) Reset() { p.stagedLedgerCount = 0 }
 
 func (p *testProtocolProcessor) ProcessLedger(_ context.Context, input ProtocolProcessorInput) error {
@@ -1972,7 +1975,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 
 		buffer := indexer.NewIndexerBuffer()
 		meta := dummyLedgerMeta(100)
-		_, _, err := svc.persistLedgerData(ctx, 100, &meta, buffer, "latest_ledger_cursor")
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// Both protocol cursors should advance to 100
@@ -2004,7 +2007,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 
 		buffer := indexer.NewIndexerBuffer()
 		meta := dummyLedgerMeta(100)
-		_, _, err := svc.persistLedgerData(ctx, 100, &meta, buffer, "latest_ledger_cursor")
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// Cursors should stay at 100 (CAS expected 99 but found 100)
@@ -2036,7 +2039,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 
 		buffer := indexer.NewIndexerBuffer()
 		meta := dummyLedgerMeta(100)
-		_, _, err := svc.persistLedgerData(ctx, 100, &meta, buffer, "latest_ledger_cursor")
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// Cursors should stay at 98 (behind, so entire block is skipped)
@@ -2073,7 +2076,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 
 		buffer := indexer.NewIndexerBuffer()
 		meta := dummyLedgerMeta(100)
-		_, _, err := svc.persistLedgerData(ctx, 100, &meta, buffer, "latest_ledger_cursor")
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// Main cursor advances; protocol persist methods were not called and
@@ -2110,7 +2113,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 		require.NoError(t, err)
 
 		meta := dummyLedgerMeta(100)
-		_, _, err = svc.persistLedgerData(ctx, 100, &meta, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
+		_, _, err = svc.persistLedgerData(ctx, 100, &meta, nil, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// History CAS succeeded.
@@ -2149,7 +2152,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 		require.NoError(t, err)
 
 		meta := dummyLedgerMeta(100)
-		_, _, err = svc.persistLedgerData(ctx, 100, &meta, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
+		_, _, err = svc.persistLedgerData(ctx, 100, &meta, nil, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// Current-state CAS succeeded.
@@ -2178,7 +2181,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 
 		buffer := indexer.NewIndexerBuffer()
 		meta := dummyLedgerMeta(100)
-		_, _, err := svc.persistLedgerData(ctx, 100, &meta, buffer, "latest_ledger_cursor")
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// Main cursor should advance
@@ -2197,14 +2200,14 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 		// First ledger succeeds and advances the current-state cursor to 100.
 		processor.processedLedger = 100
 		meta100 := dummyLedgerMeta(100)
-		_, _, err := svc.persistLedgerData(ctx, 100, &meta100, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta100, nil, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		// Next ledger fails inside PersistCurrentState, rolling back the whole
 		// transaction — the current-state cursor must stay at 100.
 		processor.processedLedger = 101
 		meta101 := dummyLedgerMeta(101)
-		_, _, err = svc.persistLedgerData(ctx, 101, &meta101, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
+		_, _, err = svc.persistLedgerData(ctx, 101, &meta101, nil, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
 		require.Error(t, err)
 
 		currentStateCursor, err := models.IngestStore.Get(ctx, "protocol_testproto_current_state_cursor")
@@ -2214,7 +2217,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 		// Retrying the same ledger succeeds and advances the cursor.
 		processor.failPersistCurrentStateAt = 0
 		processor.processedLedger = 101
-		_, _, err = svc.persistLedgerData(ctx, 101, &meta101, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
+		_, _, err = svc.persistLedgerData(ctx, 101, &meta101, nil, indexer.NewIndexerBuffer(), "latest_ledger_cursor")
 		require.NoError(t, err)
 
 		currentStateCursor, err = models.IngestStore.Get(ctx, "protocol_testproto_current_state_cursor")
@@ -2226,6 +2229,108 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 		assert.Equal(t, uint32(1), stagedCount) // retry re-staged cleanly; not doubled
 	})
 
+	t.Run("H: RequiresContractData true — processor receives non-nil ContractDataChanges", func(t *testing.T) {
+		processor := NewProtocolProcessorMock(t)
+		processor.On("ProtocolID").Return("testproto")
+		processor.On("RequiresContractData").Return(true)
+		processor.On("Reset").Return()
+		processor.On("ProcessLedger", mock.Anything, mock.MatchedBy(func(in ProtocolProcessorInput) bool {
+			return in.ContractDataChanges != nil
+		})).Return(nil)
+		processor.On("PersistHistory", mock.Anything, mock.Anything).Return(nil)
+		processor.On("PersistCurrentState", mock.Anything, mock.Anything).Return(nil)
+
+		ctx, svc, _, pool := setupTest(t, []ProtocolProcessor{processor})
+		setupDBCursors(t, ctx, pool, 99, 99)
+		setupProtocolCursors(t, ctx, pool, 99, 99)
+
+		buffer := indexer.NewIndexerBuffer()
+		meta := dummyLedgerMeta(100)
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
+		require.NoError(t, err)
+	})
+
+	t.Run("I: RequiresContractData false — processor receives nil ContractDataChanges", func(t *testing.T) {
+		processor := NewProtocolProcessorMock(t)
+		processor.On("ProtocolID").Return("testproto")
+		processor.On("RequiresContractData").Return(false)
+		processor.On("Reset").Return()
+		processor.On("ProcessLedger", mock.Anything, mock.MatchedBy(func(in ProtocolProcessorInput) bool {
+			return in.ContractDataChanges == nil
+		})).Return(nil)
+		processor.On("PersistHistory", mock.Anything, mock.Anything).Return(nil)
+		processor.On("PersistCurrentState", mock.Anything, mock.Anything).Return(nil)
+
+		ctx, svc, _, pool := setupTest(t, []ProtocolProcessor{processor})
+		setupDBCursors(t, ctx, pool, 99, 99)
+		setupProtocolCursors(t, ctx, pool, 99, 99)
+
+		buffer := indexer.NewIndexerBuffer()
+		meta := dummyLedgerMeta(100)
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
+		require.NoError(t, err)
+	})
+
+	t.Run("J: RequiresContractData true — full committed membership reaches the processor", func(t *testing.T) {
+		// A classified contract that emits NO events this ledger must still be
+		// in a ContractData-requiring processor's ProtocolContracts (entries can
+		// change without events, and event decoding disambiguates against the
+		// full tracked set). Event-only processors keep the cheaper
+		// event-derived membership, which is empty for this event-less ledger.
+		contractID := []byte("contract_no_events_this_ledger33")
+		wasmHash := []byte("wasm_hash_for_membership_test333")
+
+		requiring := NewProtocolProcessorMock(t)
+		requiring.On("ProtocolID").Return("testproto")
+		requiring.On("RequiresContractData").Return(true)
+		requiring.On("Reset").Return()
+		// ContractID scans BYTEA into its hex form (the same convention
+		// bufferedContracts keys use), so compare against the hex encoding.
+		wantContractID := hex.EncodeToString(contractID)
+		requiring.On("ProcessLedger", mock.Anything, mock.MatchedBy(func(in ProtocolProcessorInput) bool {
+			for _, c := range in.ProtocolContracts {
+				if string(c.ContractID) == wantContractID {
+					return true
+				}
+			}
+			return false
+		})).Return(nil)
+		requiring.On("PersistHistory", mock.Anything, mock.Anything).Return(nil)
+		requiring.On("PersistCurrentState", mock.Anything, mock.Anything).Return(nil)
+
+		eventOnly := NewProtocolProcessorMock(t)
+		eventOnly.On("ProtocolID").Return("otherproto")
+		eventOnly.On("RequiresContractData").Return(false)
+		eventOnly.On("Reset").Return()
+		eventOnly.On("ProcessLedger", mock.Anything, mock.MatchedBy(func(in ProtocolProcessorInput) bool {
+			return len(in.ProtocolContracts) == 0
+		})).Return(nil)
+		eventOnly.On("PersistHistory", mock.Anything, mock.Anything).Return(nil)
+		eventOnly.On("PersistCurrentState", mock.Anything, mock.Anything).Return(nil)
+
+		ctx, svc, _, pool := setupTest(t, []ProtocolProcessor{requiring, eventOnly})
+		setupDBCursors(t, ctx, pool, 99, 99)
+		setupProtocolCursors(t, ctx, pool, 99, 99)
+		// setupProtocolCursors seeds "testproto" only; otherproto needs its own
+		// cursors for the CAS swap to succeed.
+		_, err := pool.Exec(ctx,
+			`INSERT INTO ingest_store (key, value) VALUES ($1, '99'), ($2, '99')`,
+			utils.ProtocolHistoryCursorName("otherproto"), utils.ProtocolCurrentStateCursorName("otherproto"))
+		require.NoError(t, err)
+
+		_, err = pool.Exec(ctx, `INSERT INTO protocols (id) VALUES ('testproto'), ('otherproto') ON CONFLICT (id) DO NOTHING`)
+		require.NoError(t, err)
+		_, err = pool.Exec(ctx, `INSERT INTO protocol_wasms (wasm_hash, protocol_id) VALUES ($1, 'testproto')`, wasmHash)
+		require.NoError(t, err)
+		_, err = pool.Exec(ctx, `INSERT INTO protocol_contracts (contract_id, wasm_hash) VALUES ($1, $2)`, contractID, wasmHash)
+		require.NoError(t, err)
+
+		buffer := indexer.NewIndexerBuffer()
+		meta := dummyLedgerMeta(100)
+		_, _, err = svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
+		require.NoError(t, err)
+	})
+
 	t.Run("G: contract-id lookup failure fails the ledger", func(t *testing.T) {
 		processor := &testProtocolProcessor{id: "testproto"}
 		ctx, svc, models, pool := setupTest(t, []ProtocolProcessor{processor})
@@ -2249,7 +2354,7 @@ func Test_PersistLedgerData_ProtocolCASGating(t *testing.T) {
 		)
 
 		meta := dummyLedgerMeta(100)
-		_, _, err := svc.persistLedgerData(ctx, 100, &meta, buffer, "latest_ledger_cursor")
+		_, _, err := svc.persistLedgerData(ctx, 100, &meta, nil, buffer, "latest_ledger_cursor")
 		require.ErrorContains(t, err, "resolving protocol contracts for ledger 100")
 
 		// The transaction rolled back: the protocol history cursor stayed at 99.
diff --git a/internal/services/mocks.go b/internal/services/mocks.go
index 407679720..b42ae4065 100644
--- a/internal/services/mocks.go
+++ b/internal/services/mocks.go
@@ -269,6 +269,11 @@ func (m *ProtocolProcessorMock) ProcessLedger(ctx context.Context, input Protoco
 	return args.Error(0)
 }
 
+func (m *ProtocolProcessorMock) RequiresContractData() bool {
+	args := m.Called()
+	return args.Bool(0)
+}
+
 func (m *ProtocolProcessorMock) Reset() {
 	m.Called()
 }
diff --git a/internal/services/processor_registry_test.go b/internal/services/processor_registry_test.go
index 3af5809ea..b21d5cc23 100644
--- a/internal/services/processor_registry_test.go
+++ b/internal/services/processor_registry_test.go
@@ -7,6 +7,13 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestProtocolProcessor_RequiresContractData(t *testing.T) {
+	m := NewProtocolProcessorMock(t)
+	m.On("RequiresContractData").Return(false).Once()
+	var p ProtocolProcessor = m
+	assert.False(t, p.RequiresContractData())
+}
+
 func withCleanProcessorRegistry(t *testing.T) {
 	t.Helper()
 	original := processorRegistry
diff --git a/internal/services/protocol_migrate.go b/internal/services/protocol_migrate.go
index b0b6a46cd..eee8c8fb2 100644
--- a/internal/services/protocol_migrate.go
+++ b/internal/services/protocol_migrate.go
@@ -2,14 +2,17 @@ package services
 
 import (
 	"context"
+	"encoding/hex"
 	"fmt"
 	"strconv"
 	"time"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stellar/go-stellar-sdk/ingest"
 	"github.com/stellar/go-stellar-sdk/ingest/ledgerbackend"
 	"github.com/stellar/go-stellar-sdk/support/log"
+	"github.com/stellar/go-stellar-sdk/xdr"
 
 	"github.com/stellar/wallet-backend/internal/data"
 	"github.com/stellar/wallet-backend/internal/db"
@@ -221,7 +224,7 @@ func (s *protocolMigrateEngine) validate(ctx context.Context, protocolIDs []stri
 // less often than the progress-log interval.
 type stageTimers struct {
 	fetch   time.Duration // GetLedger wait (consumer stall, not download time); ≈0 while the datastore prefetch keeps up, rises only when it can't
-	extract time.Duration // ExtractContractEventsForLedger
+	extract time.Duration // ExtractContractEventsForLedger + ExtractContractDataChangesForLedger
 	process time.Duration // ProcessLedger across all trackers
 	flush   time.Duration // flushWindow (CAS + Persist), including tip flushes
 }
@@ -274,8 +277,15 @@ func (s *protocolMigrateEngine) processAllProtocols(ctx context.Context, protoco
 		})
 	}
 
-	// Load contracts once — all relevant contracts are in the DB before migration starts
-	// (validate() requires ClassificationStatus == StatusSuccess).
+	// Load each tracker's classified-contract membership. For event-only
+	// processors this run-start snapshot is final — validate() requires
+	// ClassificationStatus == StatusSuccess, so everything classified before
+	// the run is already here. Processors that require ContractData get their
+	// membership refreshed after every committed window instead (see
+	// refreshTrackerContracts): live ingestion's validator classifies newly
+	// deployed contracts concurrently with this run, and those processors
+	// interpret events and entries by membership, so folding the rest of the
+	// run against a stale snapshot would silently drop that contract's state.
 	contractsByProtocol := make(map[string][]data.ProtocolContracts, len(trackers))
 	for _, t := range trackers {
 		contracts, loadErr := s.protocolContractsModel.GetByProtocolID(ctx, t.protocolID)
@@ -338,7 +348,7 @@ func (s *protocolMigrateEngine) processAllProtocols(ctx context.Context, protoco
 		// next (blocking) GetLedger never strands the cursor behind the frontier.
 		var flushErr error
 		tipFlushStart := time.Now()
-		cachedTip, flushErr = s.flushWindowsAtTip(ctx, trackers, seq, cachedTip)
+		cachedTip, flushErr = s.flushWindowsAtTip(ctx, trackers, seq, cachedTip, contractsByProtocol)
 		timers.flush += time.Since(tipFlushStart)
 		if flushErr != nil {
 			return handedOffProtocolIDs(trackers), flushErr
@@ -369,16 +379,41 @@ func (s *protocolMigrateEngine) processAllProtocols(ctx context.Context, protoco
 		}
 		ledgerCloseTime := ledgerMeta.LedgerCloseTime()
 
+		// Extract ContractData changes once per ledger, only when a tracker that
+		// will fold this ledger requires them; all trackers share the map. The
+		// union of those trackers' memberships gates the extraction: ledgers
+		// whose transaction footprints touch no tracked contract skip the
+		// change walk entirely (the overwhelmingly common case).
+		var ledgerContractDataChanges map[string][]ingest.Change
+		if trackersRequireContractData(trackers, seq) {
+			cdStart := time.Now()
+			tracked, trackErr := trackedContractIDSet(trackers, seq, contractsByProtocol)
+			if trackErr != nil {
+				return handedOffProtocolIDs(trackers), trackErr
+			}
+			var cdErr error
+			ledgerContractDataChanges, cdErr = indexer.ExtractContractDataChangesForLedger(ledgerMeta, tracked)
+			cdDur := time.Since(cdStart)
+			timers.extract += cdDur
+			// Distinct phase label: the events extraction already observes one
+			// "extract" sample per ledger, and dashboards assume that cardinality.
+			s.metrics.PhaseDuration.WithLabelValues("extract_contract_data").Observe(cdDur.Seconds())
+			if cdErr != nil {
+				return handedOffProtocolIDs(trackers), fmt.Errorf("extracting contract data changes for ledger %d: %w", seq, cdErr)
+			}
+		}
+
 		for _, t := range trackers {
 			if t.handedOff || t.cursorValue+t.pending >= seq {
 				continue
 			}
 			input := ProtocolProcessorInput{
-				LedgerSequence:    seq,
-				LedgerCloseTime:   ledgerCloseTime,
-				ContractEvents:    ledgerEvents,
-				ProtocolContracts: contractsByProtocol[t.protocolID],
-				StagingMode:       s.strategy.Mode,
+				LedgerSequence:      seq,
+				LedgerCloseTime:     ledgerCloseTime,
+				ContractEvents:      ledgerEvents,
+				ProtocolContracts:   contractsByProtocol[t.protocolID],
+				StagingMode:         s.strategy.Mode,
+				ContractDataChanges: ledgerContractDataChanges,
 			}
 			processStart := time.Now()
 			processErr := t.processor.ProcessLedger(ctx, input)
@@ -396,6 +431,9 @@ func (s *protocolMigrateEngine) processAllProtocols(ctx context.Context, protoco
 				if flushWinErr != nil {
 					return handedOffProtocolIDs(trackers), flushWinErr
 				}
+				if refreshErr := s.refreshTrackerContracts(ctx, t, contractsByProtocol); refreshErr != nil {
+					return handedOffProtocolIDs(trackers), refreshErr
+				}
 			}
 		}
 
@@ -465,7 +503,7 @@ func (s *protocolMigrateEngine) refreshTargetTip(ctx context.Context) {
 // GetLatestLedgerSequence call per ledger; the tip is monotonic, so a stale value at worst
 // costs one extra refresh. The updated cachedTip is returned for the next iteration. A
 // flush can hand off the last active tracker, so callers must re-check allHandedOff after.
-func (s *protocolMigrateEngine) flushWindowsAtTip(ctx context.Context, trackers []*protocolTracker, seq, cachedTip uint32) (uint32, error) {
+func (s *protocolMigrateEngine) flushWindowsAtTip(ctx context.Context, trackers []*protocolTracker, seq, cachedTip uint32, contractsByProtocol map[string][]data.ProtocolContracts) (uint32, error) {
 	if seq <= cachedTip {
 		return cachedTip, nil
 	}
@@ -479,10 +517,32 @@ func (s *protocolMigrateEngine) flushWindowsAtTip(ctx context.Context, trackers
 		if err := s.flushWindow(ctx, t); err != nil {
 			return cachedTip, err
 		}
+		if err := s.refreshTrackerContracts(ctx, t, contractsByProtocol); err != nil {
+			return cachedTip, err
+		}
 	}
 	return cachedTip, nil
 }
 
+// refreshTrackerContracts re-reads a tracker's classified-contract membership
+// after a window commit, but only for processors that require ContractData —
+// they interpret events and entries by classified-set membership, so a
+// contract classified mid-run (live ingestion's validator runs concurrently)
+// must reach them on the next window rather than never. Event-only processors
+// keep the cheaper run-start snapshot. Handed-off trackers fold no further
+// ledgers, so their membership no longer matters.
+func (s *protocolMigrateEngine) refreshTrackerContracts(ctx context.Context, t *protocolTracker, contractsByProtocol map[string][]data.ProtocolContracts) error {
+	if t.handedOff || !t.processor.RequiresContractData() {
+		return nil
+	}
+	contracts, err := s.protocolContractsModel.GetByProtocolID(ctx, t.protocolID)
+	if err != nil {
+		return fmt.Errorf("refreshing contracts for %s: %w", t.protocolID, err)
+	}
+	contractsByProtocol[t.protocolID] = contracts
+	return nil
+}
+
 // flushWindow commits a tracker's open window [cursorValue+1, cursorValue+pending] in a
 // single transaction: CAS(winStart-1 -> winEnd) then merged Persist. A failed CAS means
 // live ingestion took winStart, so the whole window is discarded and the tracker hands off.
@@ -552,6 +612,43 @@ func allHandedOff(trackers []*protocolTracker) bool {
 	return true
 }
 
+// trackersRequireContractData reports whether any tracker that will fold this
+// ledger has a processor needing ContractData changes.
+func trackersRequireContractData(trackers []*protocolTracker, seq uint32) bool {
+	for _, t := range trackers {
+		if t.handedOff || t.cursorValue+t.pending >= seq {
+			continue
+		}
+		if t.processor.RequiresContractData() {
+			return true
+		}
+	}
+	return false
+}
+
+// trackedContractIDSet unions the classified-contract membership of every
+// ContractData-requiring tracker that will fold seq, as raw 32-byte contract
+// IDs — the footprint gate's lookup key. Rebuilt per ledger so a membership
+// refresh (refreshTrackerContracts) takes effect immediately; the sets are
+// small (a protocol requiring ContractData has bounded membership), so the
+// per-ledger cost is noise.
+func trackedContractIDSet(trackers []*protocolTracker, seq uint32, contractsByProtocol map[string][]data.ProtocolContracts) (map[xdr.ContractId]struct{}, error) {
+	tracked := map[xdr.ContractId]struct{}{}
+	for _, t := range trackers {
+		if t.handedOff || t.cursorValue+t.pending >= seq || !t.processor.RequiresContractData() {
+			continue
+		}
+		for _, c := range contractsByProtocol[t.protocolID] {
+			idBytes, err := hex.DecodeString(string(c.ContractID))
+			if err != nil || len(idBytes) != len(xdr.ContractId{}) {
+				return nil, fmt.Errorf("protocol %s contract id %q is not a 32-byte hex hash: %w", t.protocolID, c.ContractID, err)
+			}
+			tracked[xdr.ContractId(idBytes)] = struct{}{}
+		}
+	}
+	return tracked, nil
+}
+
 // handedOffProtocolIDs returns the IDs of trackers that have been handed off to live ingestion.
 func handedOffProtocolIDs(trackers []*protocolTracker) []string {
 	var ids []string
diff --git a/internal/services/protocol_migrate_test.go b/internal/services/protocol_migrate_test.go
index c32949fdc..527e7e2b3 100644
--- a/internal/services/protocol_migrate_test.go
+++ b/internal/services/protocol_migrate_test.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -20,6 +21,7 @@ import (
 	"github.com/stellar/wallet-backend/internal/data"
 	"github.com/stellar/wallet-backend/internal/db"
 	"github.com/stellar/wallet-backend/internal/db/dbtest"
+	"github.com/stellar/wallet-backend/internal/indexer/types"
 	"github.com/stellar/wallet-backend/internal/metrics"
 	"github.com/stellar/wallet-backend/internal/utils"
 )
@@ -180,10 +182,13 @@ type testRecordingProcessor struct {
 	persistedCurrentStateSeqs []uint32
 	lastProcessed             uint32
 	resetCount                int
+	requiresContractData      bool
 }
 
 func (p *testRecordingProcessor) ProtocolID() string { return p.id }
 
+func (p *testRecordingProcessor) RequiresContractData() bool { return p.requiresContractData }
+
 func (p *testRecordingProcessor) Reset() { p.resetCount++ }
 
 func (p *testRecordingProcessor) ProcessLedger(_ context.Context, input ProtocolProcessorInput) error {
@@ -619,6 +624,7 @@ func TestProtocolMigrateEngine(t *testing.T) {
 		protocolContractsModel.On("GetByProtocolID", mock.Anything, "testproto").Return([]data.ProtocolContracts{}, nil)
 
 		processorMock.On("ProtocolID").Return("testproto")
+		processorMock.On("RequiresContractData").Return(false)
 		processorMock.On("ProcessLedger", mock.Anything, mock.Anything).Return(fmt.Errorf("simulated ProcessLedger error"))
 
 		backend := &multiLedgerBackend{
@@ -663,6 +669,7 @@ func TestProtocolMigrateEngine(t *testing.T) {
 		protocolContractsModel.On("GetByProtocolID", mock.Anything, "testproto").Return([]data.ProtocolContracts{}, nil)
 
 		processorMock.On("ProtocolID").Return("testproto")
+		processorMock.On("RequiresContractData").Return(false)
 		processorMock.On("ProcessLedger", mock.Anything, mock.Anything).Return(nil)
 		processorMock.On("PersistHistory", mock.Anything, mock.Anything).Return(fmt.Errorf("simulated PersistHistory error"))
 
@@ -1121,6 +1128,212 @@ func TestProtocolMigrateEngine(t *testing.T) {
 		assert.Equal(t, uint32(201), cursorVal)
 		assert.Equal(t, []uint32{100}, processor.persistedHistorySeqs)
 	})
+
+	t.Run("ContractDataChanges populated when a selected processor requires it", func(t *testing.T) {
+		ctx := context.Background()
+		dbPool, ingestStore := setupTestDB(t)
+
+		setIngestStoreValue(t, ctx, dbPool, "oldest_ingest_ledger", 100)
+		setIngestStoreValue(t, ctx, dbPool, "latest_ingest_ledger", 101)
+
+		_, err := dbPool.Exec(ctx, `INSERT INTO protocols (id, classification_status) VALUES ('cdproto', 'success') ON CONFLICT (id) DO UPDATE SET classification_status = 'success'`)
+		require.NoError(t, err)
+		_, err = dbPool.Exec(ctx, `INSERT INTO protocols (id, classification_status) VALUES ('noproto', 'success') ON CONFLICT (id) DO UPDATE SET classification_status = 'success'`)
+		require.NoError(t, err)
+
+		protocolsModel := data.NewProtocolsModelMock(t)
+		protocolContractsModel := data.NewProtocolContractsModelMock(t)
+		// cdProc requires ContractData; noProc does not. Both use testCursorAdvancingProcessor
+		// to trigger CAS handoff on the last ledger, allowing the unbounded loop to terminate.
+		cdProc := &testCursorAdvancingProcessor{
+			testRecordingProcessor: testRecordingProcessor{id: "cdproto", ingestStore: ingestStore, requiresContractData: true},
+			dbPool:                 dbPool,
+			advanceAtSeq:           101,
+			cursorNameFunc:         utils.ProtocolHistoryCursorName,
+		}
+		noProc := &testCursorAdvancingProcessor{
+			testRecordingProcessor: testRecordingProcessor{id: "noproto", ingestStore: ingestStore, requiresContractData: false},
+			dbPool:                 dbPool,
+			advanceAtSeq:           101,
+			cursorNameFunc:         utils.ProtocolHistoryCursorName,
+		}
+
+		protocolsModel.On("GetByIDs", mock.Anything, []string{"cdproto", "noproto"}).Return([]data.Protocols{
+			{ID: "cdproto", ClassificationStatus: data.StatusSuccess, HistoryMigrationStatus: data.StatusNotStarted},
+			{ID: "noproto", ClassificationStatus: data.StatusSuccess, HistoryMigrationStatus: data.StatusNotStarted},
+		}, nil)
+		protocolsModel.On("UpdateHistoryMigrationStatus", mock.Anything, mock.Anything, []string{"cdproto", "noproto"}, data.StatusInProgress).Return(nil)
+		protocolsModel.On("UpdateHistoryMigrationStatus", mock.Anything, mock.Anything, []string{"cdproto", "noproto"}, data.StatusSuccess).Return(nil)
+
+		protocolContractsModel.On("GetByProtocolID", mock.Anything, "cdproto").Return([]data.ProtocolContracts{}, nil)
+		protocolContractsModel.On("GetByProtocolID", mock.Anything, "noproto").Return([]data.ProtocolContracts{}, nil)
+
+		backend := &multiLedgerBackend{
+			ledgers: map[uint32]xdr.LedgerCloseMeta{
+				100: dummyLedgerMeta(100),
+				101: dummyLedgerMeta(101),
+			},
+		}
+
+		svc, err := NewProtocolMigrateHistoryService(ProtocolMigrateHistoryConfig{
+			DB: dbPool, LedgerBackend: backend,
+			ProtocolsModel: protocolsModel, ProtocolContractsModel: protocolContractsModel,
+			IngestStore: ingestStore, NetworkPassphrase: "Test SDF Network ; September 2015",
+			Processors: []ProtocolProcessor{cdProc, noProc},
+		})
+		require.NoError(t, err)
+
+		err = svc.Run(ctx, []string{"cdproto", "noproto"})
+		require.NoError(t, err)
+
+		// The requiring processor sees a non-nil (extraction ran) map on every ledger it folded.
+		require.Len(t, cdProc.processedInputs, 2)
+		for _, input := range cdProc.processedInputs {
+			assert.NotNil(t, input.ContractDataChanges, "ledger %d: requiring processor must see non-nil ContractDataChanges", input.LedgerSequence)
+		}
+
+		// The map is computed once per ledger and shared across trackers, so the
+		// non-requiring processor also receives it (and is expected to ignore it).
+		require.Len(t, noProc.processedInputs, 2)
+		for _, input := range noProc.processedInputs {
+			assert.NotNil(t, input.ContractDataChanges, "ledger %d: shared map must also reach the non-requiring processor", input.LedgerSequence)
+		}
+	})
+
+	t.Run("ProtocolContracts membership refreshed per flushed window for requiring processors", func(t *testing.T) {
+		ctx := context.Background()
+		dbPool, ingestStore := setupTestDB(t)
+
+		setIngestStoreValue(t, ctx, dbPool, "oldest_ingest_ledger", 100)
+		setIngestStoreValue(t, ctx, dbPool, "latest_ingest_ledger", 102)
+
+		_, err := dbPool.Exec(ctx, `INSERT INTO protocols (id, classification_status) VALUES ('cdproto', 'success') ON CONFLICT (id) DO UPDATE SET classification_status = 'success'`)
+		require.NoError(t, err)
+		_, err = dbPool.Exec(ctx, `INSERT INTO protocols (id, classification_status) VALUES ('noproto', 'success') ON CONFLICT (id) DO UPDATE SET classification_status = 'success'`)
+		require.NoError(t, err)
+
+		protocolsModel := data.NewProtocolsModelMock(t)
+		protocolContractsModel := data.NewProtocolContractsModelMock(t)
+		cdProc := &testCursorAdvancingProcessor{
+			testRecordingProcessor: testRecordingProcessor{id: "cdproto", ingestStore: ingestStore, requiresContractData: true},
+			dbPool:                 dbPool,
+			advanceAtSeq:           102,
+			cursorNameFunc:         utils.ProtocolHistoryCursorName,
+		}
+		noProc := &testCursorAdvancingProcessor{
+			testRecordingProcessor: testRecordingProcessor{id: "noproto", ingestStore: ingestStore, requiresContractData: false},
+			dbPool:                 dbPool,
+			advanceAtSeq:           102,
+			cursorNameFunc:         utils.ProtocolHistoryCursorName,
+		}
+
+		protocolsModel.On("GetByIDs", mock.Anything, []string{"cdproto", "noproto"}).Return([]data.Protocols{
+			{ID: "cdproto", ClassificationStatus: data.StatusSuccess, HistoryMigrationStatus: data.StatusNotStarted},
+			{ID: "noproto", ClassificationStatus: data.StatusSuccess, HistoryMigrationStatus: data.StatusNotStarted},
+		}, nil)
+		protocolsModel.On("UpdateHistoryMigrationStatus", mock.Anything, mock.Anything, []string{"cdproto", "noproto"}, data.StatusInProgress).Return(nil)
+		protocolsModel.On("UpdateHistoryMigrationStatus", mock.Anything, mock.Anything, []string{"cdproto", "noproto"}, data.StatusSuccess).Return(nil)
+
+		// Realistic 32-byte hex IDs (the BYTEA scan form): the footprint gate
+		// hex-decodes tracked contract IDs and fails the run on malformed ones.
+		contractA := data.ProtocolContracts{ContractID: types.HashBytea(strings.Repeat("aa", 32))}
+		contractB := data.ProtocolContracts{ContractID: types.HashBytea(strings.Repeat("bb", 32))}
+		// The requiring tracker's membership is re-read after every committed
+		// window (WindowSize defaults to 1, so after every ledger): the run-start
+		// snapshot returns only A, every refresh returns A+B — simulating live
+		// ingestion classifying B while the migration is in flight. The
+		// event-only tracker keeps the run-start snapshot: exactly one load.
+		protocolContractsModel.On("GetByProtocolID", mock.Anything, "cdproto").Return([]data.ProtocolContracts{contractA}, nil).Once()
+		protocolContractsModel.On("GetByProtocolID", mock.Anything, "cdproto").Return([]data.ProtocolContracts{contractA, contractB}, nil)
+		protocolContractsModel.On("GetByProtocolID", mock.Anything, "noproto").Return([]data.ProtocolContracts{contractA}, nil).Once()
+
+		backend := &multiLedgerBackend{
+			ledgers: map[uint32]xdr.LedgerCloseMeta{
+				100: dummyLedgerMeta(100),
+				101: dummyLedgerMeta(101),
+				102: dummyLedgerMeta(102),
+			},
+		}
+
+		svc, err := NewProtocolMigrateHistoryService(ProtocolMigrateHistoryConfig{
+			DB: dbPool, LedgerBackend: backend,
+			ProtocolsModel: protocolsModel, ProtocolContractsModel: protocolContractsModel,
+			IngestStore: ingestStore, NetworkPassphrase: "Test SDF Network ; September 2015",
+			Processors: []ProtocolProcessor{cdProc, noProc},
+		})
+		require.NoError(t, err)
+
+		err = svc.Run(ctx, []string{"cdproto", "noproto"})
+		require.NoError(t, err)
+
+		// Requiring processor: ledger 100 folded with the run-start snapshot,
+		// ledgers 101-102 with the refreshed membership including contract B.
+		require.Len(t, cdProc.processedInputs, 3)
+		assert.Len(t, cdProc.processedInputs[0].ProtocolContracts, 1, "ledger 100 folds with the run-start snapshot")
+		for _, input := range cdProc.processedInputs[1:] {
+			assert.Len(t, input.ProtocolContracts, 2, "ledger %d: refreshed membership must include the newly classified contract", input.LedgerSequence)
+		}
+
+		// Event-only processor: the run-start snapshot is never refreshed (its
+		// single .Once() mock expectation also fails the test on extra calls).
+		require.Len(t, noProc.processedInputs, 3)
+		for _, input := range noProc.processedInputs {
+			assert.Len(t, input.ProtocolContracts, 1, "ledger %d: event-only membership stays the snapshot", input.LedgerSequence)
+		}
+	})
+
+	t.Run("ContractDataChanges left nil when no selected processor requires it", func(t *testing.T) {
+		ctx := context.Background()
+		dbPool, ingestStore := setupTestDB(t)
+
+		setIngestStoreValue(t, ctx, dbPool, "oldest_ingest_ledger", 100)
+		setIngestStoreValue(t, ctx, dbPool, "latest_ingest_ledger", 101)
+
+		_, err := dbPool.Exec(ctx, `INSERT INTO protocols (id, classification_status) VALUES ('noproto2', 'success') ON CONFLICT (id) DO UPDATE SET classification_status = 'success'`)
+		require.NoError(t, err)
+
+		protocolsModel := data.NewProtocolsModelMock(t)
+		protocolContractsModel := data.NewProtocolContractsModelMock(t)
+		noProc := &testCursorAdvancingProcessor{
+			testRecordingProcessor: testRecordingProcessor{id: "noproto2", ingestStore: ingestStore, requiresContractData: false},
+			dbPool:                 dbPool,
+			advanceAtSeq:           101,
+			cursorNameFunc:         utils.ProtocolHistoryCursorName,
+		}
+
+		protocolsModel.On("GetByIDs", mock.Anything, []string{"noproto2"}).Return([]data.Protocols{
+			{ID: "noproto2", ClassificationStatus: data.StatusSuccess, HistoryMigrationStatus: data.StatusNotStarted},
+		}, nil)
+		protocolsModel.On("UpdateHistoryMigrationStatus", mock.Anything, mock.Anything, []string{"noproto2"}, data.StatusInProgress).Return(nil)
+		protocolsModel.On("UpdateHistoryMigrationStatus", mock.Anything, mock.Anything, []string{"noproto2"}, data.StatusSuccess).Return(nil)
+
+		protocolContractsModel.On("GetByProtocolID", mock.Anything, "noproto2").Return([]data.ProtocolContracts{}, nil)
+
+		backend := &multiLedgerBackend{
+			ledgers: map[uint32]xdr.LedgerCloseMeta{
+				100: dummyLedgerMeta(100),
+				101: dummyLedgerMeta(101),
+			},
+		}
+
+		svc, err := NewProtocolMigrateHistoryService(ProtocolMigrateHistoryConfig{
+			DB: dbPool, LedgerBackend: backend,
+			ProtocolsModel: protocolsModel, ProtocolContractsModel: protocolContractsModel,
+			IngestStore: ingestStore, NetworkPassphrase: "Test SDF Network ; September 2015",
+			Processors: []ProtocolProcessor{noProc},
+		})
+		require.NoError(t, err)
+
+		err = svc.Run(ctx, []string{"noproto2"})
+		require.NoError(t, err)
+
+		// No selected processor requires ContractData, so extraction never ran — the
+		// field must stay nil for every recorded input.
+		require.Len(t, noProc.processedInputs, 2)
+		for _, input := range noProc.processedInputs {
+			assert.Nil(t, input.ContractDataChanges, "ledger %d: extraction must be skipped", input.LedgerSequence)
+		}
+	})
 }
 
 func TestProtocolMigrateEngine_WindowedCoalescing(t *testing.T) {
diff --git a/internal/services/protocol_processor.go b/internal/services/protocol_processor.go
index 0f880b712..42cc97db6 100644
--- a/internal/services/protocol_processor.go
+++ b/internal/services/protocol_processor.go
@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/jackc/pgx/v5"
+	"github.com/stellar/go-stellar-sdk/ingest"
 	"github.com/stellar/go-stellar-sdk/xdr"
 
 	"github.com/stellar/wallet-backend/internal/data"
@@ -39,6 +40,13 @@ type ProtocolProcessor interface {
 	// migrated with --window-size=1.
 	ProcessLedger(ctx context.Context, input ProtocolProcessorInput) error
 
+	// RequiresContractData reports whether ProcessLedger needs
+	// ProtocolProcessorInput.ContractDataChanges populated. The migration
+	// engine and live ingestion run the (heavier) ContractData extraction
+	// only when a selected processor returns true, so event-only protocols
+	// pay nothing for it.
+	RequiresContractData() bool
+
 	// Reset clears the staged sets after a window commits or hands off. The caller
 	// (engine per window; live ingestion per ledger) invokes it.
 	Reset()
@@ -69,4 +77,10 @@ type ProtocolProcessorInput struct {
 	ProtocolContracts []data.ProtocolContracts
 	// StagingMode selects which staged sets the processor builds for this ledger.
 	StagingMode StagingMode
+	// ContractDataChanges groups this ledger's ContractData ledger-entry
+	// changes (successful transactions only) by owning contract C-address.
+	// Populated only when some selected processor's RequiresContractData()
+	// returns true; nil otherwise. Processors that do not require it must
+	// ignore it.
+	ContractDataChanges map[string][]ingest.Change
 }
diff --git a/internal/services/sep41/processor.go b/internal/services/sep41/processor.go
index 25faa4595..4c1663b84 100644
--- a/internal/services/sep41/processor.go
+++ b/internal/services/sep41/processor.go
@@ -92,6 +92,10 @@ var _ services.ProtocolProcessor = (*processor)(nil)
 
 func (p *processor) ProtocolID() string { return ProtocolID }
 
+// RequiresContractData reports false: SEP-41 folds contract events only and
+// never reads ProtocolProcessorInput.ContractDataChanges.
+func (p *processor) RequiresContractData() bool { return false }
+
 // ProcessLedger consumes contract events that the indexer (or
 // ExtractContractEventsForLedger in the migration path) has already
 // extracted into the buffer. The processor never touches LedgerCloseMeta —