From d40a194ac2db36490e8dc8c86ac05ddc548eef76 Mon Sep 17 00:00:00 2001 From: jiahuihu Date: Tue, 1 Sep 2026 12:12:08 -0400 Subject: [PATCH 1/4] feat(services): expose staged state changes from protocol processors --- internal/services/ingest_test.go | 2 ++ internal/services/mocks.go | 8 ++++++++ internal/services/processor_registry_test.go | 5 +++-- internal/services/protocol_migrate_test.go | 2 ++ internal/services/protocol_processor.go | 9 +++++++++ internal/services/sep41/processor.go | 7 +++++++ 6 files changed, 31 insertions(+), 2 deletions(-) diff --git a/internal/services/ingest_test.go b/internal/services/ingest_test.go index 7cd00d8ba..663bfb859 100644 --- a/internal/services/ingest_test.go +++ b/internal/services/ingest_test.go @@ -1885,6 +1885,8 @@ func (p *testProtocolProcessor) StateChangeOrdinalBase() int64 { func (p *testProtocolProcessor) Reset() { p.stagedLedgerCount = 0 } +func (p *testProtocolProcessor) StagedStateChanges() []types.StateChange { return nil } + func (p *testProtocolProcessor) ProcessLedger(_ context.Context, input ProtocolProcessorInput) error { p.processLedgerCalls++ p.stagedLedgerCount++ diff --git a/internal/services/mocks.go b/internal/services/mocks.go index ffe78f86f..03818a4b9 100644 --- a/internal/services/mocks.go +++ b/internal/services/mocks.go @@ -277,6 +277,14 @@ func (m *ProtocolProcessorMock) Reset() { m.Called() } +func (m *ProtocolProcessorMock) StagedStateChanges() []types.StateChange { + args := m.Called() + if args.Get(0) == nil { + return nil + } + return args.Get(0).([]types.StateChange) +} + func (m *ProtocolProcessorMock) PersistHistory(ctx context.Context, dbTx pgx.Tx) error { args := m.Called(ctx, dbTx) return args.Error(0) diff --git a/internal/services/processor_registry_test.go b/internal/services/processor_registry_test.go index da3c97e37..0f644416f 100644 --- a/internal/services/processor_registry_test.go +++ b/internal/services/processor_registry_test.go @@ -17,8 +17,9 @@ type stubProcessor struct { base int64 } -func (s stubProcessor) ProtocolID() string { return s.id } -func (s stubProcessor) StateChangeOrdinalBase() int64 { return s.base } +func (s stubProcessor) ProtocolID() string { return s.id } +func (s stubProcessor) StateChangeOrdinalBase() int64 { return s.base } +func (s stubProcessor) StagedStateChanges() []types.StateChange { return nil } func registerStub(id string, base int64) { RegisterProcessor(id, func(ProtocolDeps) ProtocolProcessor { diff --git a/internal/services/protocol_migrate_test.go b/internal/services/protocol_migrate_test.go index 5d0e2c5e6..202160695 100644 --- a/internal/services/protocol_migrate_test.go +++ b/internal/services/protocol_migrate_test.go @@ -191,6 +191,8 @@ func (p *testRecordingProcessor) StateChangeOrdinalBase() int64 { func (p *testRecordingProcessor) Reset() { p.resetCount++ } +func (p *testRecordingProcessor) StagedStateChanges() []types.StateChange { return nil } + func (p *testRecordingProcessor) ProcessLedger(_ context.Context, input ProtocolProcessorInput) error { p.processedInputs = append(p.processedInputs, input) p.lastProcessed = input.LedgerSequence diff --git a/internal/services/protocol_processor.go b/internal/services/protocol_processor.go index 6e0ccf450..bea243d92 100644 --- a/internal/services/protocol_processor.go +++ b/internal/services/protocol_processor.go @@ -8,6 +8,7 @@ import ( "github.com/stellar/wallet-backend/internal/data" "github.com/stellar/wallet-backend/internal/indexer" + "github.com/stellar/wallet-backend/internal/indexer/types" ) // StagingMode tells a processor which staged sets to build. The caller stamps it @@ -73,6 +74,14 @@ type ProtocolProcessor interface { // (engine per window; live ingestion per ledger) invokes it. Reset() + // StagedStateChanges returns the history state changes accumulated by + // ProcessLedger since the last Reset, without persisting anything. Rows are + // returned before state_change_id ordinals are assigned (that happens in + // PersistHistory), so callers that never persist — transaction simulation — + // must not rely on StateChangeID. The returned slice aliases the processor's + // staged set and is invalidated by Reset(). + StagedStateChanges() []types.StateChange + // PersistHistory writes the history rows accumulated by ProcessLedger since the // last Reset, using the provided transaction. Called inside the CAS-guarded // transaction only when the cursor advances, so writes commit atomically with diff --git a/internal/services/sep41/processor.go b/internal/services/sep41/processor.go index 39f8c7d1b..eadac222b 100644 --- a/internal/services/sep41/processor.go +++ b/internal/services/sep41/processor.go @@ -323,6 +323,13 @@ func (p *processor) applyBalanceDelta(account types.AddressBytea, contractStr st // Reset clears the staged sets for the next window. ledgerNumber is intentionally // left untouched — ProcessLedger sets it each ledger. +// StagedStateChanges returns the history rows staged since the last Reset, +// without persisting. Ordinals are not assigned here (PersistHistory owns +// that), and the slice aliases the staged set. +func (p *processor) StagedStateChanges() []types.StateChange { + return p.stagedStateChanges +} + func (p *processor) Reset() { p.stagedStateChanges = nil p.stagedBalanceDelta = map[balanceKey]*big.Int{} From f5991e2542fa38d23f815d816c834dd664a99c0f Mon Sep 17 00:00:00 2001 From: jiahuihu Date: Tue, 1 Sep 2026 12:12:29 -0400 Subject: [PATCH 2/4] feat(services): preview custom SEP-41 token changes in simulateStateChanges --- internal/serve/serve.go | 3 +- internal/services/transaction_simulation.go | 97 ++++++- .../transaction_simulation_sep41_test.go | 263 ++++++++++++++++++ .../services/transaction_simulation_test.go | 8 +- 4 files changed, 353 insertions(+), 18 deletions(-) create mode 100644 internal/services/transaction_simulation_sep41_test.go diff --git a/internal/serve/serve.go b/internal/serve/serve.go index 844834abb..24074e64f 100644 --- a/internal/serve/serve.go +++ b/internal/serve/serve.go @@ -28,6 +28,7 @@ import ( "github.com/stellar/wallet-backend/internal/serve/httphandler" "github.com/stellar/wallet-backend/internal/serve/middleware" "github.com/stellar/wallet-backend/internal/services" + _ "github.com/stellar/wallet-backend/internal/services/sep41" // registers SEP-41 processor via init() for transaction simulation "github.com/stellar/wallet-backend/pkg/wbclient/auth" gqlhandler "github.com/99designs/gqlgen/graphql/handler" @@ -204,7 +205,7 @@ func initHandlerDeps(ctx context.Context, cfg Configs) (handlerDeps, error) { return handlerDeps{}, fmt.Errorf("instantiating rpc service: %w", err) } - simulationService, err := services.NewTransactionSimulationService(rpcService, cfg.NetworkPassphrase) + simulationService, err := services.NewTransactionSimulationService(rpcService, models, cfg.NetworkPassphrase) if err != nil { return handlerDeps{}, fmt.Errorf("instantiating transaction simulation service: %w", err) } diff --git a/internal/services/transaction_simulation.go b/internal/services/transaction_simulation.go index 687232a9a..c951e55f7 100644 --- a/internal/services/transaction_simulation.go +++ b/internal/services/transaction_simulation.go @@ -11,6 +11,7 @@ import ( "github.com/stellar/go-stellar-sdk/ingest" "github.com/stellar/go-stellar-sdk/xdr" + "github.com/stellar/wallet-backend/internal/data" "github.com/stellar/wallet-backend/internal/entities" "github.com/stellar/wallet-backend/internal/indexer" "github.com/stellar/wallet-backend/internal/indexer/types" @@ -51,13 +52,19 @@ type TransactionSimulationService interface { } type transactionSimulationService struct { - rpcService RPCService - ledgerIndexer *indexer.Indexer + rpcService RPCService + ledgerIndexer *indexer.Indexer + models *data.Models + networkPassphrase string } var _ TransactionSimulationService = (*transactionSimulationService)(nil) -func NewTransactionSimulationService(rpcService RPCService, networkPassphrase string) (*transactionSimulationService, error) { +// NewTransactionSimulationService builds the simulation service. models is used +// only for the read-only protocol_contracts lookup that routes contract events +// to the registered protocol processors (SEP-41); a nil models skips protocol +// processing, so previews then cover native/SAC tokens only. +func NewTransactionSimulationService(rpcService RPCService, models *data.Models, networkPassphrase string) (*transactionSimulationService, error) { if rpcService == nil { return nil, errors.New("rpcService cannot be nil") } @@ -68,8 +75,10 @@ func NewTransactionSimulationService(rpcService RPCService, networkPassphrase st return nil, fmt.Errorf("creating indexer: %w", err) } return &transactionSimulationService{ - rpcService: rpcService, - ledgerIndexer: ledgerIndexer, + rpcService: rpcService, + ledgerIndexer: ledgerIndexer, + models: models, + networkPassphrase: networkPassphrase, }, nil } @@ -379,17 +388,79 @@ func isSorobanTransaction(envelope xdr.TransactionEnvelope) bool { // stateChangesForTransaction runs a synthesized ledger transaction through the // ingestion pipeline into an in-memory buffer. It uses the same processors real -// ingestion uses, with no persistence, and returns the resulting state changes. -// -// Known limitation: this only runs the core Indexer processors, not the separate -// protocol processors. SEP-41 custom tokens are handled by internal/services/sep41, -// which the Indexer's token_transfer processor skips, and that processor is not -// wired in here yet. So previews for SEP-41 tokens miss their balance changes. -// Native and SAC token changes are covered. +// ingestion uses, with no persistence, and returns the resulting state changes: +// the core Indexer's (native/SAC tokens, contract deploys, classic effects) +// plus the registered protocol processors' (SEP-41 custom tokens). func (s *transactionSimulationService) stateChangesForTransaction(ctx context.Context, tx ingest.LedgerTransaction) ([]types.StateChange, error) { buffer := indexer.NewIndexerBuffer() if _, err := s.ledgerIndexer.ProcessLedgerTransactions(ctx, []ingest.LedgerTransaction{tx}, buffer); err != nil { return nil, fmt.Errorf("processing transaction through indexer: %w", err) } - return buffer.GetStateChanges(), nil + + protocolChanges, err := s.protocolStateChanges(ctx, tx, buffer.GetContractEvents()) + if err != nil { + return nil, err + } + return append(buffer.GetStateChanges(), protocolChanges...), nil +} + +// protocolStateChanges runs the registered protocol processors (currently +// SEP-41) over the contract events the pipeline collected, mirroring what live +// ingestion does after the indexer pass, so custom-token previews match +// history. It classifies the emitting contracts with one read-only +// protocol_contracts lookup and never persists: state changes are read from +// the processors' staged sets instead of a PersistHistory call. +// +// Known limitation: only contracts already classified in protocol_contracts +// produce rows. A token WB has not ingested and classified yet is silently +// skipped, the same ingestion-lag caveat as SAC metadata enrichment. +func (s *transactionSimulationService) protocolStateChanges(ctx context.Context, tx ingest.LedgerTransaction, contractEvents map[indexer.ContractEventKey][]xdr.ContractEvent) ([]types.StateChange, error) { + if s.models == nil || len(contractEvents) == 0 { + return nil, nil + } + eventContractIDs := distinctEventContractIDs(contractEvents) + if len(eventContractIDs) == 0 { + return nil, nil + } + committedByProtocol, err := s.models.ProtocolContracts.BatchGetByContractIDs(ctx, eventContractIDs) + if err != nil { + return nil, fmt.Errorf("resolving protocol contracts: %w", err) + } + if len(committedByProtocol) == 0 { + return nil, nil + } + + // Fresh processor instances per request: protocol processors fold state + // across ProcessLedger calls by design, so an instance must never be shared + // between simulations. + protocolProcessors, err := BuildProcessors(ProtocolDeps{ + NetworkPassphrase: s.networkPassphrase, + Models: s.models, + RPCService: s.rpcService, + }, GetAllProcessorIDs()) + if err != nil { + return nil, fmt.Errorf("building protocol processors: %w", err) + } + + var stateChanges []types.StateChange + for _, protocolProcessor := range protocolProcessors { + contracts := committedByProtocol[protocolProcessor.ProtocolID()] + if len(contracts) == 0 { + continue + } + input := ProtocolProcessorInput{ + LedgerSequence: tx.Ledger.LedgerSequence(), + LedgerCloseTime: tx.Ledger.LedgerCloseTime(), + ContractEvents: contractEvents, + ProtocolContracts: contracts, + // History rows only: a preview must never stage balance or + // allowance current-state updates. + StagingMode: StagingModeHistory, + } + if err := protocolProcessor.ProcessLedger(ctx, input); err != nil { + return nil, fmt.Errorf("running %s protocol processor: %w", protocolProcessor.ProtocolID(), err) + } + stateChanges = append(stateChanges, protocolProcessor.StagedStateChanges()...) + } + return stateChanges, nil } diff --git a/internal/services/transaction_simulation_sep41_test.go b/internal/services/transaction_simulation_sep41_test.go new file mode 100644 index 000000000..7c916fcbc --- /dev/null +++ b/internal/services/transaction_simulation_sep41_test.go @@ -0,0 +1,263 @@ +package services_test + +import ( + "context" + "crypto/rand" + "testing" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/prometheus/client_golang/prometheus" + "github.com/stellar/go-stellar-sdk/keypair" + "github.com/stellar/go-stellar-sdk/network" + "github.com/stellar/go-stellar-sdk/txnbuild" + "github.com/stellar/go-stellar-sdk/xdr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "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/entities" + "github.com/stellar/wallet-backend/internal/indexer/types" + "github.com/stellar/wallet-backend/internal/metrics" + "github.com/stellar/wallet-backend/internal/services" + _ "github.com/stellar/wallet-backend/internal/services/sep41" // registers the SEP-41 processor via init() +) + +// TestTransactionSimulationService_customSEP41Token drives the SEP-41 path of +// simulateStateChanges end to end: a simulated invocation of a custom +// (non-SAC) token that WB has classified as SEP-41 must preview the same +// balance and allowance state changes history would show. An unclassified +// contract must be silently skipped, mirroring live ingestion. +func TestTransactionSimulationService_customSEP41Token(t *testing.T) { + dbt := dbtest.Open(t) + defer dbt.Close() + ctx := context.Background() + + pool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) + require.NoError(t, err) + defer pool.Close() + + m := metrics.NewMetrics(prometheus.NewRegistry()) + models, err := data.NewModels(pool, m.DB) + require.NoError(t, err) + + contractID := randomContractID(t) + classifyAsSEP41(t, ctx, pool, models, contractID) + + holder := keypair.MustRandom().Address() + receiver := keypair.MustRandom().Address() + spender := keypair.MustRandom().Address() + + transferEvent := sep41Event(contractID, + []xdr.ScVal{scSymbol("transfer"), scAccountVal(holder), scAccountVal(receiver)}, + scI128(10_000_000), + ) + approveEvent := sep41Event(contractID, + []xdr.ScVal{scSymbol("approve"), scAccountVal(holder), scAccountVal(spender)}, + scVec(scI128(5_000_000), scU32(3_000_000)), + ) + + txXDR := invokeContractTxXDR(t, contractID) + rpcMock := &services.RPCServiceMock{} + rpcMock.On("SimulateTransaction", txXDR, entities.RPCResourceConfig{}). + Return(entities.RPCSimulateTransactionResult{ + LatestLedger: 2900148, + MinResourceFee: "100", + Events: []string{diagnosticB64(t, transferEvent), diagnosticB64(t, approveEvent)}, + }, nil).Once() + + svc, err := services.NewTransactionSimulationService(rpcMock, models, network.TestNetworkPassphrase) + require.NoError(t, err) + + result, err := svc.SimulateStateChanges(ctx, txXDR) + require.NoError(t, err) + + tokenStrkey := contractIDStrkey(t, contractID) + tokenChanges := changesForToken(result.StateChanges, tokenStrkey) + require.Len(t, tokenChanges, 3, "expected debit + credit + allowance for the custom token") + + byReason := map[types.StateChangeReason]types.StateChange{} + for _, sc := range tokenChanges { + byReason[sc.StateChangeReason] = sc + } + + debit, ok := byReason[types.StateChangeReasonDebit] + require.True(t, ok, "expected a DEBIT for the holder") + assert.Equal(t, holder, string(debit.AccountID)) + assert.Equal(t, "10000000", debit.Amount.String) + assert.Equal(t, types.StateChangeCategoryBalance, debit.StateChangeCategory) + + credit, ok := byReason[types.StateChangeReasonCredit] + require.True(t, ok, "expected a CREDIT for the receiver") + assert.Equal(t, receiver, string(credit.AccountID)) + assert.Equal(t, "10000000", credit.Amount.String) + + allowance, ok := byReason[types.StateChangeReasonUpdate] + require.True(t, ok, "expected an ALLOWANCE update for the holder") + assert.Equal(t, types.StateChangeCategoryAllowance, allowance.StateChangeCategory) + assert.Equal(t, holder, string(allowance.AccountID)) + assert.Equal(t, spender, allowance.SpenderAccountID.String()) + assert.Equal(t, "5000000", allowance.Amount.String) + // In-memory staged rows carry the value as uint32; the float64 shape only + // appears after a JSONB round-trip, which simulated rows never take. + assert.Equal(t, uint32(3_000_000), allowance.KeyValue["live_until_ledger"]) + + rpcMock.AssertExpectations(t) + + t.Run("unclassified contract is silently skipped", func(t *testing.T) { + unknownID := randomContractID(t) + unknownEvent := sep41Event(unknownID, + []xdr.ScVal{scSymbol("transfer"), scAccountVal(holder), scAccountVal(receiver)}, + scI128(1), + ) + unknownTxXDR := invokeContractTxXDR(t, unknownID) + rpcMock := &services.RPCServiceMock{} + rpcMock.On("SimulateTransaction", unknownTxXDR, entities.RPCResourceConfig{}). + Return(entities.RPCSimulateTransactionResult{ + LatestLedger: 2900149, + MinResourceFee: "100", + Events: []string{diagnosticB64(t, unknownEvent)}, + }, nil).Once() + + svc, err := services.NewTransactionSimulationService(rpcMock, models, network.TestNetworkPassphrase) + require.NoError(t, err) + + result, err := svc.SimulateStateChanges(ctx, unknownTxXDR) + require.NoError(t, err) + assert.Empty(t, changesForToken(result.StateChanges, contractIDStrkey(t, unknownID)), + "an unclassified contract must produce no token state changes") + rpcMock.AssertExpectations(t) + }) +} + +// classifyAsSEP41 registers the contract in protocol_wasms + protocol_contracts, +// the same rows the SEP-41 validator commits at classification time. +func classifyAsSEP41(t *testing.T, ctx context.Context, pool *pgxpool.Pool, models *data.Models, contractID xdr.ContractId) { + t.Helper() + wasmHash := make([]byte, 32) + _, err := rand.Read(wasmHash) + require.NoError(t, err) + + dbTx, err := pool.Begin(ctx) + require.NoError(t, err) + defer func() { _ = dbTx.Rollback(ctx) }() //nolint:errcheck // no-op after commit + + protocolID := "SEP41" + _, err = dbTx.Exec(ctx, `INSERT INTO protocols (id) VALUES ($1) ON CONFLICT (id) DO NOTHING`, protocolID) + require.NoError(t, err) + require.NoError(t, models.ProtocolWasms.BatchInsert(ctx, dbTx, []data.ProtocolWasms{{ + WasmHash: types.HashBytea(hexString(wasmHash)), + ProtocolID: &protocolID, + }})) + require.NoError(t, models.ProtocolContracts.BatchInsert(ctx, dbTx, []data.ProtocolContracts{{ + ContractID: types.HashBytea(hexString(contractID[:])), + WasmHash: types.HashBytea(hexString(wasmHash)), + }})) + require.NoError(t, dbTx.Commit(ctx)) +} + +func changesForToken(stateChanges []types.StateChange, tokenStrkey string) []types.StateChange { + var out []types.StateChange + for _, sc := range stateChanges { + if sc.TokenID.String() == tokenStrkey { + out = append(out, sc) + } + } + return out +} + +func randomContractID(t *testing.T) xdr.ContractId { + t.Helper() + var id xdr.ContractId + _, err := rand.Read(id[:]) + require.NoError(t, err) + return id +} + +func contractIDStrkey(t *testing.T, contractID xdr.ContractId) string { + t.Helper() + addr := xdr.ScAddress{Type: xdr.ScAddressTypeScAddressTypeContract, ContractId: &contractID} + s, err := addr.String() + require.NoError(t, err) + return s +} + +// sep41Event builds a contract event in the shape internal/services/sep41 +// parses: topic[0] the function symbol, address topics, and the amount payload. +func sep41Event(contractID xdr.ContractId, topics []xdr.ScVal, payload xdr.ScVal) xdr.ContractEvent { + return xdr.ContractEvent{ + Type: xdr.ContractEventTypeContract, + ContractId: &contractID, + Body: xdr.ContractEventBody{ + V: 0, + V0: &xdr.ContractEventV0{Topics: topics, Data: payload}, + }, + } +} + +func diagnosticB64(t *testing.T, event xdr.ContractEvent) string { + t.Helper() + b64, err := xdr.MarshalBase64(xdr.DiagnosticEvent{InSuccessfulContractCall: true, Event: event}) + require.NoError(t, err) + return b64 +} + +func invokeContractTxXDR(t *testing.T, contractID xdr.ContractId) string { + t.Helper() + src := txnbuild.SimpleAccount{AccountID: keypair.MustRandom().Address(), Sequence: 1} + tx, err := txnbuild.NewTransaction(txnbuild.TransactionParams{ + SourceAccount: &src, + Operations: []txnbuild.Operation{&txnbuild.InvokeHostFunction{ + HostFunction: xdr.HostFunction{ + Type: xdr.HostFunctionTypeHostFunctionTypeInvokeContract, + InvokeContract: &xdr.InvokeContractArgs{ + ContractAddress: xdr.ScAddress{Type: xdr.ScAddressTypeScAddressTypeContract, ContractId: &contractID}, + FunctionName: "transfer", + }, + }, + }}, + BaseFee: txnbuild.MinBaseFee, + Preconditions: txnbuild.Preconditions{TimeBounds: txnbuild.NewTimeout(300)}, + IncrementSequenceNum: true, + }) + require.NoError(t, err) + b64, err := tx.Base64() + require.NoError(t, err) + return b64 +} + +func scSymbol(s string) xdr.ScVal { + sym := xdr.ScSymbol(s) + return xdr.ScVal{Type: xdr.ScValTypeScvSymbol, Sym: &sym} +} + +func scAccountVal(address string) xdr.ScVal { + accountID := xdr.MustAddress(address) + addr := xdr.ScAddress{Type: xdr.ScAddressTypeScAddressTypeAccount, AccountId: &accountID} + return xdr.ScVal{Type: xdr.ScValTypeScvAddress, Address: &addr} +} + +func scI128(amount int64) xdr.ScVal { + return xdr.ScVal{Type: xdr.ScValTypeScvI128, I128: &xdr.Int128Parts{Hi: 0, Lo: xdr.Uint64(amount)}} +} + +func scU32(v uint32) xdr.ScVal { + u := xdr.Uint32(v) + return xdr.ScVal{Type: xdr.ScValTypeScvU32, U32: &u} +} + +func scVec(vals ...xdr.ScVal) xdr.ScVal { + vec := xdr.ScVec(vals) + vecPtr := &vec + return xdr.ScVal{Type: xdr.ScValTypeScvVec, Vec: &vecPtr} +} + +func hexString(b []byte) string { + const hexdigits = "0123456789abcdef" + out := make([]byte, 0, len(b)*2) + for _, c := range b { + out = append(out, hexdigits[c>>4], hexdigits[c&0x0f]) + } + return string(out) +} diff --git a/internal/services/transaction_simulation_test.go b/internal/services/transaction_simulation_test.go index a8b1671e0..8192a5934 100644 --- a/internal/services/transaction_simulation_test.go +++ b/internal/services/transaction_simulation_test.go @@ -20,7 +20,7 @@ import ( ) func TestTransactionSimulationService_SimulateStateChanges_errors(t *testing.T) { - svc, err := NewTransactionSimulationService(&RPCServiceMock{}, network.TestNetworkPassphrase) + svc, err := NewTransactionSimulationService(&RPCServiceMock{}, nil, network.TestNetworkPassphrase) require.NoError(t, err) ctx := context.Background() @@ -68,7 +68,7 @@ func TestTransactionSimulationService_SimulateStateChanges_errors(t *testing.T) rpcMock := &RPCServiceMock{} rpcMock.On("SimulateTransaction", mock.Anything, mock.Anything). Return(entities.RPCSimulateTransactionResult{Error: "contract trapped"}, nil).Once() - errSvc, err := NewTransactionSimulationService(rpcMock, network.TestNetworkPassphrase) + errSvc, err := NewTransactionSimulationService(rpcMock, nil, network.TestNetworkPassphrase) require.NoError(t, err) _, err = errSvc.SimulateStateChanges(ctx, nativeSACTransferXDR(t, keypair.MustRandom().Address())) @@ -109,7 +109,7 @@ func TestTransactionSimulationService_SimulateStateChanges_soroban(t *testing.T) Events: []string{diagnosticB64}, }, nil).Once() - svc, err := NewTransactionSimulationService(rpcMock, network.TestNetworkPassphrase) + svc, err := NewTransactionSimulationService(rpcMock, nil, network.TestNetworkPassphrase) require.NoError(t, err) result, err := svc.SimulateStateChanges(context.Background(), txXDR) @@ -235,7 +235,7 @@ func nativeSACTransferXDR(t *testing.T, sourceAccount string) string { // token-transfer processor must emit a DEBIT for the sender and a CREDIT for // the receiver. func TestTransactionSimulationService_walkingSkeleton(t *testing.T) { - svc, err := NewTransactionSimulationService(&RPCServiceMock{}, network.TestNetworkPassphrase) + svc, err := NewTransactionSimulationService(&RPCServiceMock{}, nil, network.TestNetworkPassphrase) require.NoError(t, err) from := keypair.MustRandom().Address() From ca6722de42e9b2f74a0a678048f2ed55f3a22a84 Mon Sep 17 00:00:00 2001 From: jiahuihu Date: Thu, 3 Sep 2026 16:16:13 -0400 Subject: [PATCH 3/4] fix(services): overlay same-transaction wasm bindings in simulation classification --- internal/services/sep41/processor.go | 4 +- internal/services/transaction_simulation.go | 53 ++++++-- .../transaction_simulation_sep41_test.go | 114 +++++++++++++++++- 3 files changed, 155 insertions(+), 16 deletions(-) diff --git a/internal/services/sep41/processor.go b/internal/services/sep41/processor.go index eadac222b..e93708f8d 100644 --- a/internal/services/sep41/processor.go +++ b/internal/services/sep41/processor.go @@ -321,8 +321,6 @@ func (p *processor) applyBalanceDelta(account types.AddressBytea, contractStr st p.stagedBalanceLedger[key] = p.ledgerNumber } -// Reset clears the staged sets for the next window. ledgerNumber is intentionally -// left untouched — ProcessLedger sets it each ledger. // StagedStateChanges returns the history rows staged since the last Reset, // without persisting. Ordinals are not assigned here (PersistHistory owns // that), and the slice aliases the staged set. @@ -330,6 +328,8 @@ func (p *processor) StagedStateChanges() []types.StateChange { return p.stagedStateChanges } +// Reset clears the staged sets for the next window. ledgerNumber is intentionally +// left untouched — ProcessLedger sets it each ledger. func (p *processor) Reset() { p.stagedStateChanges = nil p.stagedBalanceDelta = map[balanceKey]*big.Int{} diff --git a/internal/services/transaction_simulation.go b/internal/services/transaction_simulation.go index c951e55f7..f88504bb1 100644 --- a/internal/services/transaction_simulation.go +++ b/internal/services/transaction_simulation.go @@ -397,7 +397,7 @@ func (s *transactionSimulationService) stateChangesForTransaction(ctx context.Co return nil, fmt.Errorf("processing transaction through indexer: %w", err) } - protocolChanges, err := s.protocolStateChanges(ctx, tx, buffer.GetContractEvents()) + protocolChanges, err := s.protocolStateChanges(ctx, tx, buffer.GetContractEvents(), buffer.GetProtocolContracts()) if err != nil { return nil, err } @@ -408,13 +408,16 @@ func (s *transactionSimulationService) stateChangesForTransaction(ctx context.Co // SEP-41) over the contract events the pipeline collected, mirroring what live // ingestion does after the indexer pass, so custom-token previews match // history. It classifies the emitting contracts with one read-only -// protocol_contracts lookup and never persists: state changes are read from -// the processors' staged sets instead of a PersistHistory call. +// protocol_contracts lookup, overlays the contract-to-wasm bindings the +// simulated transaction itself changed (bufferedContracts, mirroring +// getEffectiveProtocolContracts in live ingestion), and never persists: state +// changes are read from the processors' staged sets instead of a +// PersistHistory call. // -// Known limitation: only contracts already classified in protocol_contracts -// produce rows. A token WB has not ingested and classified yet is silently -// skipped, the same ingestion-lag caveat as SAC metadata enrichment. -func (s *transactionSimulationService) protocolStateChanges(ctx context.Context, tx ingest.LedgerTransaction, contractEvents map[indexer.ContractEventKey][]xdr.ContractEvent) ([]types.StateChange, error) { +// Known limitation: classification comes only from committed protocol_wasms +// rows; simulation runs no validators. A wasm WB has never classified is +// silently skipped, the same ingestion-lag caveat as SAC metadata enrichment. +func (s *transactionSimulationService) protocolStateChanges(ctx context.Context, tx ingest.LedgerTransaction, contractEvents map[indexer.ContractEventKey][]xdr.ContractEvent, bufferedContracts map[string]data.ProtocolContracts) ([]types.StateChange, error) { if s.models == nil || len(contractEvents) == 0 { return nil, nil } @@ -426,7 +429,20 @@ func (s *transactionSimulationService) protocolStateChanges(ctx context.Context, if err != nil { return nil, fmt.Errorf("resolving protocol contracts: %w", err) } - if len(committedByProtocol) == 0 { + + // A binding changed by the simulated transaction itself must shadow the + // committed row: a contract upgraded away from a protocol emits no rows, and + // a contract bound to an already-classified wasm emits rows even without a + // committed protocol_contracts entry. Classification for the buffered wasm + // hashes comes from committed protocol_wasms. + var classification map[types.HashBytea]string + if len(bufferedContracts) > 0 { + classification, err = s.models.ProtocolWasms.GetClassifiedByHashes(ctx, s.models.DB, distinctWasmHashes(bufferedContracts)) + if err != nil { + return nil, fmt.Errorf("classifying simulated wasm bindings: %w", err) + } + } + if len(committedByProtocol) == 0 && len(classification) == 0 { return nil, nil } @@ -444,7 +460,7 @@ func (s *transactionSimulationService) protocolStateChanges(ctx context.Context, var stateChanges []types.StateChange for _, protocolProcessor := range protocolProcessors { - contracts := committedByProtocol[protocolProcessor.ProtocolID()] + contracts := getEffectiveProtocolContracts(protocolProcessor.ProtocolID(), committedByProtocol[protocolProcessor.ProtocolID()], bufferedContracts, classification) if len(contracts) == 0 { continue } @@ -457,6 +473,10 @@ func (s *transactionSimulationService) protocolStateChanges(ctx context.Context, // allowance current-state updates. StagingMode: StagingModeHistory, } + // The lifecycle contract makes the caller reset before folding + // (ProtocolProcessor.Reset); fresh construction is not a documented + // reset guarantee for arbitrary registered factories. + protocolProcessor.Reset() if err := protocolProcessor.ProcessLedger(ctx, input); err != nil { return nil, fmt.Errorf("running %s protocol processor: %w", protocolProcessor.ProtocolID(), err) } @@ -464,3 +484,18 @@ func (s *transactionSimulationService) protocolStateChanges(ctx context.Context, } return stateChanges, nil } + +// distinctWasmHashes returns the deduplicated wasm hashes of the given +// contract bindings. +func distinctWasmHashes(contracts map[string]data.ProtocolContracts) []types.HashBytea { + seen := make(map[types.HashBytea]struct{}, len(contracts)) + hashes := make([]types.HashBytea, 0, len(contracts)) + for _, contract := range contracts { + if _, ok := seen[contract.WasmHash]; ok { + continue + } + seen[contract.WasmHash] = struct{}{} + hashes = append(hashes, contract.WasmHash) + } + return hashes +} diff --git a/internal/services/transaction_simulation_sep41_test.go b/internal/services/transaction_simulation_sep41_test.go index 7c916fcbc..35df07377 100644 --- a/internal/services/transaction_simulation_sep41_test.go +++ b/internal/services/transaction_simulation_sep41_test.go @@ -43,7 +43,7 @@ func TestTransactionSimulationService_customSEP41Token(t *testing.T) { require.NoError(t, err) contractID := randomContractID(t) - classifyAsSEP41(t, ctx, pool, models, contractID) + classifiedWasm := classifyAsSEP41(t, ctx, pool, models, contractID) holder := keypair.MustRandom().Address() receiver := keypair.MustRandom().Address() @@ -105,6 +105,63 @@ func TestTransactionSimulationService_customSEP41Token(t *testing.T) { rpcMock.AssertExpectations(t) + // The two cases below cover the same-transaction binding overlay: like live + // ingestion's getEffectiveProtocolContracts, a contract whose executable the + // simulated transaction itself changes must be classified by its NEW wasm, + // not its committed row. + t.Run("contract upgraded away this transaction emits no rows", func(t *testing.T) { + var unclassifiedWasm [32]byte + _, err := rand.Read(unclassifiedWasm[:]) + require.NoError(t, err) + + upgradeTxXDR := invokeContractTxXDR(t, contractID) + rpcMock := &services.RPCServiceMock{} + rpcMock.On("SimulateTransaction", upgradeTxXDR, entities.RPCResourceConfig{}). + Return(entities.RPCSimulateTransactionResult{ + LatestLedger: 2900150, + MinResourceFee: "100", + Events: []string{diagnosticB64(t, transferEvent)}, + StateChanges: []entities.RPCSimulateStateChange{instanceBindingStateChange(t, contractID, unclassifiedWasm)}, + }, nil).Once() + + svc, err := services.NewTransactionSimulationService(rpcMock, models, network.TestNetworkPassphrase) + require.NoError(t, err) + + result, err := svc.SimulateStateChanges(ctx, upgradeTxXDR) + require.NoError(t, err) + assert.Empty(t, changesForToken(result.StateChanges, tokenStrkey), + "a contract rebound to an unclassified wasm in this transaction must not use its stale committed classification") + rpcMock.AssertExpectations(t) + }) + + t.Run("contract bound to an already-classified wasm this transaction emits rows", func(t *testing.T) { + freshID := randomContractID(t) + freshTransfer := sep41Event(freshID, + []xdr.ScVal{scSymbol("transfer"), scAccountVal(holder), scAccountVal(receiver)}, + scI128(2_000_000), + ) + + bindTxXDR := invokeContractTxXDR(t, freshID) + rpcMock := &services.RPCServiceMock{} + rpcMock.On("SimulateTransaction", bindTxXDR, entities.RPCResourceConfig{}). + Return(entities.RPCSimulateTransactionResult{ + LatestLedger: 2900151, + MinResourceFee: "100", + Events: []string{diagnosticB64(t, freshTransfer)}, + StateChanges: []entities.RPCSimulateStateChange{instanceBindingStateChange(t, freshID, classifiedWasm)}, + }, nil).Once() + + svc, err := services.NewTransactionSimulationService(rpcMock, models, network.TestNetworkPassphrase) + require.NoError(t, err) + + result, err := svc.SimulateStateChanges(ctx, bindTxXDR) + require.NoError(t, err) + freshChanges := changesForToken(result.StateChanges, contractIDStrkey(t, freshID)) + assert.Len(t, freshChanges, 2, + "a contract with no committed row but bound this transaction to a classified wasm must emit debit + credit") + rpcMock.AssertExpectations(t) + }) + t.Run("unclassified contract is silently skipped", func(t *testing.T) { unknownID := randomContractID(t) unknownEvent := sep41Event(unknownID, @@ -132,12 +189,14 @@ func TestTransactionSimulationService_customSEP41Token(t *testing.T) { } // classifyAsSEP41 registers the contract in protocol_wasms + protocol_contracts, -// the same rows the SEP-41 validator commits at classification time. -func classifyAsSEP41(t *testing.T, ctx context.Context, pool *pgxpool.Pool, models *data.Models, contractID xdr.ContractId) { +// the same rows the SEP-41 validator commits at classification time. It returns +// the classified wasm hash so tests can bind other contracts to it. +func classifyAsSEP41(t *testing.T, ctx context.Context, pool *pgxpool.Pool, models *data.Models, contractID xdr.ContractId) [32]byte { t.Helper() - wasmHash := make([]byte, 32) - _, err := rand.Read(wasmHash) + var wasmArr [32]byte + _, err := rand.Read(wasmArr[:]) require.NoError(t, err) + wasmHash := wasmArr[:] dbTx, err := pool.Begin(ctx) require.NoError(t, err) @@ -155,6 +214,51 @@ func classifyAsSEP41(t *testing.T, ctx context.Context, pool *pgxpool.Pool, mode WasmHash: types.HashBytea(hexString(wasmHash)), }})) require.NoError(t, dbTx.Commit(ctx)) + return wasmArr +} + +// instanceBindingStateChange builds the RPC state-change entry a simulation +// returns when a transaction creates or upgrades a contract instance: a +// ContractData instance entry whose executable points at wasmHash. The +// indexer's protocol-contracts processor turns it into a buffered +// contract-to-wasm binding. +func instanceBindingStateChange(t *testing.T, contractID xdr.ContractId, wasmHash [32]byte) entities.RPCSimulateStateChange { + t.Helper() + hash := xdr.Hash(wasmHash) + contractAddr := xdr.ScAddress{Type: xdr.ScAddressTypeScAddressTypeContract, ContractId: &contractID} + + entryB64, err := xdr.MarshalBase64(xdr.LedgerEntry{ + Data: xdr.LedgerEntryData{ + Type: xdr.LedgerEntryTypeContractData, + ContractData: &xdr.ContractDataEntry{ + Contract: contractAddr, + Key: xdr.ScVal{Type: xdr.ScValTypeScvLedgerKeyContractInstance}, + Durability: xdr.ContractDataDurabilityPersistent, + Val: xdr.ScVal{ + Type: xdr.ScValTypeScvContractInstance, + Instance: &xdr.ScContractInstance{ + Executable: xdr.ContractExecutable{ + Type: xdr.ContractExecutableTypeContractExecutableWasm, + WasmHash: &hash, + }, + }, + }, + }, + }, + }) + require.NoError(t, err) + + keyB64, err := xdr.MarshalBase64(xdr.LedgerKey{ + Type: xdr.LedgerEntryTypeContractData, + ContractData: &xdr.LedgerKeyContractData{ + Contract: contractAddr, + Key: xdr.ScVal{Type: xdr.ScValTypeScvLedgerKeyContractInstance}, + Durability: xdr.ContractDataDurabilityPersistent, + }, + }) + require.NoError(t, err) + + return entities.RPCSimulateStateChange{Type: "created", Key: keyB64, After: &entryB64} } func changesForToken(stateChanges []types.StateChange, tokenStrkey string) []types.StateChange { From b68e1bfb43d160facb9aa6c8bb9e3b275040415e Mon Sep 17 00:00:00 2001 From: jiahuihu Date: Mon, 14 Sep 2026 12:56:30 -0400 Subject: [PATCH 4/4] test(services): construct the simulation service with models in the over-bid fee test --- internal/services/transaction_simulation_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/services/transaction_simulation_test.go b/internal/services/transaction_simulation_test.go index 8192a5934..93650789d 100644 --- a/internal/services/transaction_simulation_test.go +++ b/internal/services/transaction_simulation_test.go @@ -190,7 +190,7 @@ func TestTransactionSimulationService_SimulateStateChanges_feeOverBid(t *testing MinResourceFee: "100", Events: []string{diagnosticB64}, }, nil).Once() - svc, err := NewTransactionSimulationService(rpcMock, network.TestNetworkPassphrase) + svc, err := NewTransactionSimulationService(rpcMock, nil, network.TestNetworkPassphrase) require.NoError(t, err) result, err := svc.SimulateStateChanges(context.Background(), paddedXDR)