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 {

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

@@ -140,32 +140,27 @@ 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
 }
 
 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)