From b875cd08ba9e8d8dd6d104f2d06d2c96089b19e4 Mon Sep 17 00:00:00 2001 From: jiahuihu Date: Fri, 21 Aug 2026 17:27:31 -0400 Subject: [PATCH 1/2] fix(ingest): verify SAC before recording contract balances --- internal/data/contract_tokens.go | 27 ++++ internal/data/mocks.go | 8 + internal/ingest/ingest.go | 1 + internal/services/checkpoint.go | 68 +++++--- internal/services/checkpoint_test.go | 185 +++++++--------------- internal/services/token_ingestion.go | 36 +++++ internal/services/token_ingestion_test.go | 75 +++++++++ 7 files changed, 247 insertions(+), 153 deletions(-) diff --git a/internal/data/contract_tokens.go b/internal/data/contract_tokens.go index df557f8f4..8c5cfbd21 100644 --- a/internal/data/contract_tokens.go +++ b/internal/data/contract_tokens.go @@ -30,6 +30,12 @@ func DeterministicContractID(contractID string) uuid.UUID { // ContractModelInterface defines the interface for contract token operations. type ContractModelInterface interface { GetExisting(ctx context.Context, dbTx pgx.Tx, contractIDs []string) ([]string, error) + // GetExistingSACByID returns which of the given deterministic contract UUIDs + // exist in contract_tokens as a SAC (type='SAC'). Used to associate SAC balances + // only with contracts confirmed to be SACs via their instance entry (see + // sac.AssetFromContractData in the sac_instances processor), since a balance + // entry's shape alone does not identify the contract as a SAC. + GetExistingSACByID(ctx context.Context, dbTx pgx.Tx, ids []uuid.UUID) ([]uuid.UUID, error) // GetSACContractsMissingMetadata returns the contract_id of every SAC-typed row // whose name is still NULL — a balance-derived SAC row created with ledger-derived // defaults whose RPC enrichment has not yet succeeded. Enrichment populates name @@ -89,6 +95,27 @@ func (m *ContractModel) GetExisting(ctx context.Context, dbTx pgx.Tx, contractID return ids, nil } +// GetExistingSACByID returns which of the given deterministic contract UUIDs +// exist in contract_tokens verified as SAC (type='SAC'). +func (m *ContractModel) GetExistingSACByID(ctx context.Context, dbTx pgx.Tx, ids []uuid.UUID) ([]uuid.UUID, error) { + if len(ids) == 0 { + return nil, nil + } + + const query = `SELECT id FROM contract_tokens WHERE id = ANY($1) AND type = $2` + + start := time.Now() + found, err := db.QueryMany[uuid.UUID](ctx, dbTx, query, ids, string(types.ContractTypeSAC)) + duration := time.Since(start).Seconds() + m.Metrics.QueryDuration.WithLabelValues("GetExistingSACByID", "contract_tokens").Observe(duration) + m.Metrics.QueriesTotal.WithLabelValues("GetExistingSACByID", "contract_tokens").Inc() + if err != nil { + m.Metrics.QueryErrors.WithLabelValues("GetExistingSACByID", "contract_tokens", utils.GetDBErrorType(err)).Inc() + return nil, fmt.Errorf("querying existing SAC contract IDs: %w", err) + } + return found, nil +} + // GetSACContractsMissingMetadata returns the contract_id of every SAC-typed // contract_tokens row whose name is still NULL. See the interface godoc for why // name (not code) is the convergent staleness marker. diff --git a/internal/data/mocks.go b/internal/data/mocks.go index cce1ac645..f7847ffb3 100644 --- a/internal/data/mocks.go +++ b/internal/data/mocks.go @@ -42,6 +42,14 @@ func (m *ContractModelMock) GetExisting(ctx context.Context, dbTx pgx.Tx, contra return args.Get(0).([]string), args.Error(1) } +func (m *ContractModelMock) GetExistingSACByID(ctx context.Context, dbTx pgx.Tx, ids []uuid.UUID) ([]uuid.UUID, error) { + args := m.Called(ctx, dbTx, ids) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).([]uuid.UUID), args.Error(1) +} + func (m *ContractModelMock) GetSACContractsMissingMetadata(ctx context.Context, q db.Querier) ([]string, error) { args := m.Called(ctx, q) if args.Get(0) == nil { diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 21130d05e..3e2c39a5b 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -233,6 +233,7 @@ func setupDeps(ctx context.Context, cfg Configs) (services.IngestService, func() TrustlineBalanceModel: models.TrustlineBalance, NativeBalanceModel: models.NativeBalance, SACBalanceModel: models.SACBalance, + ContractModel: models.Contract, LiquidityPoolModel: models.LiquidityPool, LiquidityPoolBalanceModel: models.LiquidityPoolBalance, NetworkPassphrase: cfg.NetworkPassphrase, diff --git a/internal/services/checkpoint.go b/internal/services/checkpoint.go index ed06ffe9f..6ead428ff 100644 --- a/internal/services/checkpoint.go +++ b/internal/services/checkpoint.go @@ -140,12 +140,10 @@ func newCheckpointData() checkpointData { type batch struct { trustlineBalances []wbdata.TrustlineBalance nativeBalances []wbdata.NativeBalance - sacBalances []wbdata.SACBalance liquidityPools []wbdata.LiquidityPool liquidityPoolBalances []wbdata.LiquidityPoolBalance trustlineBalanceModel wbdata.TrustlineBalanceModelInterface nativeBalanceModel wbdata.NativeBalanceModelInterface - sacBalanceModel wbdata.SACBalanceModelInterface liquidityPoolModel wbdata.LiquidityPoolModelInterface liquidityPoolBalanceModel wbdata.LiquidityPoolBalanceModelInterface } @@ -153,19 +151,16 @@ type batch struct { func newBatch( trustlineBalanceModel wbdata.TrustlineBalanceModelInterface, nativeBalanceModel wbdata.NativeBalanceModelInterface, - sacBalanceModel wbdata.SACBalanceModelInterface, liquidityPoolModel wbdata.LiquidityPoolModelInterface, liquidityPoolBalanceModel wbdata.LiquidityPoolBalanceModelInterface, ) *batch { return &batch{ trustlineBalances: make([]wbdata.TrustlineBalance, 0, flushBatchSize), nativeBalances: make([]wbdata.NativeBalance, 0, flushBatchSize), - sacBalances: make([]wbdata.SACBalance, 0, flushBatchSize), liquidityPools: make([]wbdata.LiquidityPool, 0, flushBatchSize), liquidityPoolBalances: make([]wbdata.LiquidityPoolBalance, 0, flushBatchSize), trustlineBalanceModel: trustlineBalanceModel, nativeBalanceModel: nativeBalanceModel, - sacBalanceModel: sacBalanceModel, liquidityPoolModel: liquidityPoolModel, liquidityPoolBalanceModel: liquidityPoolBalanceModel, } @@ -196,10 +191,6 @@ func (b *batch) addNativeBalance(accountAddress string, balance, minimumBalance, }) } -func (b *batch) addSACBalance(sacBalance wbdata.SACBalance) { - b.sacBalances = append(b.sacBalances, sacBalance) -} - func (b *batch) addLiquidityPool(pool wbdata.LiquidityPool) { b.liquidityPools = append(b.liquidityPools, pool) } @@ -221,9 +212,6 @@ func (b *batch) flush(ctx context.Context, dbTx pgx.Tx) error { if err := b.nativeBalanceModel.BatchCopy(ctx, dbTx, b.nativeBalances); err != nil { return fmt.Errorf("batch inserting native balances: %w", err) } - if err := b.sacBalanceModel.BatchCopy(ctx, dbTx, b.sacBalances); err != nil { - return fmt.Errorf("batch inserting SAC balances: %w", err) - } if err := b.liquidityPoolModel.BatchCopy(ctx, dbTx, b.liquidityPools); err != nil { return fmt.Errorf("batch inserting liquidity pools: %w", err) } @@ -234,14 +222,13 @@ func (b *batch) flush(ctx context.Context, dbTx pgx.Tx) error { } func (b *batch) count() int { - return len(b.trustlineBalances) + len(b.nativeBalances) + len(b.sacBalances) + + return len(b.trustlineBalances) + len(b.nativeBalances) + len(b.liquidityPools) + len(b.liquidityPoolBalances) } func (b *batch) reset() { b.trustlineBalances = b.trustlineBalances[:0] b.nativeBalances = b.nativeBalances[:0] - b.sacBalances = b.sacBalances[:0] b.liquidityPools = b.liquidityPools[:0] b.liquidityPoolBalances = b.liquidityPoolBalances[:0] } @@ -263,6 +250,13 @@ type checkpointProcessor struct { // call; PopulateFromCheckpoint fetches metadata for these IDs in a short // follow-up transaction after the load commits. pendingSACMetadata []string + // pendingSACBalances holds SAC-shaped balance entries seen during the scan. They + // are not written to the batch during the pass because a balance entry's shape does + // not by itself identify its contract as a SAC. finalize keeps only those whose + // contract was confirmed as a SAC via its instance entry (uniqueContractTokens with + // type=SAC), then copies them in — before the deferred fk_contract_token is checked + // at COMMIT. + pendingSACBalances []wbdata.SACBalance } // PopulateFromCheckpoint performs initial cache population from Stellar history archive. @@ -305,7 +299,7 @@ func (s *checkpointService) PopulateFromCheckpoint(ctx context.Context, checkpoi dbTx: dbTx, checkpointLedger: checkpointLedger, data: newCheckpointData(), - batch: newBatch(s.trustlineBalanceModel, s.nativeBalanceModel, s.sacBalanceModel, s.liquidityPoolModel, s.liquidityPoolBalanceModel), + batch: newBatch(s.trustlineBalanceModel, s.nativeBalanceModel, s.liquidityPoolModel, s.liquidityPoolBalanceModel), wasmClassifications: make(map[xdr.Hash]types.ContractType), contractAddressesByWasmHash: make(map[xdr.Hash][]xdr.Hash), startTime: time.Now(), @@ -519,17 +513,14 @@ func (p *checkpointProcessor) processEntry(change ingest.Change) { _, _, ok := sac.ContractBalanceFromContractData(*change.Post, p.service.networkPassphrase) if ok { + // Shape matches a SAC balance, but that alone does not identify the + // contract as a SAC. Defer the balance to finalize, which keeps it + // only if the contract was confirmed via its instance entry. Do NOT + // create a contract_tokens row from the balance shape: the instance + // entry is the authoritative source of a contract's type and metadata. contractUUID := wbdata.DeterministicContractID(contractAddressStr) - if _, exists := p.data.uniqueContractTokens[contractUUID]; !exists { - p.data.uniqueContractTokens[contractUUID] = &wbdata.Contract{ - ID: contractUUID, - ContractID: contractAddressStr, - Type: string(types.ContractTypeSAC), - } - } - balanceStr, authorized, clawback := p.service.extractSACBalanceFields(contractDataEntry.Val) - p.batch.addSACBalance(wbdata.SACBalance{ + p.pendingSACBalances = append(p.pendingSACBalances, wbdata.SACBalance{ AccountID: types.AddressBytea(holderAddress), ContractID: contractUUID, Balance: balanceStr, @@ -594,6 +585,35 @@ func (p *checkpointProcessor) finalize(ctx context.Context, dbTx pgx.Tx) error { return fmt.Errorf("storing tokens in postgres: %w", err) } + // Persist SAC balances, but only for contracts confirmed as SAC via their + // instance entry. The scan populated uniqueContractTokens; a balance whose + // contract is not among them is not a SAC balance and is dropped. This also + // guarantees every retained balance has a contract_tokens parent, so the + // deferred fk_contract_token holds at COMMIT. + verifiedSAC := make(map[uuid.UUID]struct{}, len(p.data.uniqueContractTokens)) + for id, contract := range p.data.uniqueContractTokens { + if contract.Type == string(types.ContractTypeSAC) { + verifiedSAC[id] = struct{}{} + } + } + keptSACBalances := p.pendingSACBalances[:0] + skippedSACBalances := 0 + for _, bal := range p.pendingSACBalances { + if _, ok := verifiedSAC[bal.ContractID]; ok { + keptSACBalances = append(keptSACBalances, bal) + } else { + skippedSACBalances++ + } + } + if skippedSACBalances > 0 { + log.Ctx(ctx).Warnf("checkpoint: skipped %d SAC balance(s) for contracts not verified as SAC", skippedSACBalances) + } + if len(keptSACBalances) > 0 { + if err := p.service.sacBalanceModel.BatchCopy(ctx, dbTx, keptSACBalances); err != nil { + return fmt.Errorf("copying verified SAC balances: %w", err) + } + } + // Persist protocol WASMs if err := p.service.persistProtocolWasms(ctx, dbTx, p.wasmClassifications); err != nil { return fmt.Errorf("persisting protocol wasms: %w", err) diff --git a/internal/services/checkpoint_test.go b/internal/services/checkpoint_test.go index 60161f667..a233a18cc 100644 --- a/internal/services/checkpoint_test.go +++ b/internal/services/checkpoint_test.go @@ -11,6 +11,7 @@ import ( "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/sac" "github.com/stellar/go-stellar-sdk/network" "github.com/stellar/go-stellar-sdk/strkey" "github.com/stellar/go-stellar-sdk/xdr" @@ -246,9 +247,8 @@ func TestCheckpointService_PopulateFromCheckpoint_AccountEntry(t *testing.T) { f.trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.TrustlineBalance) bool { return len(b) == 0 }), ).Return(nil).Once() - f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, - mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 0 }), - ).Return(nil).Once() + // SAC balances are handled separately in finalize (not via the batch), and only + // copied when there are verified balances — none here. err := f.svc.PopulateFromCheckpoint(context.Background(), 100, func(_ pgx.Tx) error { return nil }) require.NoError(t, err) @@ -319,7 +319,6 @@ func TestCheckpointService_PopulateFromCheckpoint_LiquidityPoolEntries(t *testin // trustline/native/sac (those flush empty). trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.TrustlineBalance) bool { return len(b) == 0 })).Return(nil).Once() nativeBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.NativeBalance) bool { return len(b) == 0 })).Return(nil).Once() - sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 0 })).Return(nil).Once() lpModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(pools []wbdata.LiquidityPool) bool { return len(pools) == 1 && pools[0].PoolID == expectedPoolID && pools[0].AssetA == "native" && pools[0].AmountA == 100 && @@ -375,12 +374,29 @@ func TestCheckpointService_PopulateFromCheckpoint_ContractDataEntry(t *testing.T require.NoError(t, err) } +// makeSACInstanceChange builds a verified SAC contract-instance entry for the +// given classic asset. The contract ID is the deterministically derived Stellar +// Asset Contract ID for that asset, so sac.AssetFromContractData authenticates it +// — this is what identifies a contract as a SAC rather than a shape look-alike. +func makeSACInstanceChange(t *testing.T, code, issuer, passphrase string) (ingest.Change, [32]byte) { + t.Helper() + asset := xdr.MustNewCreditAsset(code, issuer) + contractID, err := asset.ContractID(passphrase) + require.NoError(t, err) + data, err := sac.AssetToContractData(false, code, issuer, contractID) + require.NoError(t, err) + return ingest.Change{ + Type: xdr.LedgerEntryTypeContractData, + Post: &xdr.LedgerEntry{Data: data}, + }, contractID +} + // makeSACBalanceChange builds an ingest.Change for a ContractData Balance // entry whose holder is itself a contract — the shape // sac.ContractBalanceFromContractData requires to recognize a SAC balance. -// With no preceding contract-instance entry in the same checkpoint, this is -// how a SAC contract ends up in contract_tokens with Code/Name/Symbol unset, -// needing RPC enrichment (see finalize's pendingSACMetadata bookkeeping). +// The shape alone does not identify the contract as a SAC; the checkpoint +// records such a balance only when the contract is also confirmed via its +// instance entry (see makeSACInstanceChange). func makeSACBalanceChange(tokenContractHash, holderContractHash [32]byte) ingest.Change { return ingest.Change{ Type: xdr.LedgerEntryTypeContractData, @@ -413,148 +429,60 @@ func makeSACBalanceChange(tokenContractHash, holderContractHash [32]byte) ingest } } -// checkpointCommitProbeKey is a scratch ingest_store row written by an -// initializeCursors callback, inside the load's own transaction, purely so a -// test can prove — via a real, separate connection — whether the load -// transaction has actually committed yet. Postgres MVCC hides the row from -// any other connection until commit, which is what makes this a genuine -// after-commit check rather than an assertion on mock call ordering alone. -const checkpointCommitProbeKey = "checkpoint_commit_probe" - -func writeCheckpointCommitProbe(dbTx pgx.Tx) error { - _, err := dbTx.Exec(context.Background(), - `INSERT INTO ingest_store (key, value) VALUES ($1, $2) ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value`, - checkpointCommitProbeKey, "committed") - return err -} - -// TestCheckpointService_PopulateFromCheckpoint_SACMetadataEnrichedAfterCommit -// is the ING-10 regression test: it proves FetchSACMetadata (the RPC call) -// only happens once the load transaction — including cursor initialization — -// has already committed, by querying the commit probe row from the real DB -// pool (a connection independent of the load's transaction) from inside the -// FetchSACMetadata mock's callback. If the RPC call were still made inside -// the load transaction (the ING-10 bug), the probe row would not yet be -// visible and this assertion would fail. -func TestCheckpointService_PopulateFromCheckpoint_SACMetadataEnrichedAfterCommit(t *testing.T) { +// TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded proves a SAC +// balance is recorded when its contract is confirmed as a SAC via +// its instance entry. The instance carries the asset info, so contract_tokens is written +// with type=SAC and code/issuer set — no RPC enrichment is needed. +func TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded(t *testing.T) { f := setupCheckpointTest(t) - tokenHash := [32]byte{9, 9, 9} + issuer := "GAFOZZL77R57WMGES6BO6WJDEIFJ6662GMCVEX6ZESULRX3FRBGSSV5N" + instanceChange, contractID := makeSACInstanceChange(t, "USDC", issuer, f.svc.networkPassphrase) + contractAddr := strkey.MustEncode(strkey.VersionByteContract, contractID[:]) holderHash := [32]byte{8, 8, 8} - change := makeSACBalanceChange(tokenHash, holderHash) - tokenAddr := strkey.MustEncode(strkey.VersionByteContract, tokenHash[:]) + balanceChange := makeSACBalanceChange(contractID, holderHash) - f.reader.On("Read").Return(change, nil).Once() + f.reader.On("Read").Return(instanceChange, nil).Once() + f.reader.On("Read").Return(balanceChange, nil).Once() f.reader.On("Read").Return(ingest.Change{}, io.EOF).Once() f.reader.On("Close").Return(nil).Once() - f.trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.TrustlineBalance) bool { return len(b) == 0 })).Return(nil).Once() - f.nativeBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.NativeBalance) bool { return len(b) == 0 })).Return(nil).Once() - f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 1 })).Return(nil).Once() - + // The verified SAC balance is recorded in finalize (trustline/native/lp batch is empty). + f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.SACBalance) bool { + return len(b) == 1 && b[0].ContractID == wbdata.DeterministicContractID(contractAddr) + })).Return(nil).Once() + // contract_tokens is written from the instance with code/issuer already set (no enrichment). f.contractModel.On("BatchInsert", mock.Anything, mock.Anything, mock.MatchedBy(func(cs []*wbdata.Contract) bool { - return len(cs) == 1 && cs[0].ContractID == tokenAddr && cs[0].Type == string(types.ContractTypeSAC) && cs[0].Code == nil + return len(cs) == 1 && cs[0].ContractID == contractAddr && cs[0].Type == string(types.ContractTypeSAC) && + cs[0].Code != nil && *cs[0].Code == "USDC" })).Return(nil).Once() - enrichedName := "Test Token" - enrichedSymbol := "TST" - f.contractMetadataService.On("FetchSACMetadata", mock.Anything, []string{tokenAddr}). - Run(func(_ mock.Arguments) { - var value string - queryErr := f.svc.db.QueryRow(context.Background(), - `SELECT value FROM ingest_store WHERE key = $1`, checkpointCommitProbeKey).Scan(&value) - require.NoError(t, queryErr, "the load transaction (including cursor init) must already be committed before SAC metadata enrichment runs") - assert.Equal(t, "committed", value) - }). - Return([]*wbdata.Contract{{ - ID: wbdata.DeterministicContractID(tokenAddr), - ContractID: tokenAddr, - Type: string(types.ContractTypeSAC), - Name: &enrichedName, - Symbol: &enrichedSymbol, - }}, nil).Once() - - f.contractModel.On("BatchUpdateMetadata", mock.Anything, mock.Anything, mock.MatchedBy(func(cs []*wbdata.Contract) bool { - return len(cs) == 1 && cs[0].ContractID == tokenAddr && cs[0].Name != nil && *cs[0].Name == enrichedName - })).Return(nil).Once() + // FetchSACMetadata must NOT be called: the verified instance already provides metadata. - err := f.svc.PopulateFromCheckpoint(context.Background(), 100, writeCheckpointCommitProbe) + err := f.svc.PopulateFromCheckpoint(context.Background(), 100, func(_ pgx.Tx) error { return nil }) require.NoError(t, err) } -// TestCheckpointService_PopulateFromCheckpoint_SACMetadataFetchFailureDoesNotFailLoad -// is the other half of ING-10: a failed SAC metadata fetch must be logged and -// leave the already-committed load's rows in place (defaults, unenriched), -// not fail PopulateFromCheckpoint. -func TestCheckpointService_PopulateFromCheckpoint_SACMetadataFetchFailureDoesNotFailLoad(t *testing.T) { +// TestCheckpointService_PopulateFromCheckpoint_UnverifiedSACBalanceDropped verifies that a +// Balance-shaped contract-data entry whose contract is NOT confirmed as a SAC (no instance +// entry in the checkpoint) is dropped: no balance is recorded, no contract_tokens row is +// created from the shape, and no RPC enrichment runs. Recording it would associate a balance +// with a contract of unknown type and, in live ingestion, violate the deferred +// fk_contract_token at COMMIT. +func TestCheckpointService_PopulateFromCheckpoint_UnverifiedSACBalanceDropped(t *testing.T) { f := setupCheckpointTest(t) - tokenHash := [32]byte{7, 7, 7} - holderHash := [32]byte{6, 6, 6} - change := makeSACBalanceChange(tokenHash, holderHash) - - f.reader.On("Read").Return(change, nil).Once() - f.reader.On("Read").Return(ingest.Change{}, io.EOF).Once() - f.reader.On("Close").Return(nil).Once() - - f.trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - f.nativeBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - - f.contractModel.On("BatchInsert", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() - - // The enrichment is retried; every attempt fails, and the load still must not fail. - f.contractMetadataService.On("FetchSACMetadata", mock.Anything, mock.Anything). - Return(nil, errors.New("rpc unavailable")).Times(f.svc.sacEnrichmentRetries) - - // No "BatchUpdateMetadata" expectation is registered: if enrichSACMetadata - // called it despite the fetch failing, the mock would fail this test for - // an unexpected call. - - cursorsCalled := false - err := f.svc.PopulateFromCheckpoint(context.Background(), 100, func(_ pgx.Tx) error { - cursorsCalled = true - return nil - }) - require.NoError(t, err, "a SAC metadata fetch failure must not fail the completed load") - assert.True(t, cursorsCalled) -} - -// TestCheckpointService_PopulateFromCheckpoint_SACMetadataRetriedThenSucceeds proves the -// bounded retry absorbs a transient enrichment blip: FetchSACMetadata fails twice, then -// succeeds, and the enrichment write runs exactly once. -func TestCheckpointService_PopulateFromCheckpoint_SACMetadataRetriedThenSucceeds(t *testing.T) { - f := setupCheckpointTest(t) - - tokenHash := [32]byte{4, 4, 4} - holderHash := [32]byte{3, 3, 3} + tokenHash := [32]byte{9, 9, 9} + holderHash := [32]byte{8, 8, 8} change := makeSACBalanceChange(tokenHash, holderHash) - tokenAddr := strkey.MustEncode(strkey.VersionByteContract, tokenHash[:]) f.reader.On("Read").Return(change, nil).Once() f.reader.On("Read").Return(ingest.Change{}, io.EOF).Once() f.reader.On("Close").Return(nil).Once() - f.trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - f.nativeBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() - f.contractModel.On("BatchInsert", mock.Anything, mock.Anything, mock.Anything).Return(nil).Once() - - enrichedName := "Retry Token" - enrichedSymbol := "RTY" - f.contractMetadataService.On("FetchSACMetadata", mock.Anything, []string{tokenAddr}). - Return(nil, errors.New("rpc blip")).Twice() - f.contractMetadataService.On("FetchSACMetadata", mock.Anything, []string{tokenAddr}). - Return([]*wbdata.Contract{{ - ID: wbdata.DeterministicContractID(tokenAddr), - ContractID: tokenAddr, - Type: string(types.ContractTypeSAC), - Name: &enrichedName, - Symbol: &enrichedSymbol, - }}, nil).Once() - f.contractModel.On("BatchUpdateMetadata", mock.Anything, mock.Anything, mock.MatchedBy(func(cs []*wbdata.Contract) bool { - return len(cs) == 1 && cs[0].ContractID == tokenAddr && cs[0].Name != nil && *cs[0].Name == enrichedName - })).Return(nil).Once() + // The unverified balance is dropped, so nothing is written: no SAC BatchCopy (finalize + // skips the empty set), no contractModel.BatchInsert from the shape, and no + // FetchSACMetadata. The strict mocks fail the test on any such unexpected call. err := f.svc.PopulateFromCheckpoint(context.Background(), 100, func(_ pgx.Tx) error { return nil }) require.NoError(t, err) @@ -823,7 +751,6 @@ func TestCheckpointProcessor_ProcessEntry(t *testing.T) { batch: &batch{ nativeBalances: make([]wbdata.NativeBalance, 0), trustlineBalances: make([]wbdata.TrustlineBalance, 0), - sacBalances: make([]wbdata.SACBalance, 0), }, } } @@ -925,7 +852,7 @@ func TestCheckpointProcessor_ProcessEntry(t *testing.T) { // Non-SAC balance entries are no longer tracked (SEP-41 tracking removed) assert.Equal(t, 0, proc.entries) - assert.Empty(t, proc.batch.sacBalances) + assert.Empty(t, proc.pendingSACBalances) }) t.Run("unhandled_entry_type_ignored", func(t *testing.T) { diff --git a/internal/services/token_ingestion.go b/internal/services/token_ingestion.go index 548881bcd..94fbc1589 100644 --- a/internal/services/token_ingestion.go +++ b/internal/services/token_ingestion.go @@ -6,6 +6,7 @@ import ( "context" "fmt" + "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/stellar/go-stellar-sdk/support/log" @@ -29,6 +30,7 @@ type TokenIngestionServiceConfig struct { TrustlineBalanceModel wbdata.TrustlineBalanceModelInterface NativeBalanceModel wbdata.NativeBalanceModelInterface SACBalanceModel wbdata.SACBalanceModelInterface + ContractModel wbdata.ContractModelInterface LiquidityPoolModel wbdata.LiquidityPoolModelInterface LiquidityPoolBalanceModel wbdata.LiquidityPoolBalanceModelInterface NetworkPassphrase string @@ -39,6 +41,7 @@ type tokenIngestionService struct { trustlineBalanceModel wbdata.TrustlineBalanceModelInterface nativeBalanceModel wbdata.NativeBalanceModelInterface sacBalanceModel wbdata.SACBalanceModelInterface + contractModel wbdata.ContractModelInterface liquidityPoolModel wbdata.LiquidityPoolModelInterface liquidityPoolBalanceModel wbdata.LiquidityPoolBalanceModelInterface networkPassphrase string @@ -50,6 +53,7 @@ func NewTokenIngestionService(cfg TokenIngestionServiceConfig) *tokenIngestionSe trustlineBalanceModel: cfg.TrustlineBalanceModel, nativeBalanceModel: cfg.NativeBalanceModel, sacBalanceModel: cfg.SACBalanceModel, + contractModel: cfg.ContractModel, liquidityPoolModel: cfg.LiquidityPoolModel, liquidityPoolBalanceModel: cfg.LiquidityPoolBalanceModel, networkPassphrase: cfg.NetworkPassphrase, @@ -155,10 +159,39 @@ func (s *tokenIngestionService) processSACBalanceChanges(ctx context.Context, db return nil } + // A SAC balance entry's shape (key ["Balance", holder], value {amount, authorized, + // clawback}) does not by itself identify the contract as a SAC — any contract can + // write an entry of that shape into its own storage. Record a balance only for a + // contract confirmed to be a SAC via its instance entry (present in contract_tokens + // with type='SAC', inserted earlier in this same transaction by prepareNewSACContracts + // or in a prior ledger). This also keeps every sac_balances row backed by a + // contract_tokens parent, so the fk_contract_token constraint holds at COMMIT. + contractIDSet := make(map[uuid.UUID]struct{}, len(changesByKey)) + for _, change := range changesByKey { + contractIDSet[wbdata.DeterministicContractID(change.ContractID)] = struct{}{} + } + contractIDs := make([]uuid.UUID, 0, len(contractIDSet)) + for id := range contractIDSet { + contractIDs = append(contractIDs, id) + } + verifiedIDs, err := s.contractModel.GetExistingSACByID(ctx, dbTx, contractIDs) + if err != nil { + return fmt.Errorf("checking verified SAC contracts: %w", err) + } + verified := make(map[uuid.UUID]struct{}, len(verifiedIDs)) + for _, id := range verifiedIDs { + verified[id] = struct{}{} + } + var upserts []wbdata.SACBalance var deletes []wbdata.SACBalance + var skipped int for _, change := range changesByKey { contractID := wbdata.DeterministicContractID(change.ContractID) + if _, ok := verified[contractID]; !ok { + skipped++ + continue + } sacBal := wbdata.SACBalance{ AccountID: types.AddressBytea(change.AccountID), ContractID: contractID, @@ -173,6 +206,9 @@ func (s *tokenIngestionService) processSACBalanceChanges(ctx context.Context, db upserts = append(upserts, sacBal) } } + if skipped > 0 { + log.Ctx(ctx).Warnf("skipped %d SAC balance change(s) for contracts not verified as SAC", skipped) + } if len(upserts) > 0 || len(deletes) > 0 { if err := s.sacBalanceModel.BatchUpsert(ctx, dbTx, upserts, deletes); err != nil { diff --git a/internal/services/token_ingestion_test.go b/internal/services/token_ingestion_test.go index 35a886958..3b4c7a404 100644 --- a/internal/services/token_ingestion_test.go +++ b/internal/services/token_ingestion_test.go @@ -9,6 +9,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/stellar/go-stellar-sdk/ingest" "github.com/stellar/go-stellar-sdk/keypair" + "github.com/stellar/go-stellar-sdk/strkey" "github.com/stellar/go-stellar-sdk/xdr" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -374,3 +375,77 @@ func TestProcessTokenChanges(t *testing.T) { assert.NoError(t, err) }) } + +// TestProcessSACBalanceChanges_GatesOnVerifiedSAC verifies that a SAC balance change is +// recorded only when its contract has a SAC row in contract_tokens. A change for a contract +// not confirmed as a SAC is silently skipped, never persisted and never allowed to trigger +// the deferred fk_contract_token violation that would halt live ingestion. +func TestProcessSACBalanceChanges_GatesOnVerifiedSAC(t *testing.T) { + ctx := context.Background() + + dbt := dbtest.Open(t) + defer dbt.Close() + dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) + require.NoError(t, err) + defer dbConnectionPool.Close() + + dbMetrics := metrics.NewMetrics(prometheus.NewRegistry()).DB + sacBalanceModel := &wbdata.SACBalanceModel{DB: dbConnectionPool, Metrics: dbMetrics} + contractModel := &wbdata.ContractModel{DB: dbConnectionPool, Metrics: dbMetrics} + + service := NewTokenIngestionService(TokenIngestionServiceConfig{ + SACBalanceModel: sacBalanceModel, + ContractModel: contractModel, + NetworkPassphrase: "Test SDF Network ; September 2015", + }) + + verifiedContract := strkey.MustEncode(strkey.VersionByteContract, bytes32(0x11)) + unverifiedContract := strkey.MustEncode(strkey.VersionByteContract, bytes32(0x22)) + holder := strkey.MustEncode(strkey.VersionByteContract, bytes32(0x33)) + + // The SAC has been confirmed via its instance entry, so it already has a + // contract_tokens row with code/issuer. The unverified contract has no such row. + code, issuer := "USDC", "GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN" + err = db.RunInTransaction(ctx, dbConnectionPool, func(dbTx pgx.Tx) error { + return contractModel.BatchInsert(ctx, dbTx, []*wbdata.Contract{{ + ID: wbdata.DeterministicContractID(verifiedContract), + ContractID: verifiedContract, + Type: string(types.ContractTypeSAC), + Code: &code, + Issuer: &issuer, + Decimals: 7, + }}) + }) + require.NoError(t, err) + + changes := map[indexer.SACBalanceChangeKey]types.SACBalanceChange{ + {AccountID: holder, ContractID: verifiedContract}: { + AccountID: holder, ContractID: verifiedContract, Operation: types.SACBalanceOpAdd, + Balance: "1000", IsAuthorized: true, LedgerNumber: 100, + }, + {AccountID: holder, ContractID: unverifiedContract}: { + AccountID: holder, ContractID: unverifiedContract, Operation: types.SACBalanceOpAdd, + Balance: "9999", IsAuthorized: true, LedgerNumber: 100, + }, + } + + err = db.RunInTransaction(ctx, dbConnectionPool, func(dbTx pgx.Tx) error { + return service.ProcessTokenChanges(ctx, dbTx, nil, nil, changes, nil, nil) + }) + require.NoError(t, err, "the unverified balance must be skipped without a foreign-key violation") + + got, err := sacBalanceModel.GetByAccount(ctx, holder, nil, nil, wbdata.ASC) + require.NoError(t, err) + require.Len(t, got, 1, "only the verified SAC balance is persisted") + assert.Equal(t, wbdata.DeterministicContractID(verifiedContract), got[0].ContractID) + assert.Equal(t, "1000", got[0].Balance) +} + +// bytes32 returns a 32-byte slice filled with b, for deterministic test contract IDs. +func bytes32(b byte) []byte { + out := make([]byte, 32) + for i := range out { + out[i] = b + } + return out +} From 39fd73e1e68f1b3bf96837361382f6060eb5c5f8 Mon Sep 17 00:00:00 2001 From: jiahuihu Date: Fri, 21 Aug 2026 17:53:23 -0400 Subject: [PATCH 2/2] fix(ingest): bound checkpoint SAC-balance memory during bootstrap --- internal/data/mocks.go | 5 ++ internal/data/sac_balances.go | 27 ++++++++++ internal/data/sac_balances_test.go | 52 ++++++++++++++++++ internal/services/checkpoint.go | 80 ++++++++++++++-------------- internal/services/checkpoint_test.go | 42 +++++++++------ 5 files changed, 151 insertions(+), 55 deletions(-) diff --git a/internal/data/mocks.go b/internal/data/mocks.go index f7847ffb3..71c9f19b1 100644 --- a/internal/data/mocks.go +++ b/internal/data/mocks.go @@ -235,6 +235,11 @@ func (m *SACBalanceModelMock) BatchCopy(ctx context.Context, dbTx pgx.Tx, balanc return args.Error(0) } +func (m *SACBalanceModelMock) DeleteUnverified(ctx context.Context, dbTx pgx.Tx) (int64, error) { + args := m.Called(ctx, dbTx) + return args.Get(0).(int64), args.Error(1) +} + // ProtocolWasmsModelMock is a mock implementation of ProtocolWasmsModelInterface. type ProtocolWasmsModelMock struct { mock.Mock diff --git a/internal/data/sac_balances.go b/internal/data/sac_balances.go index 3ee2df725..14888de73 100644 --- a/internal/data/sac_balances.go +++ b/internal/data/sac_balances.go @@ -47,6 +47,13 @@ type SACBalanceModelInterface interface { // Batch operations (for initial population) BatchCopy(ctx context.Context, dbTx pgx.Tx, balances []SACBalance) error + + // DeleteUnverified removes every sac_balances row whose contract is not present in + // contract_tokens as a SAC (type='SAC'), returning the number deleted. Checkpoint + // population streams balances in from their shape and calls this in finalize, once + // contract_tokens is fully populated, to drop rows for contracts not confirmed as + // SACs — before the deferred fk_contract_token is checked at COMMIT. + DeleteUnverified(ctx context.Context, dbTx pgx.Tx) (int64, error) } // SACBalanceModel implements SACBalanceModelInterface. @@ -202,6 +209,26 @@ func (m *SACBalanceModel) BatchUpsert(ctx context.Context, dbTx pgx.Tx, upserts return nil } +// DeleteUnverified removes sac_balances rows whose contract is not a SAC in contract_tokens. +func (m *SACBalanceModel) DeleteUnverified(ctx context.Context, dbTx pgx.Tx) (int64, error) { + const query = ` + DELETE FROM sac_balances sb + WHERE NOT EXISTS ( + SELECT 1 FROM contract_tokens ct + WHERE ct.id = sb.contract_id AND ct.type = $1 + )` + + start := time.Now() + tag, err := dbTx.Exec(ctx, query, string(types.ContractTypeSAC)) + m.Metrics.QueryDuration.WithLabelValues("DeleteUnverified", "sac_balances").Observe(time.Since(start).Seconds()) + m.Metrics.QueriesTotal.WithLabelValues("DeleteUnverified", "sac_balances").Inc() + if err != nil { + m.Metrics.QueryErrors.WithLabelValues("DeleteUnverified", "sac_balances", utils.GetDBErrorType(err)).Inc() + return 0, fmt.Errorf("deleting unverified SAC balances: %w", err) + } + return tag.RowsAffected(), nil +} + // BatchCopy performs bulk insert using COPY protocol for speed during checkpoint population. func (m *SACBalanceModel) BatchCopy(ctx context.Context, dbTx pgx.Tx, balances []SACBalance) error { if len(balances) == 0 { diff --git a/internal/data/sac_balances_test.go b/internal/data/sac_balances_test.go index fb6182152..1256a1bc6 100644 --- a/internal/data/sac_balances_test.go +++ b/internal/data/sac_balances_test.go @@ -7,6 +7,7 @@ import ( "slices" "testing" + "github.com/jackc/pgx/v5" "github.com/prometheus/client_golang/prometheus" "github.com/stellar/go-stellar-sdk/strkey" "github.com/stretchr/testify/require" @@ -531,3 +532,54 @@ func TestSACBalanceModel_BatchCopy(t *testing.T) { require.Equal(t, 3, count) }) } + +func TestSACBalanceModel_DeleteUnverified(t *testing.T) { + ctx := context.Background() + + dbt := dbtest.Open(t) + defer dbt.Close() + dbConnectionPool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) + require.NoError(t, err) + defer dbConnectionPool.Close() + + dbMetrics := metrics.NewMetrics(prometheus.NewRegistry()).DB + model := &SACBalanceModel{DB: dbConnectionPool, Metrics: dbMetrics} + + sacAddr := randomContractAddress(t) + sacID := DeterministicContractID(sacAddr) + orphanAddr := randomContractAddress(t) // deliberately has no contract_tokens row + orphanID := DeterministicContractID(orphanAddr) + holder := randomContractAddress(t) + + // A confirmed SAC parent for the verified balance. + _, err = dbConnectionPool.Exec(ctx, ` + INSERT INTO contract_tokens (id, contract_id, type, code, issuer, decimals) + VALUES ($1, $2, 'SAC', 'USDC', 'GISSUER1AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA', 7)`, + sacID, sacAddr) + require.NoError(t, err) + + err = db.RunInTransaction(ctx, dbConnectionPool, func(dbTx pgx.Tx) error { + // Stream in a verified balance and an orphan balance (no contract_tokens parent). + // The deferred fk_contract_token permits the orphan until COMMIT. + if bErr := model.BatchCopy(ctx, dbTx, []SACBalance{ + {AccountID: types.AddressBytea(holder), ContractID: sacID, Balance: "1000", IsAuthorized: true, LedgerNumber: 100}, + {AccountID: types.AddressBytea(holder), ContractID: orphanID, Balance: "9999", IsAuthorized: true, LedgerNumber: 100}, + }); bErr != nil { + return bErr + } + deleted, dErr := model.DeleteUnverified(ctx, dbTx) + if dErr != nil { + return dErr + } + require.Equal(t, int64(1), deleted, "only the orphan balance is deleted") + return nil + }) + require.NoError(t, err, "the orphan balance must be deleted so the deferred FK holds at COMMIT") + + // Only the verified SAC balance survived. + got, err := model.GetByAccount(ctx, holder, nil, nil, ASC) + require.NoError(t, err) + require.Len(t, got, 1) + require.Equal(t, sacID, got[0].ContractID) + require.Equal(t, "1000", got[0].Balance) +} diff --git a/internal/services/checkpoint.go b/internal/services/checkpoint.go index 6ead428ff..645373b8d 100644 --- a/internal/services/checkpoint.go +++ b/internal/services/checkpoint.go @@ -140,10 +140,12 @@ func newCheckpointData() checkpointData { type batch struct { trustlineBalances []wbdata.TrustlineBalance nativeBalances []wbdata.NativeBalance + sacBalances []wbdata.SACBalance liquidityPools []wbdata.LiquidityPool liquidityPoolBalances []wbdata.LiquidityPoolBalance trustlineBalanceModel wbdata.TrustlineBalanceModelInterface nativeBalanceModel wbdata.NativeBalanceModelInterface + sacBalanceModel wbdata.SACBalanceModelInterface liquidityPoolModel wbdata.LiquidityPoolModelInterface liquidityPoolBalanceModel wbdata.LiquidityPoolBalanceModelInterface } @@ -151,16 +153,19 @@ type batch struct { func newBatch( trustlineBalanceModel wbdata.TrustlineBalanceModelInterface, nativeBalanceModel wbdata.NativeBalanceModelInterface, + sacBalanceModel wbdata.SACBalanceModelInterface, liquidityPoolModel wbdata.LiquidityPoolModelInterface, liquidityPoolBalanceModel wbdata.LiquidityPoolBalanceModelInterface, ) *batch { return &batch{ trustlineBalances: make([]wbdata.TrustlineBalance, 0, flushBatchSize), nativeBalances: make([]wbdata.NativeBalance, 0, flushBatchSize), + sacBalances: make([]wbdata.SACBalance, 0, flushBatchSize), liquidityPools: make([]wbdata.LiquidityPool, 0, flushBatchSize), liquidityPoolBalances: make([]wbdata.LiquidityPoolBalance, 0, flushBatchSize), trustlineBalanceModel: trustlineBalanceModel, nativeBalanceModel: nativeBalanceModel, + sacBalanceModel: sacBalanceModel, liquidityPoolModel: liquidityPoolModel, liquidityPoolBalanceModel: liquidityPoolBalanceModel, } @@ -191,6 +196,10 @@ func (b *batch) addNativeBalance(accountAddress string, balance, minimumBalance, }) } +func (b *batch) addSACBalance(sacBalance wbdata.SACBalance) { + b.sacBalances = append(b.sacBalances, sacBalance) +} + func (b *batch) addLiquidityPool(pool wbdata.LiquidityPool) { b.liquidityPools = append(b.liquidityPools, pool) } @@ -212,6 +221,13 @@ func (b *batch) flush(ctx context.Context, dbTx pgx.Tx) error { if err := b.nativeBalanceModel.BatchCopy(ctx, dbTx, b.nativeBalances); err != nil { return fmt.Errorf("batch inserting native balances: %w", err) } + // SAC balances are inserted here from their shape, but only kept if their contract + // is confirmed as a SAC. finalize deletes any that lack a verified contract_tokens + // parent before COMMIT, so the deferred fk_contract_token holds. Because the FK is + // DEFERRABLE INITIALLY DEFERRED, inserting an as-yet-unparented row here is safe. + if err := b.sacBalanceModel.BatchCopy(ctx, dbTx, b.sacBalances); err != nil { + return fmt.Errorf("batch inserting SAC balances: %w", err) + } if err := b.liquidityPoolModel.BatchCopy(ctx, dbTx, b.liquidityPools); err != nil { return fmt.Errorf("batch inserting liquidity pools: %w", err) } @@ -222,13 +238,14 @@ func (b *batch) flush(ctx context.Context, dbTx pgx.Tx) error { } func (b *batch) count() int { - return len(b.trustlineBalances) + len(b.nativeBalances) + + return len(b.trustlineBalances) + len(b.nativeBalances) + len(b.sacBalances) + len(b.liquidityPools) + len(b.liquidityPoolBalances) } func (b *batch) reset() { b.trustlineBalances = b.trustlineBalances[:0] b.nativeBalances = b.nativeBalances[:0] + b.sacBalances = b.sacBalances[:0] b.liquidityPools = b.liquidityPools[:0] b.liquidityPoolBalances = b.liquidityPoolBalances[:0] } @@ -250,13 +267,11 @@ type checkpointProcessor struct { // call; PopulateFromCheckpoint fetches metadata for these IDs in a short // follow-up transaction after the load commits. pendingSACMetadata []string - // pendingSACBalances holds SAC-shaped balance entries seen during the scan. They - // are not written to the batch during the pass because a balance entry's shape does - // not by itself identify its contract as a SAC. finalize keeps only those whose - // contract was confirmed as a SAC via its instance entry (uniqueContractTokens with - // type=SAC), then copies them in — before the deferred fk_contract_token is checked - // at COMMIT. - pendingSACBalances []wbdata.SACBalance + // sawSACBalance records whether any SAC-shaped balance entry was streamed to the + // batch during the scan. Such entries are inserted from their shape alone, so if any + // were seen, finalize runs a cleanup that deletes the rows whose contract is not a + // confirmed SAC before COMMIT (see deleteUnverifiedSACBalances). + sawSACBalance bool } // PopulateFromCheckpoint performs initial cache population from Stellar history archive. @@ -299,7 +314,7 @@ func (s *checkpointService) PopulateFromCheckpoint(ctx context.Context, checkpoi dbTx: dbTx, checkpointLedger: checkpointLedger, data: newCheckpointData(), - batch: newBatch(s.trustlineBalanceModel, s.nativeBalanceModel, s.liquidityPoolModel, s.liquidityPoolBalanceModel), + batch: newBatch(s.trustlineBalanceModel, s.nativeBalanceModel, s.sacBalanceModel, s.liquidityPoolModel, s.liquidityPoolBalanceModel), wasmClassifications: make(map[xdr.Hash]types.ContractType), contractAddressesByWasmHash: make(map[xdr.Hash][]xdr.Hash), startTime: time.Now(), @@ -513,14 +528,13 @@ func (p *checkpointProcessor) processEntry(change ingest.Change) { _, _, ok := sac.ContractBalanceFromContractData(*change.Post, p.service.networkPassphrase) if ok { - // Shape matches a SAC balance, but that alone does not identify the - // contract as a SAC. Defer the balance to finalize, which keeps it - // only if the contract was confirmed via its instance entry. Do NOT - // create a contract_tokens row from the balance shape: the instance - // entry is the authoritative source of a contract's type and metadata. + // Shape matches a SAC balance. Stream it to the batch, but do NOT create + // a contract_tokens row from the balance shape: the instance entry is the + // authoritative source of a contract's type and metadata. finalize deletes + // any balance whose contract is not a confirmed SAC before COMMIT. contractUUID := wbdata.DeterministicContractID(contractAddressStr) balanceStr, authorized, clawback := p.service.extractSACBalanceFields(contractDataEntry.Val) - p.pendingSACBalances = append(p.pendingSACBalances, wbdata.SACBalance{ + p.batch.addSACBalance(wbdata.SACBalance{ AccountID: types.AddressBytea(holderAddress), ContractID: contractUUID, Balance: balanceStr, @@ -528,6 +542,7 @@ func (p *checkpointProcessor) processEntry(change ingest.Change) { IsClawbackEnabled: clawback, LedgerNumber: p.checkpointLedger, }) + p.sawSACBalance = true p.entries++ } } @@ -585,32 +600,17 @@ func (p *checkpointProcessor) finalize(ctx context.Context, dbTx pgx.Tx) error { return fmt.Errorf("storing tokens in postgres: %w", err) } - // Persist SAC balances, but only for contracts confirmed as SAC via their - // instance entry. The scan populated uniqueContractTokens; a balance whose - // contract is not among them is not a SAC balance and is dropped. This also - // guarantees every retained balance has a contract_tokens parent, so the - // deferred fk_contract_token holds at COMMIT. - verifiedSAC := make(map[uuid.UUID]struct{}, len(p.data.uniqueContractTokens)) - for id, contract := range p.data.uniqueContractTokens { - if contract.Type == string(types.ContractTypeSAC) { - verifiedSAC[id] = struct{}{} - } - } - keptSACBalances := p.pendingSACBalances[:0] - skippedSACBalances := 0 - for _, bal := range p.pendingSACBalances { - if _, ok := verifiedSAC[bal.ContractID]; ok { - keptSACBalances = append(keptSACBalances, bal) - } else { - skippedSACBalances++ + // SAC balances were streamed in from their shape alone. Now that contract_tokens is + // fully populated (above), delete any SAC balance whose contract is not a confirmed + // SAC. This runs before COMMIT, so every remaining balance has a verified + // contract_tokens parent and the deferred fk_contract_token holds. + if p.sawSACBalance { + deleted, err := p.service.sacBalanceModel.DeleteUnverified(ctx, dbTx) + if err != nil { + return fmt.Errorf("deleting unverified SAC balances: %w", err) } - } - if skippedSACBalances > 0 { - log.Ctx(ctx).Warnf("checkpoint: skipped %d SAC balance(s) for contracts not verified as SAC", skippedSACBalances) - } - if len(keptSACBalances) > 0 { - if err := p.service.sacBalanceModel.BatchCopy(ctx, dbTx, keptSACBalances); err != nil { - return fmt.Errorf("copying verified SAC balances: %w", err) + if deleted > 0 { + log.Ctx(ctx).Warnf("checkpoint: deleted %d SAC balance(s) for contracts not verified as SAC", deleted) } } diff --git a/internal/services/checkpoint_test.go b/internal/services/checkpoint_test.go index a233a18cc..89b1079ca 100644 --- a/internal/services/checkpoint_test.go +++ b/internal/services/checkpoint_test.go @@ -247,8 +247,9 @@ func TestCheckpointService_PopulateFromCheckpoint_AccountEntry(t *testing.T) { f.trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.TrustlineBalance) bool { return len(b) == 0 }), ).Return(nil).Once() - // SAC balances are handled separately in finalize (not via the batch), and only - // copied when there are verified balances — none here. + f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, + mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 0 }), + ).Return(nil).Once() err := f.svc.PopulateFromCheckpoint(context.Background(), 100, func(_ pgx.Tx) error { return nil }) require.NoError(t, err) @@ -319,6 +320,7 @@ func TestCheckpointService_PopulateFromCheckpoint_LiquidityPoolEntries(t *testin // trustline/native/sac (those flush empty). trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.TrustlineBalance) bool { return len(b) == 0 })).Return(nil).Once() nativeBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.NativeBalance) bool { return len(b) == 0 })).Return(nil).Once() + sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 0 })).Return(nil).Once() lpModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(pools []wbdata.LiquidityPool) bool { return len(pools) == 1 && pools[0].PoolID == expectedPoolID && pools[0].AssetA == "native" && pools[0].AmountA == 100 && @@ -429,10 +431,10 @@ func makeSACBalanceChange(tokenContractHash, holderContractHash [32]byte) ingest } } -// TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded proves a SAC -// balance is recorded when its contract is confirmed as a SAC via -// its instance entry. The instance carries the asset info, so contract_tokens is written -// with type=SAC and code/issuer set — no RPC enrichment is needed. +// TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded proves that when a +// SAC balance's contract is confirmed as a SAC via its instance entry, contract_tokens is +// written with type=SAC and code/issuer from the instance (no RPC enrichment), and the +// balance survives the finalize cleanup that deletes rows for unconfirmed contracts. func TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded(t *testing.T) { f := setupCheckpointTest(t) @@ -447,7 +449,10 @@ func TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded(t * f.reader.On("Read").Return(ingest.Change{}, io.EOF).Once() f.reader.On("Close").Return(nil).Once() - // The verified SAC balance is recorded in finalize (trustline/native/lp batch is empty). + // The presence of the SAC balance triggers a batch flush; the other balance types flush empty. + f.trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.TrustlineBalance) bool { return len(b) == 0 })).Return(nil).Once() + f.nativeBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.NativeBalance) bool { return len(b) == 0 })).Return(nil).Once() + // The balance is streamed to sac_balances during the scan. f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 1 && b[0].ContractID == wbdata.DeterministicContractID(contractAddr) })).Return(nil).Once() @@ -456,6 +461,8 @@ func TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded(t * return len(cs) == 1 && cs[0].ContractID == contractAddr && cs[0].Type == string(types.ContractTypeSAC) && cs[0].Code != nil && *cs[0].Code == "USDC" })).Return(nil).Once() + // finalize runs the cleanup; the verified balance has a parent, so nothing is deleted. + f.sacBalanceModel.On("DeleteUnverified", mock.Anything, mock.Anything).Return(int64(0), nil).Once() // FetchSACMetadata must NOT be called: the verified instance already provides metadata. @@ -465,10 +472,9 @@ func TestCheckpointService_PopulateFromCheckpoint_VerifiedSACBalanceRecorded(t * // TestCheckpointService_PopulateFromCheckpoint_UnverifiedSACBalanceDropped verifies that a // Balance-shaped contract-data entry whose contract is NOT confirmed as a SAC (no instance -// entry in the checkpoint) is dropped: no balance is recorded, no contract_tokens row is -// created from the shape, and no RPC enrichment runs. Recording it would associate a balance -// with a contract of unknown type and, in live ingestion, violate the deferred -// fk_contract_token at COMMIT. +// entry in the checkpoint) does not create a contract_tokens row from the shape and is removed +// by the finalize cleanup (DeleteUnverified). The row is streamed in but deleted before +// COMMIT, so the deferred fk_contract_token holds and no unconfirmed contract is classified. func TestCheckpointService_PopulateFromCheckpoint_UnverifiedSACBalanceDropped(t *testing.T) { f := setupCheckpointTest(t) @@ -480,9 +486,14 @@ func TestCheckpointService_PopulateFromCheckpoint_UnverifiedSACBalanceDropped(t f.reader.On("Read").Return(ingest.Change{}, io.EOF).Once() f.reader.On("Close").Return(nil).Once() - // The unverified balance is dropped, so nothing is written: no SAC BatchCopy (finalize - // skips the empty set), no contractModel.BatchInsert from the shape, and no - // FetchSACMetadata. The strict mocks fail the test on any such unexpected call. + // The presence of the SAC balance triggers a batch flush; the other balance types flush empty. + f.trustlineBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.TrustlineBalance) bool { return len(b) == 0 })).Return(nil).Once() + f.nativeBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.NativeBalance) bool { return len(b) == 0 })).Return(nil).Once() + // The balance is streamed in during the scan... + f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 1 })).Return(nil).Once() + // ...but no contract_tokens row is created from the shape, and finalize deletes it. + f.sacBalanceModel.On("DeleteUnverified", mock.Anything, mock.Anything).Return(int64(1), nil).Once() + // No contractModel.BatchInsert and no FetchSACMetadata: the strict mocks fail on any such call. err := f.svc.PopulateFromCheckpoint(context.Background(), 100, func(_ pgx.Tx) error { return nil }) require.NoError(t, err) @@ -751,6 +762,7 @@ func TestCheckpointProcessor_ProcessEntry(t *testing.T) { batch: &batch{ nativeBalances: make([]wbdata.NativeBalance, 0), trustlineBalances: make([]wbdata.TrustlineBalance, 0), + sacBalances: make([]wbdata.SACBalance, 0), }, } } @@ -852,7 +864,7 @@ func TestCheckpointProcessor_ProcessEntry(t *testing.T) { // Non-SAC balance entries are no longer tracked (SEP-41 tracking removed) assert.Equal(t, 0, proc.entries) - assert.Empty(t, proc.pendingSACBalances) + assert.Empty(t, proc.batch.sacBalances) }) t.Run("unhandled_entry_type_ignored", func(t *testing.T) {