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..71c9f19b1 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 { @@ -227,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/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..645373b8d 100644 --- a/internal/services/checkpoint.go +++ b/internal/services/checkpoint.go @@ -221,6 +221,10 @@ 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) } @@ -263,6 +267,11 @@ type checkpointProcessor struct { // call; PopulateFromCheckpoint fetches metadata for these IDs in a short // follow-up transaction after the load commits. pendingSACMetadata []string + // 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. @@ -519,15 +528,11 @@ func (p *checkpointProcessor) processEntry(change ingest.Change) { _, _, ok := sac.ContractBalanceFromContractData(*change.Post, p.service.networkPassphrase) if ok { + // 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) - 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{ AccountID: types.AddressBytea(holderAddress), @@ -537,6 +542,7 @@ func (p *checkpointProcessor) processEntry(change ingest.Change) { IsClawbackEnabled: clawback, LedgerNumber: p.checkpointLedger, }) + p.sawSACBalance = true p.entries++ } } @@ -594,6 +600,20 @@ func (p *checkpointProcessor) finalize(ctx context.Context, dbTx pgx.Tx) error { return fmt.Errorf("storing tokens in postgres: %w", err) } + // 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 deleted > 0 { + log.Ctx(ctx).Warnf("checkpoint: deleted %d SAC balance(s) for contracts not verified as SAC", deleted) + } + } + // 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..89b1079ca 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" @@ -375,12 +376,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 +431,69 @@ 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 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) - 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() + // 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() - f.sacBalanceModel.On("BatchCopy", mock.Anything, mock.Anything, mock.MatchedBy(func(b []wbdata.SACBalance) bool { return len(b) == 1 })).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() + // 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() + // 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() - 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) { - 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) { +// 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) 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) - 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 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) 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 +}