fix(ingest): verify SAC before recording contract balances

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.

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

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 {

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)
+}

internal/ingest/ingest.go

@@ -171,7 +171,7 @@
 	// must not mutate or remove the policies the live pods rely on; and an active
 	// retention policy would drop the very history a backfill is writing.
 	if cfg.IngestionMode == services.IngestionModeLive {
 		if err := configureHypertableSettings(ctx, dbConnectionPool, cfg.ChunkInterval, cfg.RetentionPeriod, cfg.OldestLedgerCursorName, cfg.CompressionScheduleInterval, cfg.CompressAfter, cfg.MaxChunksToCompress); err != nil {
 			return nil, nil, fmt.Errorf("configuring hypertable settings: %w", err)
 		}
 	}
@@ -233,6 +233,7 @@
 		TrustlineBalanceModel:     models.TrustlineBalance,
 		NativeBalanceModel:        models.NativeBalance,
 		SACBalanceModel:           models.SACBalance,
+		ContractModel:             models.Contract,
 		LiquidityPoolModel:        models.LiquidityPool,
 		LiquidityPoolBalanceModel: models.LiquidityPoolBalance,
 		NetworkPassphrase:         cfg.NetworkPassphrase,

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)