diff --git a/cmd/ingest.go b/cmd/ingest.go index 35de96210..e25a416cd 100644 --- a/cmd/ingest.go +++ b/cmd/ingest.go @@ -35,6 +35,8 @@ func (c *ingestCmd) Command() *cobra.Command { utils.IngestServerPortOption(&cfg.ServerPort), utils.AdminPortOption(&cfg.AdminPort), utils.GetLedgersLimitOption(&cfg.GetLedgersLimit), + utils.BlendPriceIntervalOption(&cfg.BlendPriceInterval), + utils.BlendBackstopLPContractIDOption(&cfg.BlendBackstopLPContractID), { Name: "ingestion-mode", Usage: "What mode to run ingestion in - live or backfill", diff --git a/cmd/utils/global_options.go b/cmd/utils/global_options.go index 2e34be257..9ab20fd17 100644 --- a/cmd/utils/global_options.go +++ b/cmd/utils/global_options.go @@ -153,6 +153,34 @@ func NetworkOption(configKey *string) *config.ConfigOption { } } +// BlendPriceIntervalOption returns the config option for the wait between Blend v2 oracle +// price snapshot passes. Setting it to 0 disables the snapshot task entirely. +func BlendPriceIntervalOption(configKey *time.Duration) *config.ConfigOption { + return &config.ConfigOption{ + Name: "blend-price-interval", + Usage: `Interval between Blend v2 oracle price snapshot passes (Go duration string, e.g. "60s"). 0 disables the snapshot task.`, + OptType: types.String, + CustomSetValue: SetConfigOptionDuration, + ConfigKey: configKey, + FlagDefault: "60s", + Required: false, + } +} + +// BlendBackstopLPContractIDOption returns the config option for the Blend v2 backstop's +// Comet BLND:USDC weighted pool. Its C-address enables the price snapshot task's BLND/LP-share +// derived-pricing leg; leaving it empty disables that leg. +func BlendBackstopLPContractIDOption(configKey *string) *config.ConfigOption { + return &config.ConfigOption{ + Name: "blend-backstop-lp-contract-id", + Usage: "C-address of the Blend v2 backstop's Comet BLND:USDC weighted pool, enabling the BLND/LP-share price leg. Empty disables it.", + OptType: types.String, + ConfigKey: configKey, + FlagDefault: "", + Required: false, + } +} + func GraphQLComplexityLimitOption(configKey *int) *config.ConfigOption { return &config.ConfigOption{ Name: "graphql-complexity-limit", diff --git a/docker-compose.yaml b/docker-compose.yaml index bf402fe09..e967b03bb 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -175,6 +175,8 @@ services: TRACKER_DSN: ${TRACKER_DSN} STELLAR_ENVIRONMENT: ${STELLAR_ENVIRONMENT:-development} # ADMIN_PORT: 9094 # Optional: Port for pprof endpoints at /debug/pprof. WARNING: Do NOT expose publicly! + # BLEND_PRICE_INTERVAL: 60s # Optional: interval between Blend v2 oracle price snapshot passes. 0 disables it. + # BLEND_BACKSTOP_LP_CONTRACT_ID: "" # Optional: Comet BLND:USDC LP C-address enabling the BLND/LP price leg. ingest-mainnet: container_name: ingest-mainnet @@ -217,6 +219,8 @@ services: TRACKER_DSN: ${TRACKER_DSN} STELLAR_ENVIRONMENT: ${STELLAR_ENVIRONMENT:-development} # ADMIN_PORT: 9095 # Optional: Port for pprof endpoints at /debug/pprof. WARNING: Do NOT expose publicly! + # BLEND_PRICE_INTERVAL: 60s # Optional: interval between Blend v2 oracle price snapshot passes. 0 disables it. + # BLEND_BACKSTOP_LP_CONTRACT_ID: "" # Optional: Comet BLND:USDC LP C-address enabling the BLND/LP price leg. volumes: postgres-db: driver: local diff --git a/internal/data/blend/models.go b/internal/data/blend/models.go index 9f039747e..09aa079d1 100644 --- a/internal/data/blend/models.go +++ b/internal/data/blend/models.go @@ -19,7 +19,7 @@ type Models struct { PoolClaimed *PoolClaimedModel BackstopClaimed *BackstopClaimedModel Auctions *AuctionModel - // OraclePrices *OraclePriceModel is added in PR4 alongside oracle_prices.go. + OraclePrices *OraclePriceModel } // NewModels constructs the Blend Models aggregate wired to the given pool/metrics. @@ -35,5 +35,6 @@ func NewModels(pool *pgxpool.Pool, dbMetrics *metrics.DBMetrics) Models { PoolClaimed: &PoolClaimedModel{DB: pool, Metrics: dbMetrics}, BackstopClaimed: &BackstopClaimedModel{DB: pool, Metrics: dbMetrics}, Auctions: &AuctionModel{DB: pool, Metrics: dbMetrics}, + OraclePrices: &OraclePriceModel{DB: pool, Metrics: dbMetrics}, } } diff --git a/internal/data/blend/oracle_prices.go b/internal/data/blend/oracle_prices.go new file mode 100644 index 000000000..b56b0d9c5 --- /dev/null +++ b/internal/data/blend/oracle_prices.go @@ -0,0 +1,151 @@ +// Oracle price snapshot storage for Blend v2 (blend_oracle_prices). +package blend + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/stellar/wallet-backend/internal/indexer/types" + "github.com/stellar/wallet-backend/internal/metrics" + "github.com/stellar/wallet-backend/internal/utils" +) + +// oraclePricesTable is the metrics label for blend_oracle_prices queries. +const oraclePricesTable = "blend_oracle_prices" + +// PriceTarget identifies one (oracle, asset) pair that some pool's reserve +// prices against — the input list the SEP-40 lastprice snapshot task queries +// on each tick. +type PriceTarget struct { + OracleContractID types.AddressBytea + AssetContractID types.AddressBytea +} + +// OraclePrice is a full row of blend_oracle_prices: the current-only (not +// historical) latest known SEP-40 price for one (oracle, asset) pair. +type OraclePrice struct { + OracleContractID types.AddressBytea + AssetContractID types.AddressBytea + Price string // raw i128 fixed-point value at PriceDecimals + PriceDecimals int32 + PriceTimestamp int64 // oracle-reported timestamp + // UpdatedAt is populated when scanning an existing row. BatchUpsert ignores + // it on write and always sets updated_at to NOW() itself. + UpdatedAt time.Time +} + +// OraclePriceModelInterface exposes Blend v2 oracle price snapshot storage +// operations. +// +// Unlike sibling Blend models, both methods here operate directly on the +// pgxpool rather than a caller-supplied pgx.Tx: the price snapshot task runs +// on its own schedule, independent of ledger ingestion, so there is no +// enclosing ingest transaction for it to join. +type OraclePriceModelInterface interface { + // GetPriceTargets returns the deduplicated set of (oracle, asset) pairs + // that some pool's reserve prices against — every blend_reserves row + // joined to its pool's oracle. Pools with no oracle wired yet (NULL + // oracle_contract_id) are excluded. + GetPriceTargets(ctx context.Context) ([]PriceTarget, error) + // BatchUpsert inserts or fully replaces price rows keyed by + // (oracle_contract_id, asset_contract_id). updated_at is always set to + // NOW(), whether the row is freshly inserted or replaced. + BatchUpsert(ctx context.Context, rows []OraclePrice) error +} + +// OraclePriceModel implements OraclePriceModelInterface against blend_oracle_prices. +type OraclePriceModel struct { + DB *pgxpool.Pool + Metrics *metrics.DBMetrics +} + +var _ OraclePriceModelInterface = (*OraclePriceModel)(nil) + +// GetPriceTargets returns the deduplicated (oracle, asset) pairs derived from +// blend_reserves joined to their pool's oracle. See OraclePriceModelInterface. +func (m *OraclePriceModel) GetPriceTargets(ctx context.Context) ([]PriceTarget, error) { + start := time.Now() + + const query = ` + SELECT DISTINCT p.oracle_contract_id, r.asset_contract_id + FROM blend_pools p + JOIN blend_reserves r ON r.pool_contract_id = p.pool_contract_id + WHERE p.oracle_contract_id IS NOT NULL + ORDER BY p.oracle_contract_id, r.asset_contract_id` + rows, err := m.DB.Query(ctx, query) + if err != nil { + m.Metrics.QueryErrors.WithLabelValues("GetPriceTargets", oraclePricesTable, utils.GetDBErrorType(err)).Inc() + return nil, fmt.Errorf("querying blend oracle price targets: %w", err) + } + defer rows.Close() + + var targets []PriceTarget + for rows.Next() { + var t PriceTarget + if err := rows.Scan(&t.OracleContractID, &t.AssetContractID); err != nil { + m.Metrics.QueryErrors.WithLabelValues("GetPriceTargets", oraclePricesTable, utils.GetDBErrorType(err)).Inc() + return nil, fmt.Errorf("scanning blend oracle price target row: %w", err) + } + targets = append(targets, t) + } + if err := rows.Err(); err != nil { + m.Metrics.QueryErrors.WithLabelValues("GetPriceTargets", oraclePricesTable, utils.GetDBErrorType(err)).Inc() + return nil, fmt.Errorf("iterating blend oracle price target rows: %w", err) + } + + duration := time.Since(start).Seconds() + m.Metrics.QueryDuration.WithLabelValues("GetPriceTargets", oraclePricesTable).Observe(duration) + m.Metrics.QueriesTotal.WithLabelValues("GetPriceTargets", oraclePricesTable).Inc() + return targets, nil +} + +// BatchUpsert inserts or fully replaces blend_oracle_prices rows. See OraclePriceModelInterface. +func (m *OraclePriceModel) BatchUpsert(ctx context.Context, rows []OraclePrice) error { + if len(rows) == 0 { + return nil + } + + start := time.Now() + + oracles := make([][]byte, len(rows)) + assets := make([][]byte, len(rows)) + prices := make([]string, len(rows)) + decimals := make([]int32, len(rows)) + timestamps := make([]int64, len(rows)) + for i, r := range rows { + oracleBytes, err := addressToBytes(string(r.OracleContractID)) + if err != nil { + return fmt.Errorf("converting oracle address for price upsert: %w", err) + } + assetBytes, err := addressToBytes(string(r.AssetContractID)) + if err != nil { + return fmt.Errorf("converting asset address for price upsert: %w", err) + } + oracles[i] = oracleBytes + assets[i] = assetBytes + prices[i] = r.Price + decimals[i] = r.PriceDecimals + timestamps[i] = r.PriceTimestamp + } + + const upsertQuery = ` + INSERT INTO blend_oracle_prices (oracle_contract_id, asset_contract_id, price, price_decimals, price_timestamp, updated_at) + SELECT u.*, NOW() FROM UNNEST($1::bytea[], $2::bytea[], $3::text[], $4::integer[], $5::bigint[]) + AS u(oracle_contract_id, asset_contract_id, price, price_decimals, price_timestamp) + ON CONFLICT (oracle_contract_id, asset_contract_id) DO UPDATE SET + price = EXCLUDED.price, price_decimals = EXCLUDED.price_decimals, + price_timestamp = EXCLUDED.price_timestamp, updated_at = NOW()` + if _, err := m.DB.Exec(ctx, upsertQuery, oracles, assets, prices, decimals, timestamps); err != nil { + m.Metrics.QueryErrors.WithLabelValues("BatchUpsert", oraclePricesTable, utils.GetDBErrorType(err)).Inc() + return fmt.Errorf("upserting blend oracle prices: %w", err) + } + + duration := time.Since(start).Seconds() + m.Metrics.QueryDuration.WithLabelValues("BatchUpsert", oraclePricesTable).Observe(duration) + m.Metrics.QueriesTotal.WithLabelValues("BatchUpsert", oraclePricesTable).Inc() + m.Metrics.BatchSize.WithLabelValues("BatchUpsert", oraclePricesTable).Observe(float64(len(rows))) + return nil +} diff --git a/internal/data/blend/oracle_prices_test.go b/internal/data/blend/oracle_prices_test.go new file mode 100644 index 000000000..87f077713 --- /dev/null +++ b/internal/data/blend/oracle_prices_test.go @@ -0,0 +1,173 @@ +// Unit tests for the Blend v2 OraclePriceModel. +// These tests exercise real SQL and require a PostgreSQL test database. +// Uses an external test package to avoid an import cycle with internal/data. +package blend_test + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/prometheus/client_golang/prometheus" + "github.com/stellar/go-stellar-sdk/keypair" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/stellar/wallet-backend/internal/data/blend" + "github.com/stellar/wallet-backend/internal/db" + "github.com/stellar/wallet-backend/internal/db/dbtest" + "github.com/stellar/wallet-backend/internal/indexer/types" + "github.com/stellar/wallet-backend/internal/metrics" +) + +func newOraclePricesFixture(t *testing.T) (context.Context, *pgxpool.Pool, *blend.OraclePriceModel, func()) { + t.Helper() + ctx := context.Background() + + dbt := dbtest.Open(t) + pool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) + require.NoError(t, err) + + m := &blend.OraclePriceModel{ + DB: pool, + Metrics: metrics.NewMetrics(prometheus.NewRegistry()).DB, + } + + cleanup := func() { + pool.Close() + dbt.Close() + } + return ctx, pool, m, cleanup +} + +// insertPool seeds a blend_pools row directly via raw SQL. An empty oracleAddr +// stores oracle_contract_id as SQL NULL. +func insertPool(t *testing.T, ctx context.Context, pool *pgxpool.Pool, poolAddr, oracleAddr string) { + t.Helper() + var oracle any + if oracleAddr != "" { + oracle = types.AddressBytea(oracleAddr) + } + _, err := pool.Exec(ctx, ` + INSERT INTO blend_pools (pool_contract_id, oracle_contract_id, last_modified_ledger) + VALUES ($1, $2, 1) + `, types.AddressBytea(poolAddr), oracle) + require.NoError(t, err) +} + +type oraclePriceRow struct { + Price string + PriceDecimals int32 + PriceTimestamp int64 + UpdatedAt time.Time +} + +func getOraclePrice(t *testing.T, ctx context.Context, pool *pgxpool.Pool, oracleAddr, assetAddr string) (oraclePriceRow, bool) { + t.Helper() + var row oraclePriceRow + err := pool.QueryRow(ctx, ` + SELECT price, price_decimals, price_timestamp, updated_at + FROM blend_oracle_prices WHERE oracle_contract_id = $1 AND asset_contract_id = $2 + `, types.AddressBytea(oracleAddr), types.AddressBytea(assetAddr)).Scan( + &row.Price, &row.PriceDecimals, &row.PriceTimestamp, &row.UpdatedAt, + ) + if err != nil { + return oraclePriceRow{}, false + } + return row, true +} + +func TestGetPriceTargets(t *testing.T) { + ctx, pool, m, cleanup := newOraclePricesFixture(t) + defer cleanup() + + oracle1 := keypair.MustRandom().Address() + oracle2 := keypair.MustRandom().Address() + poolA := keypair.MustRandom().Address() + poolB := keypair.MustRandom().Address() + poolC := keypair.MustRandom().Address() + poolNoOracle := keypair.MustRandom().Address() + assetX := keypair.MustRandom().Address() + assetY := keypair.MustRandom().Address() + assetZ := keypair.MustRandom().Address() + assetExcluded := keypair.MustRandom().Address() + + // poolA and poolB share oracle1. + insertPool(t, ctx, pool, poolA, oracle1) + insertPool(t, ctx, pool, poolB, oracle1) + // poolC has its own oracle. + insertPool(t, ctx, pool, poolC, oracle2) + // poolNoOracle has no oracle wired yet; its reserves must be excluded. + insertPool(t, ctx, pool, poolNoOracle, "") + + insertReserve(t, ctx, pool, poolA, assetX, 0, "1", "1") + insertReserve(t, ctx, pool, poolB, assetX, 0, "1", "1") // same asset via a different pool -> dedup + insertReserve(t, ctx, pool, poolB, assetY, 1, "1", "1") + insertReserve(t, ctx, pool, poolC, assetZ, 0, "1", "1") + insertReserve(t, ctx, pool, poolNoOracle, assetExcluded, 0, "1", "1") + + targets, err := m.GetPriceTargets(ctx) + require.NoError(t, err) + + want := []blend.PriceTarget{ + {OracleContractID: types.AddressBytea(oracle1), AssetContractID: types.AddressBytea(assetX)}, + {OracleContractID: types.AddressBytea(oracle1), AssetContractID: types.AddressBytea(assetY)}, + {OracleContractID: types.AddressBytea(oracle2), AssetContractID: types.AddressBytea(assetZ)}, + } + assert.ElementsMatch(t, want, targets) +} + +func TestOraclePriceBatchUpsert(t *testing.T) { + ctx, pool, m, cleanup := newOraclePricesFixture(t) + defer cleanup() + + oracleAddr := keypair.MustRandom().Address() + assetA := keypair.MustRandom().Address() + assetB := keypair.MustRandom().Address() + + require.NoError(t, m.BatchUpsert(ctx, []blend.OraclePrice{ + { + OracleContractID: types.AddressBytea(oracleAddr), AssetContractID: types.AddressBytea(assetA), + Price: "100000000", PriceDecimals: 7, PriceTimestamp: 1000, + }, + { + OracleContractID: types.AddressBytea(oracleAddr), AssetContractID: types.AddressBytea(assetB), + Price: "200000000", PriceDecimals: 7, PriceTimestamp: 1000, + }, + })) + + rowA, ok := getOraclePrice(t, ctx, pool, oracleAddr, assetA) + require.True(t, ok) + assert.Equal(t, "100000000", rowA.Price) + assert.Equal(t, int64(1000), rowA.PriceTimestamp) + + rowB, ok := getOraclePrice(t, ctx, pool, oracleAddr, assetB) + require.True(t, ok) + assert.Equal(t, "200000000", rowB.Price) + + // Ensure NOW() is measurably different on the re-upsert below. + time.Sleep(10 * time.Millisecond) + + require.NoError(t, m.BatchUpsert(ctx, []blend.OraclePrice{ + { + OracleContractID: types.AddressBytea(oracleAddr), AssetContractID: types.AddressBytea(assetA), + Price: "150000000", PriceDecimals: 7, PriceTimestamp: 2000, + }, + })) + + rowA2, ok := getOraclePrice(t, ctx, pool, oracleAddr, assetA) + require.True(t, ok) + assert.Equal(t, "150000000", rowA2.Price, "re-upserted row's price must change") + assert.Equal(t, int64(2000), rowA2.PriceTimestamp, "re-upserted row's timestamp must change") + assert.True(t, rowA2.UpdatedAt.After(rowA.UpdatedAt), "updated_at must advance on re-upsert") + + rowB2, ok := getOraclePrice(t, ctx, pool, oracleAddr, assetB) + require.True(t, ok) + assert.Equal(t, "200000000", rowB2.Price, "untouched row's price must not change") + assert.Equal(t, rowB.UpdatedAt, rowB2.UpdatedAt, "untouched row's updated_at must not change") + + t.Run("is a no-op when no rows are staged", func(t *testing.T) { + require.NoError(t, m.BatchUpsert(ctx, nil)) + }) +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index eeaf0aff9..b9bd69ecf 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -24,7 +24,7 @@ import ( "github.com/stellar/wallet-backend/internal/metrics" httphandler "github.com/stellar/wallet-backend/internal/serve/httphandler" "github.com/stellar/wallet-backend/internal/services" - _ "github.com/stellar/wallet-backend/internal/services/blend" // registers BLEND validator + processor via init() + "github.com/stellar/wallet-backend/internal/services/blend" // also registers BLEND validator + processor via init() _ "github.com/stellar/wallet-backend/internal/services/sep41" // registers SEP-41 validator + processor via init() ) @@ -92,6 +92,12 @@ type Configs struct { DBMinConns int DBMaxConnLifetime time.Duration DBMaxConnIdleTime time.Duration + // BlendPriceInterval is the wait between Blend v2 oracle price snapshot passes. + // 0 disables the snapshot task. Only takes effect in live ingestion mode. + BlendPriceInterval time.Duration + // BlendBackstopLPContractID is the Blend v2 backstop's Comet BLND:USDC weighted pool + // C-address, enabling the price snapshot task's BLND/LP-share derived-pricing leg. + BlendBackstopLPContractID string } func (c Configs) BuildPoolConfig() db.PoolConfig { @@ -169,6 +175,25 @@ func setupDeps(cfg Configs) (services.IngestService, error) { return nil, fmt.Errorf("instantiating contract metadata service: %w", err) } + // The Blend v2 oracle price snapshot task. Live-only: backfill replays historical + // ledgers and has no use for a periodic wall-clock snapshot of current prices. + var postLockTasks []func(context.Context) + if cfg.IngestionMode == services.IngestionModeLive && cfg.BlendPriceInterval > 0 { + blendPrices, err := blend.NewPriceSnapshotService(blend.PriceSnapshotConfig{ + OraclePrices: models.Blend.OraclePrices, + Metadata: contractMetadataService, + Interval: cfg.BlendPriceInterval, + BackstopLPContractID: cfg.BlendBackstopLPContractID, + Metrics: m.BlendPrices, + }) + if err != nil { + return nil, fmt.Errorf("instantiating blend price snapshot service: %w", err) + } + // Started by the ingest service only after it acquires the live + // advisory lock — a lock-losing instance must not snapshot prices. + postLockTasks = append(postLockTasks, blendPrices.Run) + } + // Build a single ProtocolDeps to pass through both the validator and // processor registries. cmd/ingest knows nothing about specific // protocols — adding a new one is a blank import elsewhere plus a SQL @@ -270,6 +295,7 @@ func setupDeps(cfg Configs) (services.IngestService, error) { ProtocolValidators: protocolValidators, WasmSpecExtractor: wasmExtractor, ContractMetadataService: contractMetadataService, + PostLockTasks: postLockTasks, }) if err != nil { return nil, fmt.Errorf("instantiating ingest service: %w", err) diff --git a/internal/metrics/blend.go b/internal/metrics/blend.go new file mode 100644 index 000000000..b0cab73fd --- /dev/null +++ b/internal/metrics/blend.go @@ -0,0 +1,53 @@ +package metrics + +import "github.com/prometheus/client_golang/prometheus" + +// BlendPriceMetrics holds Prometheus collectors for the Blend v2 oracle price +// snapshot service. Unlike Migration/Ingestion, this runs on its own ticker +// independent of ledger ingestion (see services/blend.PriceSnapshotService), +// so it carries no ledger-sequence-shaped metrics of its own. +type BlendPriceMetrics struct { + // SnapshotDuration observes one full SnapshotOnce pass's wall time + // (target discovery + every oracle/Comet fetch + the final batch upsert). + // PromQL: histogram_quantile(0.99, rate(wallet_blend_price_snapshot_duration_seconds_bucket[$__rate_interval])) + SnapshotDuration prometheus.Histogram + // FetchesTotal counts per-(oracle,asset) lastprice fetch outcomes (and the + // Comet leg's single derived-valuation outcome), labeled by result: + // success, error, none (SEP-40 Option::None — no price recorded yet), + // stale (oracle-reported timestamp older than the pool contract's 24h + // rejection window), or invalid (price ≤ 0). + // PromQL: rate(wallet_blend_price_fetches_total{result="error"}[$__rate_interval]) + FetchesTotal *prometheus.CounterVec + // PricesTracked is the row count written by the most recent snapshot pass. + // PromQL: wallet_blend_prices_tracked + PricesTracked prometheus.Gauge + // OldestPriceAge is the age in seconds of the oldest oracle-reported price + // observed by the most recent snapshot pass, including prices skipped as + // stale — a dead oracle shows up as this gauge growing without bound. + // PromQL: wallet_blend_price_oldest_age_seconds + OldestPriceAge prometheus.Gauge +} + +func newBlendPriceMetrics(reg prometheus.Registerer) *BlendPriceMetrics { + m := &BlendPriceMetrics{ + SnapshotDuration: prometheus.NewHistogram(prometheus.HistogramOpts{ + Name: "wallet_blend_price_snapshot_duration_seconds", + Help: "Duration of a full Blend v2 oracle price snapshot pass.", + Buckets: prometheus.DefBuckets, + }), + FetchesTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "wallet_blend_price_fetches_total", + Help: "Blend v2 oracle price fetch outcomes, labeled by result (success, error, none, stale, invalid).", + }, []string{"result"}), + PricesTracked: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "wallet_blend_prices_tracked", + Help: "Number of (oracle, asset) price rows written by the most recent Blend v2 snapshot pass.", + }), + OldestPriceAge: prometheus.NewGauge(prometheus.GaugeOpts{ + Name: "wallet_blend_price_oldest_age_seconds", + Help: "Age in seconds of the oldest oracle-reported price observed by the most recent Blend v2 snapshot pass, including prices skipped as stale.", + }), + } + reg.MustRegister(m.SnapshotDuration, m.FetchesTotal, m.PricesTracked, m.OldestPriceAge) + return m +} diff --git a/internal/metrics/blend_test.go b/internal/metrics/blend_test.go new file mode 100644 index 000000000..c5183fdd4 --- /dev/null +++ b/internal/metrics/blend_test.go @@ -0,0 +1,47 @@ +package metrics + +import ( + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBlendPriceMetrics_Registration(t *testing.T) { + reg := prometheus.NewRegistry() + m := newBlendPriceMetrics(reg) + + require.NotNil(t, m.SnapshotDuration) + require.NotNil(t, m.FetchesTotal) + require.NotNil(t, m.PricesTracked) +} + +func TestBlendPriceMetrics_Record(t *testing.T) { + reg := prometheus.NewRegistry() + m := newBlendPriceMetrics(reg) + + m.SnapshotDuration.Observe(0.25) + m.FetchesTotal.WithLabelValues("success").Inc() + m.FetchesTotal.WithLabelValues("success").Inc() + m.FetchesTotal.WithLabelValues("error").Inc() + m.FetchesTotal.WithLabelValues("none").Inc() + m.PricesTracked.Set(2) + + assert.Equal(t, 2.0, testutil.ToFloat64(m.FetchesTotal.WithLabelValues("success"))) + assert.Equal(t, 1.0, testutil.ToFloat64(m.FetchesTotal.WithLabelValues("error"))) + assert.Equal(t, 1.0, testutil.ToFloat64(m.FetchesTotal.WithLabelValues("none"))) + assert.Equal(t, 2.0, testutil.ToFloat64(m.PricesTracked)) +} + +func TestBlendPriceMetrics_Lint(t *testing.T) { + reg := prometheus.NewRegistry() + m := newBlendPriceMetrics(reg) + + for _, c := range []prometheus.Collector{m.SnapshotDuration, m.FetchesTotal, m.PricesTracked} { + problems, err := testutil.CollectAndLint(c) + require.NoError(t, err) + assert.Empty(t, problems) + } +} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index bee78e568..08b59bfbe 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -9,14 +9,15 @@ import ( // Metrics holds all Prometheus collectors for the wallet-backend service. type Metrics struct { - DB *DBMetrics - RPC *RPCMetrics - Ingestion *IngestionMetrics - HTTP *HTTPMetrics - GraphQL *GraphQLMetrics - Auth *AuthMetrics - Migration *MigrationMetrics - registry *prometheus.Registry + DB *DBMetrics + RPC *RPCMetrics + Ingestion *IngestionMetrics + HTTP *HTTPMetrics + GraphQL *GraphQLMetrics + Auth *AuthMetrics + Migration *MigrationMetrics + BlendPrices *BlendPriceMetrics + registry *prometheus.Registry } // NewMetrics creates a new Metrics instance with all sub-struct collectors registered. @@ -28,14 +29,15 @@ func NewMetrics(reg *prometheus.Registry) *Metrics { collectors.NewProcessCollector(collectors.ProcessCollectorOpts{}), ) return &Metrics{ - DB: newDBMetrics(reg), - RPC: newRPCMetrics(reg), - Ingestion: newIngestionMetrics(reg), - HTTP: newHTTPMetrics(reg), - GraphQL: NewGraphQLMetrics(reg), - Auth: newAuthMetrics(reg), - Migration: newMigrationMetrics(reg), - registry: reg, + DB: newDBMetrics(reg), + RPC: newRPCMetrics(reg), + Ingestion: newIngestionMetrics(reg), + HTTP: newHTTPMetrics(reg), + GraphQL: NewGraphQLMetrics(reg), + Auth: newAuthMetrics(reg), + Migration: newMigrationMetrics(reg), + BlendPrices: newBlendPriceMetrics(reg), + registry: reg, } } diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index eff258f7c..d80ef84c9 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -21,6 +21,7 @@ func TestNewMetrics(t *testing.T) { require.NotNil(t, m.GraphQL) require.NotNil(t, m.Auth) require.NotNil(t, m.Migration) + require.NotNil(t, m.BlendPrices) assert.Same(t, reg, m.Registry()) } diff --git a/internal/services/blend/comet.go b/internal/services/blend/comet.go new file mode 100644 index 000000000..baf7fb75c --- /dev/null +++ b/internal/services/blend/comet.go @@ -0,0 +1,249 @@ +// Package blend — comet.go derives the BLND/USD spot price and the Comet +// backstop-deposit LP token's USD price from a Comet weighted pool's raw +// on-chain state. The Blend v2 backstop's deposit token is a Comet BLND:USDC +// weighted LP (mainnet +// CAS3FL6TLZKDGGSISDBWGGPXT3NRR4DYTZD7YOD3HMYO6LTJUVGRVEAM, testnet +// CA5UTUUPHYL5K22UBRUVC37EARZUGYOSGK3IKIXG2JLCC5ZZLI4BDWDM, wasm hash +// 8abc28913035c07411ed5d134e6bfeab4723d97ddd4d1a22a0605d35c94d1a36 — pinned +// 2026-07-08); no on-chain oracle prices it directly, so the price snapshot +// task derives it from the pool's own balances/weights instead. +// +// Verification (2026-07-08): `stellar contract info interface --contract-id +// CAS3FL6TLZKDGGSISDBWGGPXT3NRR4DYTZD7YOD3HMYO6LTJUVGRVEAM --rpc-url +// https://mainnet.sorobanrpc.com --network-passphrase "Public Global Stellar +// Network ; September 2015"` (stellar-cli 27.0.0) against the live mainnet +// contract, cross-checked against CometDEX/comet-contracts-v1 (GitHub, +// contracts/src/c_pool/comet.rs and contracts/src/c_consts.rs) confirms: +// +// - get_tokens(env: Env) -> Vec
— no-arg +// - get_balance(env: Env, token: Address) -> i128 — 1-arg +// - get_normalized_weight(env: Env, token: Address) -> i128 — 1-arg; +// comet.rs's own doc comment: "Get the weight of the token in decimal +// form with 7 decimals". c_consts.rs defines `STROOP: i128 = 10^7` as +// Comet's fixed-point scale, confirming the weight scalar is 7 decimals +// — the same scale Comet uses for balances and get_total_supply (the LP +// token's total shares), and the scale this file's own inputs/outputs use. +// - get_total_supply(env: Env) -> i128 — no-arg +// +// services.ContractMetadataService.FetchSingleField already accepts +// variadic xdr.ScVal arguments, so the two 1-arg getters need no plumbing +// changes — fetchCometState calls them directly with a single Address +// argument built by contractAddressScVal (scval.go). +package blend + +import ( + "context" + "fmt" + "math/big" + + "github.com/stellar/wallet-backend/internal/services" +) + +// cometPriceDecimals is the fixed-point precision cometValuation's outputs +// use — 7 decimals, matching Comet's own STROOP scale (see package doc) so +// callers can treat blndPrice/lpPrice the same way they treat any other +// on-chain 7-decimal amount (i128String's raw-integer-string convention, +// also used by blend_oracle_prices.Price). +const cometPriceDecimals = 7 + +// cometTokenCount is the number of legs a Comet BLND:USDC weighted pool +// has. fetchCometState errors on any other count rather than guessing which +// tokens to treat as BLND/USDC. +const cometTokenCount = 2 + +// cometState is a Comet weighted pool's raw on-chain balances, weights, and +// LP total supply, already split into BLND and USDC legs. Balance/weight/ +// supply fields are 7-decimal fixed-point big.Ints (Comet's STROOP scale) — +// the same shape cometValuation consumes. BLNDAddress is the strkey +// C-address of the BLND leg's token contract (get_tokens()[blndIdx]) — the +// price snapshot task (prices.go) needs it to key the derived BLND price row +// by the real BLND token address rather than the Comet pool's own address. +type cometState struct { + BLNDAddress string + BLNDBalance *big.Int + USDCBalance *big.Int + BLNDWeight *big.Int + USDCWeight *big.Int + LPSupply *big.Int +} + +// cometValuation derives the BLND spot price (denominated in USDC, taken as +// USD by convention — this pool has no on-chain USDC/USD leg of its own) and +// the Comet LP token's USD price from a weighted two-token Comet pool's raw +// balances/weights/supply. +// +// Spot price (Balancer-style weighted pool; see comet-contracts-v1's +// calc_spot_price, and the identical formula in the Balancer V1 whitepaper): +// for legs with balances B and normalized weights W, +// spot_price(quote, base) = (B_quote/W_quote) / (B_base/W_base). Here the +// quote leg is USDC and the base leg is BLND, so: +// +// blndPrice = (usdcBal/usdcWeight) / (blndBal/blndWeight) +// +// LP token USD price is the pool's total USD value divided by lpSupply, +// where pool value = blndBal*blndPrice + usdcBal (USDC contributes at its +// own balance since USDC ≡ $1). This NAV-per-share definition holds for any +// weight split — it does not depend on blndWeight/usdcWeight directly, only +// through blndPrice. +// +// Inputs are 7-decimal fixed-point big.Ints (Comet's STROOP scale); outputs +// are 7-decimal fixed-point decimal strings (e.g. "2000000" for 0.2, not +// "0.2000000" — see i128String). All arithmetic is done with big.Rat to +// avoid intermediate overflow/precision loss, matching real-world Comet +// balances that run into the trillions of raw units. +// +// Returns an error — never a panic — for a nil, zero, or negative balance, +// weight, or lpSupply: valuing a pool with no liquidity, no weight, or no +// outstanding shares is meaningless, not just numerically undefined. +func cometValuation(blndBal, usdcBal, blndWeight, usdcWeight, lpSupply *big.Int) (blndPrice, lpPrice string, err error) { + inputs := []struct { + name string + val *big.Int + }{ + {"blndBal", blndBal}, + {"usdcBal", usdcBal}, + {"blndWeight", blndWeight}, + {"usdcWeight", usdcWeight}, + {"lpSupply", lpSupply}, + } + for _, in := range inputs { + if in.val == nil { + return "", "", fmt.Errorf("blend: cometValuation: %s is nil", in.name) + } + if in.val.Sign() <= 0 { + return "", "", fmt.Errorf("blend: cometValuation: %s must be positive, got %s", in.name, in.val) + } + } + + // blndPriceRat = (usdcBal/usdcWeight) / (blndBal/blndWeight) + // = (usdcBal * blndWeight) / (usdcWeight * blndBal) + num := new(big.Int).Mul(usdcBal, blndWeight) + den := new(big.Int).Mul(usdcWeight, blndBal) + blndPriceRat := new(big.Rat).SetFrac(num, den) + + blndValue := new(big.Rat).Mul(new(big.Rat).SetInt(blndBal), blndPriceRat) + poolValue := new(big.Rat).Add(blndValue, new(big.Rat).SetInt(usdcBal)) + lpPriceRat := new(big.Rat).Quo(poolValue, new(big.Rat).SetInt(lpSupply)) + + return ratToFixedString(blndPriceRat, cometPriceDecimals), ratToFixedString(lpPriceRat, cometPriceDecimals), nil +} + +// ratToFixedString converts r into a base-10 fixed-point integer string at +// decimals digits of precision — e.g. 0.2 at 7 decimals renders as +// "2000000", the same "raw integer, no decimal point" convention i128String +// and blend_oracle_prices.Price use elsewhere in this codebase. Rounds to +// the nearest representable value, with exact halves rounding away from +// zero. +func ratToFixedString(r *big.Rat, decimals int) string { + scale := new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(decimals)), nil) + scaled := new(big.Rat).Mul(r, new(big.Rat).SetInt(scale)) + + // big.Rat always normalizes with a positive denominator, so num's sign + // is scaled's sign and QuoRem (truncating/T-division) leaves rem with + // that same sign. + num := scaled.Num() + den := scaled.Denom() + q, rem := new(big.Int).QuoRem(num, den, new(big.Int)) + + twiceRem := new(big.Int).Lsh(new(big.Int).Abs(rem), 1) + if twiceRem.Cmp(den) >= 0 { + if num.Sign() < 0 { + q.Sub(q, big.NewInt(1)) + } else { + q.Add(q, big.NewInt(1)) + } + } + return q.String() +} + +// fetchCometState calls the Comet getters (get_tokens, get_balance, +// get_normalized_weight, get_total_supply — see package doc for verified +// signatures) via metadata against the Comet pool at cometID, and splits the +// two legs into BLND and USDC by weight: the higher-weighted leg is BLND +// (the pinned Comet pool is an 80/20 BLND:USDC split), regardless of which +// index get_tokens() happens to return it at. Any RPC failure, unexpected +// return shape, token count other than 2, or a tie between the two weights +// (which would make the BLND/USDC split ambiguous) is reported as an error. +func fetchCometState(ctx context.Context, metadata services.ContractMetadataService, cometID string) (*cometState, error) { + if metadata == nil { + return nil, fmt.Errorf("blend: fetchCometState: nil ContractMetadataService") + } + + tokensVal, err := metadata.FetchSingleField(ctx, cometID, "get_tokens") + if err != nil { + return nil, fmt.Errorf("blend: fetchCometState: fetching get_tokens: %w", err) + } + tokenVals, ok := vecVal(tokensVal) + if !ok { + return nil, fmt.Errorf("blend: fetchCometState: get_tokens: expected Vec, got %v", tokensVal.Type) + } + if len(tokenVals) != cometTokenCount { + return nil, fmt.Errorf("blend: fetchCometState: expected %d tokens in Comet pool %s, got %d", cometTokenCount, cometID, len(tokenVals)) + } + + tokens := make([]string, cometTokenCount) + for i, tv := range tokenVals { + addr, ok := addrString(tv) + if !ok { + return nil, fmt.Errorf("blend: fetchCometState: get_tokens[%d]: expected Address, got %v", i, tv.Type) + } + tokens[i] = addr + } + + balances := make([]*big.Int, cometTokenCount) + weights := make([]*big.Int, cometTokenCount) + for i, tok := range tokens { + argVal, err := contractAddressScVal(tok) + if err != nil { + return nil, fmt.Errorf("blend: fetchCometState: encoding token %s argument: %w", tok, err) + } + + balVal, err := metadata.FetchSingleField(ctx, cometID, "get_balance", argVal) + if err != nil { + return nil, fmt.Errorf("blend: fetchCometState: fetching get_balance(%s): %w", tok, err) + } + balParts, ok := balVal.GetI128() + if !ok { + return nil, fmt.Errorf("blend: fetchCometState: get_balance(%s): expected i128, got %v", tok, balVal.Type) + } + balances[i] = i128ToBigInt(balParts) + + weightVal, err := metadata.FetchSingleField(ctx, cometID, "get_normalized_weight", argVal) + if err != nil { + return nil, fmt.Errorf("blend: fetchCometState: fetching get_normalized_weight(%s): %w", tok, err) + } + weightParts, ok := weightVal.GetI128() + if !ok { + return nil, fmt.Errorf("blend: fetchCometState: get_normalized_weight(%s): expected i128, got %v", tok, weightVal.Type) + } + weights[i] = i128ToBigInt(weightParts) + } + + supplyVal, err := metadata.FetchSingleField(ctx, cometID, "get_total_supply") + if err != nil { + return nil, fmt.Errorf("blend: fetchCometState: fetching get_total_supply: %w", err) + } + supplyParts, ok := supplyVal.GetI128() + if !ok { + return nil, fmt.Errorf("blend: fetchCometState: get_total_supply: expected i128, got %v", supplyVal.Type) + } + lpSupply := i128ToBigInt(supplyParts) + + blndIdx := 0 + switch weights[0].Cmp(weights[1]) { + case 0: + return nil, fmt.Errorf("blend: fetchCometState: cannot identify BLND leg: both tokens have equal weight %s", weights[0]) + case -1: + blndIdx = 1 + } + usdcIdx := 1 - blndIdx + + return &cometState{ + BLNDAddress: tokens[blndIdx], + BLNDBalance: balances[blndIdx], + USDCBalance: balances[usdcIdx], + BLNDWeight: weights[blndIdx], + USDCWeight: weights[usdcIdx], + LPSupply: lpSupply, + }, nil +} diff --git a/internal/services/blend/comet_test.go b/internal/services/blend/comet_test.go new file mode 100644 index 000000000..a1abfad33 --- /dev/null +++ b/internal/services/blend/comet_test.go @@ -0,0 +1,301 @@ +package blend + +import ( + "context" + "math/big" + "testing" + + "github.com/stellar/go-stellar-sdk/xdr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/stellar/wallet-backend/internal/services" +) + +// bigFromStroop scales a float amount into a 7-decimal ("STROOP") fixed-point +// big.Int, matching how Comet balances, get_normalized_weight, and +// get_total_supply are all denominated on-chain (see comet.go's verification +// note). Test-only: real callers get their big.Ints from i128ToBigInt. +func bigFromStroop(amount float64) *big.Int { + scaled := new(big.Rat).Mul(new(big.Rat).SetFloat64(amount), new(big.Rat).SetInt64(1e7)) + // amount is always an exact multiple of 1e-7 in these tests, so this is exact. + return new(big.Int).Quo(scaled.Num(), scaled.Denom()) +} + +func TestRatToFixedString(t *testing.T) { + t.Run("exact integer", func(t *testing.T) { + assert.Equal(t, "10000000", ratToFixedString(big.NewRat(1, 1), 7)) + }) + + t.Run("rounds down below half", func(t *testing.T) { + // 1/3 at 0 decimals = 0.333... -> 0 + assert.Equal(t, "0", ratToFixedString(big.NewRat(1, 3), 0)) + }) + + t.Run("rounds up at exactly half, away from zero", func(t *testing.T) { + assert.Equal(t, "1", ratToFixedString(big.NewRat(1, 2), 0)) + assert.Equal(t, "-1", ratToFixedString(big.NewRat(-1, 2), 0)) + }) + + t.Run("rounds up above half", func(t *testing.T) { + // 2/3 at 0 decimals = 0.666... -> 1 + assert.Equal(t, "1", ratToFixedString(big.NewRat(2, 3), 0)) + }) + + t.Run("zero", func(t *testing.T) { + assert.Equal(t, "0", ratToFixedString(big.NewRat(0, 1), 7)) + }) +} + +// TestCometValuation is pure math — no mocks. Expected values were +// cross-checked with exact rational arithmetic (Python's fractions module) +// independently of the Go implementation. +func TestCometValuation(t *testing.T) { + t.Run("canonical 80/20 pool", func(t *testing.T) { + // blndBal=4,000,000, usdcBal=200,000, weights 0.8/0.2, lpSupply=1,000,000. + // blndPrice = (200k/0.2)/(4M/0.8) = 1,000,000/5,000,000 = 0.2 USD. + // lpPrice = (4M*0.2 + 200k)/1M = (800k+200k)/1M = 1.0 USD. + blndPrice, lpPrice, err := cometValuation( + bigFromStroop(4_000_000), bigFromStroop(200_000), + bigFromStroop(0.8), bigFromStroop(0.2), + bigFromStroop(1_000_000), + ) + require.NoError(t, err) + assert.Equal(t, "2000000", blndPrice) + assert.Equal(t, "10000000", lpPrice) + }) + + t.Run("uneven supply, non-terminating spot price", func(t *testing.T) { + // blndBal=3,000,000, usdcBal=100,000, weights 0.8/0.2, lpSupply=700,000. + // blndPrice = (100k*0.8)/(0.2*3M) = 80,000/600,000 = 2/15 = 0.1333333... + // -> rounds to 1333333 at 7 decimals. + // poolValue = 3M*(2/15) + 100k = 400,000 + 100,000 = 500,000. + // lpPrice = 500,000/700,000 = 5/7 = 0.7142857142... -> rounds to 7142857. + blndPrice, lpPrice, err := cometValuation( + bigFromStroop(3_000_000), bigFromStroop(100_000), + bigFromStroop(0.8), bigFromStroop(0.2), + bigFromStroop(700_000), + ) + require.NoError(t, err) + assert.Equal(t, "1333333", blndPrice) + assert.Equal(t, "7142857", lpPrice) + }) + + t.Run("tiny balances still round correctly", func(t *testing.T) { + // Smallest possible non-zero 7-decimal amounts (raw i128 == 3, 7, 5). + blndBal := big.NewInt(3) + usdcBal := big.NewInt(7) + blndWeight := bigFromStroop(0.8) + usdcWeight := bigFromStroop(0.2) + lpSupply := big.NewInt(5) + + // blndPrice = (7/0.2)/(3/0.8) = (7*0.8)/(0.2*3) = 5.6/0.6 = 28/3 = 9.3333... -> 93333333. + // poolValue = 3*(28/3) + 7 = 28+7 = 35 (raw). lpPrice = 35/5 = 7.0 -> 70000000. + blndPrice, lpPrice, err := cometValuation(blndBal, usdcBal, blndWeight, usdcWeight, lpSupply) + require.NoError(t, err) + assert.Equal(t, "93333333", blndPrice) + assert.Equal(t, "70000000", lpPrice) + }) + + t.Run("rejects nil and non-positive inputs without dividing by zero", func(t *testing.T) { + one := bigFromStroop(1) + zero := big.NewInt(0) + neg := big.NewInt(-1) + + cases := map[string]struct { + blndBal, usdcBal, blndWeight, usdcWeight, lpSupply *big.Int + }{ + "nil blndBal": {nil, one, one, one, one}, + "nil usdcBal": {one, nil, one, one, one}, + "nil blndWeight": {one, one, nil, one, one}, + "nil usdcWeight": {one, one, one, nil, one}, + "nil lpSupply": {one, one, one, one, nil}, + "zero blndBal": {zero, one, one, one, one}, + "zero usdcBal": {one, zero, one, one, one}, + "zero blndWeight": {one, one, zero, one, one}, + "zero usdcWeight": {one, one, one, zero, one}, + "zero lpSupply": {one, one, one, one, zero}, + "negative blndBal": {neg, one, one, one, one}, + "negative lpSupply": {one, one, one, one, neg}, + } + + for name, c := range cases { + t.Run(name, func(t *testing.T) { + _, _, err := cometValuation(c.blndBal, c.usdcBal, c.blndWeight, c.usdcWeight, c.lpSupply) + assert.Error(t, err) + }) + } + }) +} + +// TestFetchCometState mocks services.ContractMetadataService — no real RPC. +func TestFetchCometState(t *testing.T) { + ctx := context.Background() + // cometID (mainnet) / cometIDTestnet exercise fetchCometState against two + // distinct pool addresses — the price snapshot task (a later change) will + // call it once per network, so the tests deliberately don't hardcode a + // single address throughout. + const ( + cometID = "CAS3FL6TLZKDGGSISDBWGGPXT3NRR4DYTZD7YOD3HMYO6LTJUVGRVEAM" + cometIDTestnet = "CA5UTUUPHYL5K22UBRUVC37EARZUGYOSGK3IKIXG2JLCC5ZZLI4BDWDM" + ) + + // setupHappyMock wires up get_tokens/get_balance/get_normalized_weight/ + // get_total_supply against the pool at poolID so the token at index + // blndIdx is BLND (weight 0.8) and the other is USDC (weight 0.2), + // proving fetchCometState identifies BLND by weight rather than by + // get_tokens() position. + setupHappyMock := func(t *testing.T, poolID string, blndIdx int) *services.ContractMetadataServiceMock { + t.Helper() + tokenA := randomContractAddr(t) + tokenB := randomContractAddr(t) + tokens := []string{tokenA, tokenB} + blndAddr, usdcAddr := tokens[blndIdx], tokens[1-blndIdx] + + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, poolID, "get_tokens", []xdr.ScVal(nil)). + Return(vecScVal(contractAddrScVal(t, tokenA), contractAddrScVal(t, tokenB)), nil).Once() + + blndArg, err := contractAddressScVal(blndAddr) + require.NoError(t, err) + usdcArg, err := contractAddressScVal(usdcAddr) + require.NoError(t, err) + + m.On("FetchSingleField", mock.Anything, poolID, "get_balance", []xdr.ScVal{blndArg}). + Return(i128ScVal(40_000_000_000_000), nil).Once() // 4,000,000.0000000 + m.On("FetchSingleField", mock.Anything, poolID, "get_balance", []xdr.ScVal{usdcArg}). + Return(i128ScVal(2_000_000_000_000), nil).Once() // 200,000.0000000 + m.On("FetchSingleField", mock.Anything, poolID, "get_normalized_weight", []xdr.ScVal{blndArg}). + Return(i128ScVal(8_000_000), nil).Once() // 0.8 + m.On("FetchSingleField", mock.Anything, poolID, "get_normalized_weight", []xdr.ScVal{usdcArg}). + Return(i128ScVal(2_000_000), nil).Once() // 0.2 + m.On("FetchSingleField", mock.Anything, poolID, "get_total_supply", []xdr.ScVal(nil)). + Return(i128ScVal(10_000_000_000_000), nil).Once() // 1,000,000.0000000 + + return m + } + + t.Run("identifies BLND leg by weight when it is get_tokens()[0]", func(t *testing.T) { + m := setupHappyMock(t, cometID, 0) + + state, err := fetchCometState(ctx, m, cometID) + require.NoError(t, err) + require.NotNil(t, state) + assert.Equal(t, big.NewInt(40_000_000_000_000), state.BLNDBalance) + assert.Equal(t, big.NewInt(2_000_000_000_000), state.USDCBalance) + assert.Equal(t, big.NewInt(8_000_000), state.BLNDWeight) + assert.Equal(t, big.NewInt(2_000_000), state.USDCWeight) + assert.Equal(t, big.NewInt(10_000_000_000_000), state.LPSupply) + }) + + t.Run("identifies BLND leg by weight when it is get_tokens()[1], against a different pool address", func(t *testing.T) { + m := setupHappyMock(t, cometIDTestnet, 1) + + state, err := fetchCometState(ctx, m, cometIDTestnet) + require.NoError(t, err) + require.NotNil(t, state) + assert.Equal(t, big.NewInt(40_000_000_000_000), state.BLNDBalance) + assert.Equal(t, big.NewInt(2_000_000_000_000), state.USDCBalance) + }) + + t.Run("nil metadata service degrades with error", func(t *testing.T) { + _, err := fetchCometState(ctx, nil, cometID) + assert.Error(t, err) + }) + + t.Run("get_tokens RPC error propagates", func(t *testing.T) { + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(xdr.ScVal{}, assert.AnError).Once() + + _, err := fetchCometState(ctx, m, cometID) + assert.Error(t, err) + }) + + t.Run("get_tokens wrong shape errors", func(t *testing.T) { + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(u32ScVal(1), nil).Once() + + _, err := fetchCometState(ctx, m, cometID) + assert.Error(t, err) + }) + + t.Run("get_tokens wrong count errors", func(t *testing.T) { + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(vecScVal(contractAddrScVal(t, randomContractAddr(t))), nil).Once() + + _, err := fetchCometState(ctx, m, cometID) + assert.Error(t, err) + }) + + t.Run("get_balance RPC error propagates", func(t *testing.T) { + tokenA := randomContractAddr(t) + tokenB := randomContractAddr(t) + + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(vecScVal(contractAddrScVal(t, tokenA), contractAddrScVal(t, tokenB)), nil).Once() + m.On("FetchSingleField", mock.Anything, cometID, "get_balance", mock.Anything). + Return(xdr.ScVal{}, assert.AnError).Once() + + _, err := fetchCometState(ctx, m, cometID) + assert.Error(t, err) + }) + + t.Run("get_normalized_weight wrong type errors", func(t *testing.T) { + tokenA := randomContractAddr(t) + tokenB := randomContractAddr(t) + + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(vecScVal(contractAddrScVal(t, tokenA), contractAddrScVal(t, tokenB)), nil).Once() + m.On("FetchSingleField", mock.Anything, cometID, "get_balance", mock.Anything). + Return(i128ScVal(1), nil) + m.On("FetchSingleField", mock.Anything, cometID, "get_normalized_weight", mock.Anything). + Return(u32ScVal(1), nil).Once() + + _, err := fetchCometState(ctx, m, cometID) + assert.Error(t, err) + }) + + t.Run("get_total_supply RPC error propagates", func(t *testing.T) { + tokenA := randomContractAddr(t) + tokenB := randomContractAddr(t) + + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(vecScVal(contractAddrScVal(t, tokenA), contractAddrScVal(t, tokenB)), nil).Once() + m.On("FetchSingleField", mock.Anything, cometID, "get_balance", mock.Anything). + Return(i128ScVal(1), nil) + m.On("FetchSingleField", mock.Anything, cometID, "get_normalized_weight", mock.Anything). + Return(i128ScVal(8_000_000), nil).Once(). + On("FetchSingleField", mock.Anything, cometID, "get_normalized_weight", mock.Anything). + Return(i128ScVal(2_000_000), nil).Once() + m.On("FetchSingleField", mock.Anything, cometID, "get_total_supply", []xdr.ScVal(nil)). + Return(xdr.ScVal{}, assert.AnError).Once() + + _, err := fetchCometState(ctx, m, cometID) + assert.Error(t, err) + }) + + t.Run("equal weights are ambiguous and error", func(t *testing.T) { + tokenA := randomContractAddr(t) + tokenB := randomContractAddr(t) + + m := services.NewContractMetadataServiceMock(t) + m.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(vecScVal(contractAddrScVal(t, tokenA), contractAddrScVal(t, tokenB)), nil).Once() + m.On("FetchSingleField", mock.Anything, cometID, "get_balance", mock.Anything). + Return(i128ScVal(1), nil) + m.On("FetchSingleField", mock.Anything, cometID, "get_normalized_weight", mock.Anything). + Return(i128ScVal(5_000_000), nil) + m.On("FetchSingleField", mock.Anything, cometID, "get_total_supply", []xdr.ScVal(nil)). + Return(i128ScVal(1), nil) + + _, err := fetchCometState(ctx, m, cometID) + assert.Error(t, err) + }) +} diff --git a/internal/services/blend/prices.go b/internal/services/blend/prices.go new file mode 100644 index 000000000..c5c2ec8e5 --- /dev/null +++ b/internal/services/blend/prices.go @@ -0,0 +1,301 @@ +// Package blend — prices.go implements the periodic SEP-40 oracle price +// snapshot task: on a fixed interval, it discovers every (oracle, asset) pair +// some tracked pool's reserve prices against (blenddata.OraclePriceModel. +// GetPriceTargets), fetches each oracle's decimals() once and every target's +// lastprice(asset) (scval.go's buildSep40StellarAsset/decodePriceData), and +// optionally derives the Blend v2 backstop's Comet BLND:USDC LP deposit +// token's BLND and LP-share USD prices from the pool's own on-chain state +// (comet.go) — no on-chain oracle prices a Comet LP token directly. Every row +// from one pass is written in a single blenddata.OraclePriceModel.BatchUpsert +// call. +// +// Comet leg row convention (locked cross-PR contract — PR5 reads these rows): +// both rows are stored under oracle_contract_id = the Comet pool's own +// C-address (a schema convenience key, not a claim the pool implements +// SEP-40). The LP-share row is self-priced: asset_contract_id == +// oracle_contract_id. The BLND row's asset_contract_id is the real BLND +// token address (cometState.BLNDAddress). Both are recorded at Comet's +// 7-decimal STROOP scale (cometPriceDecimals) with PriceTimestamp set to the +// snapshot's own wall-clock time (unix seconds), since the Comet getters +// carry no oracle-reported timestamp of their own. +// +// A price the pool contract itself would reject is never persisted: an +// oracle-reported timestamp older than maxPriceAge is skipped as "stale" and +// a price ≤ 0 as "invalid" (both observable via the fetch-outcome counter, +// staleness additionally via the oldest-price-age gauge). +// +// A failure fetching one oracle's decimals or any of its lastprice targets +// is isolated to that oracle: it never aborts the rest of the pass. All such +// failures (and a Comet leg failure, when configured) are joined with +// errors.Join and returned to the caller for logging; SnapshotOnce still +// upserts whatever rows the pass did manage to collect. +package blend + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/stellar/go-stellar-sdk/support/log" + + blenddata "github.com/stellar/wallet-backend/internal/data/blend" + "github.com/stellar/wallet-backend/internal/indexer/types" + "github.com/stellar/wallet-backend/internal/metrics" + "github.com/stellar/wallet-backend/internal/services" +) + +// maxPriceAge is how old an oracle-reported price may be and still be +// persisted. It mirrors the Blend v2 pool contract's own rejection window: +// pool.rs::load_price panics on prices older than 24h (and on prices ≤ 0), +// so a price past this age could never be used by the pool it prices for. +const maxPriceAge = 24 * time.Hour + +// PriceSnapshotConfig configures a PriceSnapshotService. +type PriceSnapshotConfig struct { + // OraclePrices is the blend_oracle_prices data access layer: GetPriceTargets + // discovers what to fetch, BatchUpsert persists what was fetched. + OraclePrices *blenddata.OraclePriceModel + // Metadata performs the RPC simulation calls (decimals, lastprice, and the + // Comet getters) against contracts. + Metadata services.ContractMetadataService + // Interval is the wait between the end of one snapshot pass and the start + // of the next. + Interval time.Duration + // BackstopLPContractID is the Comet BLND:USDC weighted pool backing the + // Blend v2 backstop's deposit token, as a strkey C-address. Empty disables + // the BLND/LP-share derived-pricing leg entirely. + BackstopLPContractID string + // Metrics receives per-pass duration, per-fetch outcome, and tracked-row- + // count observations. + Metrics *metrics.BlendPriceMetrics +} + +// PriceSnapshotService periodically snapshots SEP-40 oracle prices — and, +// when configured, the Comet backstop LP/BLND leg — into blend_oracle_prices. +// See the package doc for the persistence and error-isolation contract. +type PriceSnapshotService struct { + oraclePrices *blenddata.OraclePriceModel + metadata services.ContractMetadataService + interval time.Duration + cometID string + metrics *metrics.BlendPriceMetrics +} + +// NewPriceSnapshotService validates cfg and constructs a PriceSnapshotService. +// BackstopLPContractID is the only optional field. +func NewPriceSnapshotService(cfg PriceSnapshotConfig) (*PriceSnapshotService, error) { + if cfg.OraclePrices == nil { + return nil, fmt.Errorf("blend: PriceSnapshotConfig.OraclePrices is required") + } + if cfg.Metadata == nil { + return nil, fmt.Errorf("blend: PriceSnapshotConfig.Metadata is required") + } + if cfg.Interval <= 0 { + return nil, fmt.Errorf("blend: PriceSnapshotConfig.Interval must be positive, got %s", cfg.Interval) + } + if cfg.Metrics == nil { + return nil, fmt.Errorf("blend: PriceSnapshotConfig.Metrics is required") + } + return &PriceSnapshotService{ + oraclePrices: cfg.OraclePrices, + metadata: cfg.Metadata, + interval: cfg.Interval, + cometID: cfg.BackstopLPContractID, + metrics: cfg.Metrics, + }, nil +} + +// Run snapshots once immediately, then repeatedly until ctx is cancelled, +// waiting s.interval AFTER each pass completes (end-to-start delay). A +// ticker would queue a tick while a slow pass runs — many oracle targets +// plus RPC retries can exceed the interval — and the next pass would then +// start immediately, hammering RPC/DB back-to-back. A failed pass is +// logged, never fatal — the next pass tries again. +func (s *PriceSnapshotService) Run(ctx context.Context) { + timer := time.NewTimer(0) // fires immediately: the first pass runs at startup + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return + case <-timer.C: + } + if err := s.SnapshotOnce(ctx); err != nil { + log.Ctx(ctx).Warnf("blend: price snapshot pass failed: %v", err) + } + timer.Reset(s.interval) + } +} + +// SnapshotOnce runs a single price-snapshot pass: discover targets, fetch +// every oracle's prices (isolating per-oracle failures), optionally derive +// the Comet leg, and upsert everything collected in one BatchUpsert call. +// Returns an errors.Join of every isolated failure (nil if there were none); +// a non-nil error does not mean no rows were written. +func (s *PriceSnapshotService) SnapshotOnce(ctx context.Context) error { + start := time.Now() + defer func() { + s.metrics.SnapshotDuration.Observe(time.Since(start).Seconds()) + }() + + targets, err := s.oraclePrices.GetPriceTargets(ctx) + if err != nil { + return fmt.Errorf("blend: price snapshot: fetching price targets: %w", err) + } + + byOracle := make(map[string][]blenddata.PriceTarget) + for _, target := range targets { + oracle := string(target.OracleContractID) + byOracle[oracle] = append(byOracle[oracle], target) + } + + now := time.Now().Unix() + var rows []blenddata.OraclePrice + var errs []error + var oldestAge int64 + for oracle, oracleTargets := range byOracle { + oracleRows, oracleOldestAge, oracleErr := s.snapshotOracle(ctx, oracle, oracleTargets, now) + if oracleErr != nil { + errs = append(errs, fmt.Errorf("blend: price snapshot: oracle %s: %w", oracle, oracleErr)) + } + if oracleOldestAge > oldestAge { + oldestAge = oracleOldestAge + } + rows = append(rows, oracleRows...) + } + s.metrics.OldestPriceAge.Set(float64(oldestAge)) + + if s.cometID != "" { + cometRows, cometErr := s.snapshotComet(ctx) + if cometErr != nil { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + errs = append(errs, fmt.Errorf("blend: price snapshot: comet leg: %w", cometErr)) + } else { + s.metrics.FetchesTotal.WithLabelValues("success").Add(float64(len(cometRows))) + rows = append(rows, cometRows...) + } + } + + s.metrics.PricesTracked.Set(float64(len(rows))) + + if len(rows) > 0 { + if upsertErr := s.oraclePrices.BatchUpsert(ctx, rows); upsertErr != nil { + errs = append(errs, fmt.Errorf("blend: price snapshot: persisting %d rows: %w", len(rows), upsertErr)) + } + } + + return errors.Join(errs...) +} + +// snapshotOracle fetches decimals() once for oracle, then lastprice(asset) +// for each of targets, skipping (not erroring on) a SEP-40 Option::None +// response, a price the pool contract would reject as stale (older than +// maxPriceAge against now) or invalid (≤ 0). A decimals() failure aborts +// this oracle's whole batch — every target's price is meaningless without +// it — but is isolated to this oracle by the caller. A single target's +// lastprice failure is isolated to that target: the rest of this oracle's +// targets are still attempted. The second return value is the age in +// seconds of the oldest price decoded (including skipped-as-stale ones — +// that growth is the dead-oracle signal the gauge exists for). +func (s *PriceSnapshotService) snapshotOracle(ctx context.Context, oracle string, targets []blenddata.PriceTarget, now int64) ([]blenddata.OraclePrice, int64, error) { + decimalsVal, err := s.metadata.FetchSingleField(ctx, oracle, "decimals") + if err != nil { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + return nil, 0, fmt.Errorf("fetching decimals: %w", err) + } + decimals, ok := u32Val(decimalsVal) + if !ok { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + return nil, 0, fmt.Errorf("decimals: expected u32, got %v", decimalsVal.Type) + } + + rows := make([]blenddata.OraclePrice, 0, len(targets)) + var errs []error + var oldestAge int64 + for _, target := range targets { + assetArg, buildErr := buildSep40StellarAsset(string(target.AssetContractID)) + if buildErr != nil { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + errs = append(errs, fmt.Errorf("asset %s: building lastprice argument: %w", target.AssetContractID, buildErr)) + continue + } + + priceVal, fetchErr := s.metadata.FetchSingleField(ctx, oracle, "lastprice", assetArg) + if fetchErr != nil { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + errs = append(errs, fmt.Errorf("asset %s: fetching lastprice: %w", target.AssetContractID, fetchErr)) + continue + } + + priceData, decodeErr := decodePriceData(priceVal) + if decodeErr != nil { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + errs = append(errs, fmt.Errorf("asset %s: decoding lastprice: %w", target.AssetContractID, decodeErr)) + continue + } + if priceData == nil { + s.metrics.FetchesTotal.WithLabelValues("none").Inc() + continue + } + if priceData.Price.Sign() <= 0 { + s.metrics.FetchesTotal.WithLabelValues("invalid").Inc() + continue + } + if age := now - int64(priceData.Timestamp); age > oldestAge { + oldestAge = age + } + if int64(priceData.Timestamp) < now-int64(maxPriceAge/time.Second) { + s.metrics.FetchesTotal.WithLabelValues("stale").Inc() + continue + } + + s.metrics.FetchesTotal.WithLabelValues("success").Inc() + rows = append(rows, blenddata.OraclePrice{ + OracleContractID: target.OracleContractID, + AssetContractID: target.AssetContractID, + Price: priceData.Price.String(), + PriceDecimals: int32(decimals), + PriceTimestamp: int64(priceData.Timestamp), + }) + } + + return rows, oldestAge, errors.Join(errs...) +} + +// snapshotComet derives the BLND and Comet LP-share USD prices from the +// configured backstop Comet pool's raw on-chain state (fetchCometState, +// cometValuation) and returns their blend_oracle_prices rows. See the +// package doc for the row-key convention. +func (s *PriceSnapshotService) snapshotComet(ctx context.Context) ([]blenddata.OraclePrice, error) { + state, err := fetchCometState(ctx, s.metadata, s.cometID) + if err != nil { + return nil, fmt.Errorf("fetching comet state: %w", err) + } + + blndPrice, lpPrice, err := cometValuation(state.BLNDBalance, state.USDCBalance, state.BLNDWeight, state.USDCWeight, state.LPSupply) + if err != nil { + return nil, fmt.Errorf("valuing comet pool: %w", err) + } + + now := time.Now().Unix() + cometAddr := types.AddressBytea(s.cometID) + return []blenddata.OraclePrice{ + { + OracleContractID: cometAddr, + AssetContractID: types.AddressBytea(state.BLNDAddress), + Price: blndPrice, + PriceDecimals: cometPriceDecimals, + PriceTimestamp: now, + }, + { + // The LP share is self-priced: this row's asset is the Comet pool + // itself, matching the locked cross-PR row-key convention. + OracleContractID: cometAddr, + AssetContractID: cometAddr, + Price: lpPrice, + PriceDecimals: cometPriceDecimals, + PriceTimestamp: now, + }, + }, nil +} diff --git a/internal/services/blend/prices_test.go b/internal/services/blend/prices_test.go new file mode 100644 index 000000000..e58a74424 --- /dev/null +++ b/internal/services/blend/prices_test.go @@ -0,0 +1,457 @@ +// Unit tests for PriceSnapshotService. These exercise the real +// blenddata.OraclePriceModel against a PostgreSQL test database (dbtest, +// mirroring internal/data/blend/oracle_prices_test.go's fixture usage) and a +// mocked services.ContractMetadataService — no real RPC. +package blend + +import ( + "context" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stellar/go-stellar-sdk/xdr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + blenddata "github.com/stellar/wallet-backend/internal/data/blend" + "github.com/stellar/wallet-backend/internal/db" + "github.com/stellar/wallet-backend/internal/db/dbtest" + "github.com/stellar/wallet-backend/internal/indexer/types" + "github.com/stellar/wallet-backend/internal/metrics" + "github.com/stellar/wallet-backend/internal/services" +) + +// newPriceSnapshotFixture opens an isolated test database (migrations +// applied) and wraps it in a real OraclePriceModel, mirroring +// internal/data/blend/oracle_prices_test.go's newOraclePricesFixture. +func newPriceSnapshotFixture(t *testing.T) (context.Context, *pgxpool.Pool, *blenddata.OraclePriceModel, func()) { + t.Helper() + ctx := context.Background() + + dbt := dbtest.Open(t) + pool, err := db.OpenDBConnectionPool(ctx, dbt.DSN) + require.NoError(t, err) + + m := &blenddata.OraclePriceModel{ + DB: pool, + Metrics: metrics.NewMetrics(prometheus.NewRegistry()).DB, + } + + cleanup := func() { + pool.Close() + dbt.Close() + } + return ctx, pool, m, cleanup +} + +// insertBlendPool seeds a blend_pools row wiring poolAddr to oracleAddr, the +// minimum blend_pools shape GetPriceTargets' join needs. +func insertBlendPool(t *testing.T, ctx context.Context, pool *pgxpool.Pool, poolAddr, oracleAddr string) { + t.Helper() + _, err := pool.Exec(ctx, ` + INSERT INTO blend_pools (pool_contract_id, oracle_contract_id, last_modified_ledger) + VALUES ($1, $2, 1) + `, types.AddressBytea(poolAddr), types.AddressBytea(oracleAddr)) + require.NoError(t, err) +} + +// insertBlendReserve seeds a blend_reserves row for (poolAddr, assetAddr). +// Only pool_contract_id/reserve_index/asset_contract_id matter to +// GetPriceTargets; other NOT NULL columns get harmless defaults. +func insertBlendReserve(t *testing.T, ctx context.Context, pool *pgxpool.Pool, poolAddr, assetAddr string, reserveIndex int32) { + t.Helper() + _, err := pool.Exec(ctx, ` + INSERT INTO blend_reserves ( + pool_contract_id, reserve_index, asset_contract_id, + b_rate, d_rate, b_supply, d_supply, ir_mod, backstop_credit, + last_time, decimals, c_factor, l_factor, util, max_util, + r_base, r_one, r_two, r_three, reactivity, supply_cap, enabled, + last_modified_ledger + ) VALUES ( + $1, $2, $3, + '1', '1', '0', '0', '0', '0', + 0, 7, 0, 0, 0, 0, + 0, 0, 0, 0, 0, '0', true, + 0 + ) + `, types.AddressBytea(poolAddr), reserveIndex, types.AddressBytea(assetAddr)) + require.NoError(t, err) +} + +// getOraclePriceRow reads back one blend_oracle_prices row, if present. +func getOraclePriceRow(t *testing.T, ctx context.Context, pool *pgxpool.Pool, oracleAddr, assetAddr string) (blenddata.OraclePrice, bool) { + t.Helper() + var row blenddata.OraclePrice + err := pool.QueryRow(ctx, ` + SELECT price, price_decimals, price_timestamp + FROM blend_oracle_prices WHERE oracle_contract_id = $1 AND asset_contract_id = $2 + `, types.AddressBytea(oracleAddr), types.AddressBytea(assetAddr)).Scan( + &row.Price, &row.PriceDecimals, &row.PriceTimestamp, + ) + if err != nil { + return blenddata.OraclePrice{}, false + } + return row, true +} + +func countOraclePriceRows(t *testing.T, ctx context.Context, pool *pgxpool.Pool) int { + t.Helper() + var n int + require.NoError(t, pool.QueryRow(ctx, `SELECT COUNT(*) FROM blend_oracle_prices`).Scan(&n)) + return n +} + +// priceDataScVal builds the Some(PriceData) ScVal a SEP-40 lastprice(asset) +// call returns: an ScMap with field-name Symbol keys "price"/"timestamp", +// mirroring TestDecodePriceData's fixtures in scval_test.go. +func priceDataScVal(price int64, timestamp uint64) xdr.ScVal { + v := xdr.ScVal{Type: xdr.ScValTypeScvMap} + m := mapScVal( + xdr.ScMapEntry{Key: symScVal("price"), Val: i128ScVal(price)}, + xdr.ScMapEntry{Key: symScVal("timestamp"), Val: u64ScVal(timestamp)}, + ) + v.Map = &m + return v +} + +func newTestSnapshotService(t *testing.T, oraclePrices *blenddata.OraclePriceModel, meta services.ContractMetadataService, cometID string, bpm *metrics.BlendPriceMetrics) *PriceSnapshotService { + t.Helper() + svc, err := NewPriceSnapshotService(PriceSnapshotConfig{ + OraclePrices: oraclePrices, + Metadata: meta, + Interval: time.Minute, + BackstopLPContractID: cometID, + Metrics: bpm, + }) + require.NoError(t, err) + return svc +} + +func TestNewPriceSnapshotService_Validation(t *testing.T) { + _, _, oraclePrices, cleanup := newPriceSnapshotFixture(t) + defer cleanup() + meta := services.NewContractMetadataServiceMock(t) + bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices + + t.Run("requires OraclePrices", func(t *testing.T) { + _, err := NewPriceSnapshotService(PriceSnapshotConfig{Metadata: meta, Interval: time.Minute, Metrics: bpm}) + assert.Error(t, err) + }) + t.Run("requires Metadata", func(t *testing.T) { + _, err := NewPriceSnapshotService(PriceSnapshotConfig{OraclePrices: oraclePrices, Interval: time.Minute, Metrics: bpm}) + assert.Error(t, err) + }) + t.Run("requires a positive Interval", func(t *testing.T) { + _, err := NewPriceSnapshotService(PriceSnapshotConfig{OraclePrices: oraclePrices, Metadata: meta, Metrics: bpm}) + assert.Error(t, err) + }) + t.Run("requires Metrics", func(t *testing.T) { + _, err := NewPriceSnapshotService(PriceSnapshotConfig{OraclePrices: oraclePrices, Metadata: meta, Interval: time.Minute}) + assert.Error(t, err) + }) + t.Run("BackstopLPContractID is optional", func(t *testing.T) { + svc, err := NewPriceSnapshotService(PriceSnapshotConfig{OraclePrices: oraclePrices, Metadata: meta, Interval: time.Minute, Metrics: bpm}) + require.NoError(t, err) + require.NotNil(t, svc) + }) +} + +// TestSnapshotOnce_ReserveAssets covers the common path: two oracles, one +// backing two pool reserves' assets and the other backing one, each fetched +// and persisted correctly. +func TestSnapshotOnce_ReserveAssets(t *testing.T) { + ctx, pool, oraclePrices, cleanup := newPriceSnapshotFixture(t) + defer cleanup() + + oracle1 := randomContractAddr(t) + oracle2 := randomContractAddr(t) + poolA := randomContractAddr(t) + poolB := randomContractAddr(t) + assetX := randomContractAddr(t) + assetY := randomContractAddr(t) + assetZ := randomContractAddr(t) + + insertBlendPool(t, ctx, pool, poolA, oracle1) + insertBlendReserve(t, ctx, pool, poolA, assetX, 0) + insertBlendReserve(t, ctx, pool, poolA, assetY, 1) + insertBlendPool(t, ctx, pool, poolB, oracle2) + insertBlendReserve(t, ctx, pool, poolB, assetZ, 0) + + assetXArg, err := buildSep40StellarAsset(assetX) + require.NoError(t, err) + assetYArg, err := buildSep40StellarAsset(assetY) + require.NoError(t, err) + assetZArg, err := buildSep40StellarAsset(assetZ) + require.NoError(t, err) + + // Oracle-reported timestamps must be recent: SnapshotOnce skips prices + // older than maxPriceAge against the wall clock. + tsX := uint64(time.Now().Unix() - 60) + tsY := tsX + 1 + tsZ := tsX + 2 + + meta := services.NewContractMetadataServiceMock(t) + meta.On("FetchSingleField", mock.Anything, oracle1, "decimals", []xdr.ScVal(nil)). + Return(u32ScVal(7), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle1, "lastprice", []xdr.ScVal{assetXArg}). + Return(priceDataScVal(100_000_000, tsX), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle1, "lastprice", []xdr.ScVal{assetYArg}). + Return(priceDataScVal(200_000_000, tsY), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle2, "decimals", []xdr.ScVal(nil)). + Return(u32ScVal(14), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle2, "lastprice", []xdr.ScVal{assetZArg}). + Return(priceDataScVal(300_000_000, tsZ), nil).Once() + + bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices + svc := newTestSnapshotService(t, oraclePrices, meta, "", bpm) + + require.NoError(t, svc.SnapshotOnce(ctx)) + + rowX, ok := getOraclePriceRow(t, ctx, pool, oracle1, assetX) + require.True(t, ok) + assert.Equal(t, "100000000", rowX.Price) + assert.Equal(t, int32(7), rowX.PriceDecimals) + assert.Equal(t, int64(tsX), rowX.PriceTimestamp) + + rowY, ok := getOraclePriceRow(t, ctx, pool, oracle1, assetY) + require.True(t, ok) + assert.Equal(t, "200000000", rowY.Price) + assert.Equal(t, int32(7), rowY.PriceDecimals) + + rowZ, ok := getOraclePriceRow(t, ctx, pool, oracle2, assetZ) + require.True(t, ok) + assert.Equal(t, "300000000", rowZ.Price) + assert.Equal(t, int32(14), rowZ.PriceDecimals) + + assert.Equal(t, 3, countOraclePriceRows(t, ctx, pool)) + assert.Equal(t, 3.0, testutil.ToFloat64(bpm.FetchesTotal.WithLabelValues("success"))) + assert.Equal(t, 3.0, testutil.ToFloat64(bpm.PricesTracked)) + assert.InDelta(t, 60, testutil.ToFloat64(bpm.OldestPriceAge), 5, + "oldest-price-age gauge should reflect the oldest oracle-reported timestamp") +} + +// TestSnapshotOnce_NonePriceSkipped covers a SEP-40 Option::None response +// (ScvVoid): no row is written, no error is raised, and the "none" fetch +// outcome is counted. +func TestSnapshotOnce_NonePriceSkipped(t *testing.T) { + ctx, pool, oraclePrices, cleanup := newPriceSnapshotFixture(t) + defer cleanup() + + oracle := randomContractAddr(t) + poolA := randomContractAddr(t) + asset := randomContractAddr(t) + insertBlendPool(t, ctx, pool, poolA, oracle) + insertBlendReserve(t, ctx, pool, poolA, asset, 0) + + assetArg, err := buildSep40StellarAsset(asset) + require.NoError(t, err) + + meta := services.NewContractMetadataServiceMock(t) + meta.On("FetchSingleField", mock.Anything, oracle, "decimals", []xdr.ScVal(nil)). + Return(u32ScVal(7), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle, "lastprice", []xdr.ScVal{assetArg}). + Return(voidScVal(), nil).Once() + + bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices + svc := newTestSnapshotService(t, oraclePrices, meta, "", bpm) + + require.NoError(t, svc.SnapshotOnce(ctx)) + + _, ok := getOraclePriceRow(t, ctx, pool, oracle, asset) + assert.False(t, ok, "no row should be written for a None price") + assert.Equal(t, 0, countOraclePriceRows(t, ctx, pool)) + assert.Equal(t, 1.0, testutil.ToFloat64(bpm.FetchesTotal.WithLabelValues("none"))) + assert.Equal(t, 0.0, testutil.ToFloat64(bpm.PricesTracked)) +} + +// TestSnapshotOnce_StalePriceSkipped covers the staleness guard: a price +// whose oracle-reported timestamp is older than maxPriceAge is not persisted +// (the pool contract would reject it anyway), counted as a "stale" fetch +// outcome, and still drives the oldest-price-age gauge so a dead oracle is +// observable. +func TestSnapshotOnce_StalePriceSkipped(t *testing.T) { + ctx, pool, oraclePrices, cleanup := newPriceSnapshotFixture(t) + defer cleanup() + + oracle := randomContractAddr(t) + poolA := randomContractAddr(t) + assetFresh := randomContractAddr(t) + assetStale := randomContractAddr(t) + insertBlendPool(t, ctx, pool, poolA, oracle) + insertBlendReserve(t, ctx, pool, poolA, assetFresh, 0) + insertBlendReserve(t, ctx, pool, poolA, assetStale, 1) + + assetFreshArg, err := buildSep40StellarAsset(assetFresh) + require.NoError(t, err) + assetStaleArg, err := buildSep40StellarAsset(assetStale) + require.NoError(t, err) + + staleAge := int64(25 * 60 * 60) // 25h, just past the 24h maxPriceAge + meta := services.NewContractMetadataServiceMock(t) + meta.On("FetchSingleField", mock.Anything, oracle, "decimals", []xdr.ScVal(nil)). + Return(u32ScVal(7), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle, "lastprice", []xdr.ScVal{assetFreshArg}). + Return(priceDataScVal(100_000_000, uint64(time.Now().Unix())), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle, "lastprice", []xdr.ScVal{assetStaleArg}). + Return(priceDataScVal(200_000_000, uint64(time.Now().Unix()-staleAge)), nil).Once() + + bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices + svc := newTestSnapshotService(t, oraclePrices, meta, "", bpm) + + require.NoError(t, svc.SnapshotOnce(ctx)) + + _, ok := getOraclePriceRow(t, ctx, pool, oracle, assetFresh) + assert.True(t, ok, "the fresh price must persist") + _, ok = getOraclePriceRow(t, ctx, pool, oracle, assetStale) + assert.False(t, ok, "the stale price must not persist") + assert.Equal(t, 1, countOraclePriceRows(t, ctx, pool)) + + assert.Equal(t, 1.0, testutil.ToFloat64(bpm.FetchesTotal.WithLabelValues("success"))) + assert.Equal(t, 1.0, testutil.ToFloat64(bpm.FetchesTotal.WithLabelValues("stale"))) + assert.InDelta(t, staleAge, testutil.ToFloat64(bpm.OldestPriceAge), 5, + "the gauge must include skipped-as-stale prices — that's the dead-oracle signal") +} + +// TestSnapshotOnce_NonPositivePriceSkipped covers the validity guard: a zero +// or negative price is never persisted (the pool contract rejects it) and is +// counted as an "invalid" fetch outcome, without erroring the pass. +func TestSnapshotOnce_NonPositivePriceSkipped(t *testing.T) { + ctx, pool, oraclePrices, cleanup := newPriceSnapshotFixture(t) + defer cleanup() + + oracle := randomContractAddr(t) + poolA := randomContractAddr(t) + assetZero := randomContractAddr(t) + assetNegative := randomContractAddr(t) + insertBlendPool(t, ctx, pool, poolA, oracle) + insertBlendReserve(t, ctx, pool, poolA, assetZero, 0) + insertBlendReserve(t, ctx, pool, poolA, assetNegative, 1) + + assetZeroArg, err := buildSep40StellarAsset(assetZero) + require.NoError(t, err) + assetNegativeArg, err := buildSep40StellarAsset(assetNegative) + require.NoError(t, err) + + meta := services.NewContractMetadataServiceMock(t) + meta.On("FetchSingleField", mock.Anything, oracle, "decimals", []xdr.ScVal(nil)). + Return(u32ScVal(7), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle, "lastprice", []xdr.ScVal{assetZeroArg}). + Return(priceDataScVal(0, uint64(time.Now().Unix())), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle, "lastprice", []xdr.ScVal{assetNegativeArg}). + Return(priceDataScVal(-5, uint64(time.Now().Unix())), nil).Once() + + bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices + svc := newTestSnapshotService(t, oraclePrices, meta, "", bpm) + + require.NoError(t, svc.SnapshotOnce(ctx)) + + assert.Equal(t, 0, countOraclePriceRows(t, ctx, pool)) + assert.Equal(t, 2.0, testutil.ToFloat64(bpm.FetchesTotal.WithLabelValues("invalid"))) + assert.Equal(t, 0.0, testutil.ToFloat64(bpm.PricesTracked)) +} + +// TestSnapshotOnce_OracleErrorIsolated covers one oracle's failure not +// preventing another oracle's rows from being persisted. +func TestSnapshotOnce_OracleErrorIsolated(t *testing.T) { + ctx, pool, oraclePrices, cleanup := newPriceSnapshotFixture(t) + defer cleanup() + + oracleBad := randomContractAddr(t) + oracleGood := randomContractAddr(t) + poolBad := randomContractAddr(t) + poolGood := randomContractAddr(t) + assetBad := randomContractAddr(t) + assetGood := randomContractAddr(t) + + insertBlendPool(t, ctx, pool, poolBad, oracleBad) + insertBlendReserve(t, ctx, pool, poolBad, assetBad, 0) + insertBlendPool(t, ctx, pool, poolGood, oracleGood) + insertBlendReserve(t, ctx, pool, poolGood, assetGood, 0) + + assetGoodArg, err := buildSep40StellarAsset(assetGood) + require.NoError(t, err) + + meta := services.NewContractMetadataServiceMock(t) + meta.On("FetchSingleField", mock.Anything, oracleBad, "decimals", []xdr.ScVal(nil)). + Return(xdr.ScVal{}, assert.AnError).Once() + meta.On("FetchSingleField", mock.Anything, oracleGood, "decimals", []xdr.ScVal(nil)). + Return(u32ScVal(7), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracleGood, "lastprice", []xdr.ScVal{assetGoodArg}). + Return(priceDataScVal(500_000_000, uint64(time.Now().Unix())), nil).Once() + + bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices + svc := newTestSnapshotService(t, oraclePrices, meta, "", bpm) + + err = svc.SnapshotOnce(ctx) + require.Error(t, err) + assert.ErrorContains(t, err, oracleBad) + + _, ok := getOraclePriceRow(t, ctx, pool, oracleBad, assetBad) + assert.False(t, ok, "the failing oracle's target must not have a row") + + rowGood, ok := getOraclePriceRow(t, ctx, pool, oracleGood, assetGood) + require.True(t, ok, "the healthy oracle's row must still persist") + assert.Equal(t, "500000000", rowGood.Price) + + assert.Equal(t, 1, countOraclePriceRows(t, ctx, pool)) +} + +// TestSnapshotOnce_CometLeg covers the optional BLND/LP-share derived-pricing +// leg: when BackstopLPContractID is configured, fetchCometState's getters are +// called and two rows are written under oracle_contract_id = the Comet pool +// address — the BLND row keyed by the real BLND token address, the LP-share +// row self-priced (asset_contract_id == oracle_contract_id). +func TestSnapshotOnce_CometLeg(t *testing.T) { + ctx, pool, oraclePrices, cleanup := newPriceSnapshotFixture(t) + defer cleanup() + + cometID := randomContractAddr(t) + blndAddr := randomContractAddr(t) + usdcAddr := randomContractAddr(t) + + blndArg, err := contractAddressScVal(blndAddr) + require.NoError(t, err) + usdcArg, err := contractAddressScVal(usdcAddr) + require.NoError(t, err) + + meta := services.NewContractMetadataServiceMock(t) + meta.On("FetchSingleField", mock.Anything, cometID, "get_tokens", []xdr.ScVal(nil)). + Return(vecScVal(contractAddrScVal(t, blndAddr), contractAddrScVal(t, usdcAddr)), nil).Once() + meta.On("FetchSingleField", mock.Anything, cometID, "get_balance", []xdr.ScVal{blndArg}). + Return(i128ScVal(40_000_000_000_000), nil).Once() // 4,000,000.0000000 + meta.On("FetchSingleField", mock.Anything, cometID, "get_balance", []xdr.ScVal{usdcArg}). + Return(i128ScVal(2_000_000_000_000), nil).Once() // 200,000.0000000 + meta.On("FetchSingleField", mock.Anything, cometID, "get_normalized_weight", []xdr.ScVal{blndArg}). + Return(i128ScVal(8_000_000), nil).Once() // 0.8 + meta.On("FetchSingleField", mock.Anything, cometID, "get_normalized_weight", []xdr.ScVal{usdcArg}). + Return(i128ScVal(2_000_000), nil).Once() // 0.2 + meta.On("FetchSingleField", mock.Anything, cometID, "get_total_supply", []xdr.ScVal(nil)). + Return(i128ScVal(10_000_000_000_000), nil).Once() // 1,000,000.0000000 + + bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices + svc := newTestSnapshotService(t, oraclePrices, meta, cometID, bpm) + + require.NoError(t, svc.SnapshotOnce(ctx)) + + blndRow, ok := getOraclePriceRow(t, ctx, pool, cometID, blndAddr) + require.True(t, ok, "expected a BLND leg row keyed by the BLND token address") + assert.Equal(t, "2000000", blndRow.Price) + assert.Equal(t, int32(7), blndRow.PriceDecimals) + + lpRow, ok := getOraclePriceRow(t, ctx, pool, cometID, cometID) + require.True(t, ok, "expected a self-priced LP-share row") + assert.Equal(t, "10000000", lpRow.Price) + assert.Equal(t, int32(7), lpRow.PriceDecimals) + + assert.Equal(t, 2, countOraclePriceRows(t, ctx, pool)) + assert.Equal(t, 2.0, testutil.ToFloat64(bpm.FetchesTotal.WithLabelValues("success"))) + assert.Equal(t, 2.0, testutil.ToFloat64(bpm.PricesTracked)) + + now := time.Now().Unix() + assert.InDelta(t, now, blndRow.PriceTimestamp, 5, "BLND row timestamp should be ~now, not an oracle-reported one") + assert.InDelta(t, now, lpRow.PriceTimestamp, 5, "LP row timestamp should be ~now, not an oracle-reported one") +} diff --git a/internal/services/blend/scval.go b/internal/services/blend/scval.go index 2d6bb916f..325cb11a3 100644 --- a/internal/services/blend/scval.go +++ b/internal/services/blend/scval.go @@ -12,8 +12,10 @@ package blend import ( + "fmt" "math/big" + "github.com/stellar/go-stellar-sdk/strkey" "github.com/stellar/go-stellar-sdk/xdr" ) @@ -42,11 +44,19 @@ func i128String(v xdr.ScVal) (string, bool) { if !ok { return "", false } - // value = Hi * 2^64 + Lo, where Hi is signed and Lo is unsigned. + return i128ToBigInt(parts).String(), true +} + +// i128ToBigInt converts an XDR Int128Parts — Hi the signed high 64 bits, Lo +// the unsigned low 64 bits — into the equivalent signed big.Int: +// value = Hi*2^64 + Lo. Unlike i128String, this returns the big.Int itself +// so a caller (decodePriceData) can do further arithmetic on the value +// rather than just display it. +func i128ToBigInt(parts xdr.Int128Parts) *big.Int { bi := big.NewInt(int64(parts.Hi)) bi.Lsh(bi, 64) bi.Add(bi, new(big.Int).SetUint64(uint64(parts.Lo))) - return bi.String(), true + return bi } // addrString decodes an ScVal holding an Address into its strkey-encoded @@ -161,3 +171,114 @@ func mapAddrI128(v xdr.ScVal) (map[string]string, bool) { } return out, true } + +// contractIDLen is the raw byte length of a strkey-decoded contract address +// (a 32-byte sha256 hash), matching xdr.ContractId's array size. +const contractIDLen = 32 + +// contractAddressScVal builds the bare ScVal encoding of a contract C-address +// as a Soroban `Address` value — the shape any `token: Address` (or similar) +// function parameter expects, e.g. Comet's get_balance(token)/ +// get_normalized_weight(token) (see comet.go). buildSep40StellarAsset wraps +// this same encoding inside a 2-element ScVec for the SEP-40 +// Asset::Stellar(Address) enum payload. +func contractAddressScVal(contractAddress string) (xdr.ScVal, error) { + raw, err := strkey.Decode(strkey.VersionByteContract, contractAddress) + if err != nil { + return xdr.ScVal{}, fmt.Errorf("blend: decoding contract address %q: %w", contractAddress, err) + } + if len(raw) != contractIDLen { + return xdr.ScVal{}, fmt.Errorf("blend: contract address %q decoded to %d bytes, want %d", contractAddress, len(raw), contractIDLen) + } + var cid xdr.ContractId + copy(cid[:], raw) + return xdr.ScVal{ + Type: xdr.ScValTypeScvAddress, + Address: &xdr.ScAddress{ + Type: xdr.ScAddressTypeScAddressTypeContract, + ContractId: &cid, + }, + }, nil +} + +// buildSep40StellarAsset builds the SEP-40 `Asset::Stellar(Address)` +// argument ScVal for a lastprice(asset) call against a SEP-40-compatible +// oracle (e.g. Reflector), given the strkey C-address of the priced Stellar +// Asset Contract. A Soroban SDK #[contracttype] tuple-variant enum +// serializes as a 2-element ScVec — the variant's Symbol name followed by +// its single payload value — the same convention decodeVecKeyEntry already +// relies on for Blend's own persistent storage keys. +// +// Verified against SEP-40 (stellar-protocol ecosystem/sep-0040.md): +// `enum Asset { Stellar(Address), Other(Symbol) }`, +// `fn lastprice(asset: Asset) -> Option`. Live-confirmed by +// dumping `stellar contract info interface` against the deployed Reflector +// external-CEX/DEX oracle (mainnet +// CAFJZQWSED6YAWZU3GWRTOCNPPCGBN32L7QV43XX5LZLFTK6JLN34DLN), which shows +// the identical Rust shapes, and by simulating +// lastprice(Asset::Stellar()) against the Reflector Stellar-DEX +// oracle (mainnet CALI2BYU2JE6WVRUFYTS6MSBNEHGJ35P4AVCZYF3B6QOE3QKOB2PLE6M), +// which round-tripped to a real Some(PriceData{...}) response. See +// TestBuildSep40StellarAsset for the full verification note. +func buildSep40StellarAsset(contractAddress string) (xdr.ScVal, error) { + addrVal, err := contractAddressScVal(contractAddress) + if err != nil { + return xdr.ScVal{}, fmt.Errorf("blend: building SEP-40 Asset::Stellar(%q): %w", contractAddress, err) + } + + sym := xdr.ScSymbol("Stellar") + vec := &xdr.ScVec{ + {Type: xdr.ScValTypeScvSymbol, Sym: &sym}, + addrVal, + } + return xdr.ScVal{Type: xdr.ScValTypeScvVec, Vec: &vec}, nil +} + +// PriceData is the decoded form of a SEP-40 `PriceData` struct: the raw +// on-chain i128 price (scaled by the oracle's own decimals(), which the +// caller must fetch separately — this package does no rescaling) and the +// Unix-seconds timestamp the price was recorded at. +type PriceData struct { + Price *big.Int + Timestamp uint64 +} + +// decodePriceData decodes a SEP-40 lastprice(asset) return value — an +// Option — into a *PriceData. ScvVoid (Rust's None, meaning the +// oracle has no price for the asset) decodes to (nil, nil), the same +// no-error/no-value convention DecodeEntry's optional fields use elsewhere +// in this package. A populated PriceData is a #[contracttype] struct, so it +// serializes as an ScMap keyed by field-name Symbols (see mapGet). Any other +// top-level shape, or a present-but-malformed/missing field, is reported as +// an error rather than silently treated as "no price" — a caller must not +// mistake a decode failure for a legitimate absent price. +func decodePriceData(val xdr.ScVal) (*PriceData, error) { + if val.Type == xdr.ScValTypeScvVoid { + return nil, nil + } + + m, ok := val.GetMap() + if !ok { + return nil, fmt.Errorf("blend: PriceData: expected ScMap or ScvVoid, got %v", val.Type) + } + + priceVal, ok := mapGet(m, "price") + if !ok { + return nil, fmt.Errorf("blend: PriceData: missing %q field", "price") + } + parts, ok := priceVal.GetI128() + if !ok { + return nil, fmt.Errorf("blend: PriceData: %q field is not an i128 (got %v)", "price", priceVal.Type) + } + + tsVal, ok := mapGet(m, "timestamp") + if !ok { + return nil, fmt.Errorf("blend: PriceData: missing %q field", "timestamp") + } + ts, ok := u64Val(tsVal) + if !ok { + return nil, fmt.Errorf("blend: PriceData: %q field is not a u64 (got %v)", "timestamp", tsVal.Type) + } + + return &PriceData{Price: i128ToBigInt(parts), Timestamp: ts}, nil +} diff --git a/internal/services/blend/scval_test.go b/internal/services/blend/scval_test.go index c168d38d8..7625cd6f3 100644 --- a/internal/services/blend/scval_test.go +++ b/internal/services/blend/scval_test.go @@ -1,6 +1,8 @@ package blend import ( + "math" + "math/big" "testing" "github.com/stellar/go-stellar-sdk/strkey" @@ -142,3 +144,178 @@ func TestU32Val(t *testing.T) { assert.False(t, ok) }) } + +// TestBuildSep40StellarAsset verification note ---------------------------- +// +// SEP-40 (stellar-protocol ecosystem/sep-0040.md, fetched 2026-07-08) defines: +// +// #[contracttype] +// enum Asset { Stellar(Address), Other(Symbol) } +// fn lastprice(env: Env, asset: Asset) -> Option; +// #[contracttype] +// pub struct PriceData { price: i128, timestamp: u64 } +// +// The tuple-variant enum's 2-element-ScVec([Symbol, payload]) encoding +// matches this package's existing decodeVecKeyEntry convention for Blend's +// own persistent storage keys. +// +// Live-confirmed on 2026-07-08 against real mainnet contracts (stellar-cli +// 27.0.0, rpc-url https://mainnet.sorobanrpc.com): +// - `stellar contract info interface` against the deployed Reflector +// "external CEXs & DEXs" oracle +// (CAFJZQWSED6YAWZU3GWRTOCNPPCGBN32L7QV43XX5LZLFTK6JLN34DLN) dumped: +// `pub enum Asset { Stellar(soroban_sdk::Address), Other(soroban_sdk::Symbol) }`, +// `pub struct PriceData { pub price: i128, pub timestamp: u64 }`, +// `fn lastprice(env: soroban_sdk::Env, asset: Asset) -> Option;` +// — an exact match for the SEP-40 shapes above. +// - A live simulation of lastprice(Asset::Stellar()) against the +// Reflector "Stellar Mainnet DEX" oracle +// (CALI2BYU2JE6WVRUFYTS6MSBNEHGJ35P4AVCZYF3B6QOE3QKOB2PLE6M) round-tripped +// to a real `Some(PriceData{price, timestamp})` response, proving the +// Asset::Stellar(Address) argument encoding this test asserts is what a +// production oracle actually expects. +func TestBuildSep40StellarAsset(t *testing.T) { + const contractAddr = "CDLZFC3SYJYDZT7K67VZ75HPJVIEUVNIXF47ZG2FB2RMQQVU2HHGCYSC" + + t.Run("encodes Asset::Stellar(Address) as a 2-elem ScVec", func(t *testing.T) { + got, err := buildSep40StellarAsset(contractAddr) + require.NoError(t, err) + + wantVec := &xdr.ScVec{ + symScVal("Stellar"), + contractAddrScVal(t, contractAddr), + } + want := xdr.ScVal{Type: xdr.ScValTypeScvVec, Vec: &wantVec} + + assert.Equal(t, want, got) + }) + + t.Run("returns an error for a malformed address", func(t *testing.T) { + _, err := buildSep40StellarAsset("not-a-strkey-address") + assert.Error(t, err) + }) + + t.Run("returns an error for a well-formed but non-contract strkey address", func(t *testing.T) { + // A valid G... (account) strkey decodes cleanly but fails the + // VersionByteContract check inside strkey.Decode. + _, err := buildSep40StellarAsset("GCYNTH5HDQRNIQ3BSSYPWFO5AHH5ERVZ32C37QRXT6TXK3OJFFOIVXDE") + assert.Error(t, err) + }) +} + +func TestDecodePriceData(t *testing.T) { + t.Run("decodes Some(PriceData)", func(t *testing.T) { + // Field-name Symbols in alphabetical order, matching how a Soroban + // SDK #[contracttype] struct actually serializes. + v := xdr.ScVal{Type: xdr.ScValTypeScvMap} + m := mapScVal( + xdr.ScMapEntry{Key: symScVal("price"), Val: i128ScVal(1_000_000)}, + xdr.ScMapEntry{Key: symScVal("timestamp"), Val: u64ScVal(1_700_000_000)}, + ) + v.Map = &m + + pd, err := decodePriceData(v) + require.NoError(t, err) + require.NotNil(t, pd) + assert.Equal(t, big.NewInt(1_000_000), pd.Price) + assert.Equal(t, uint64(1_700_000_000), pd.Timestamp) + }) + + t.Run("decodes None (ScvVoid) to (nil, nil)", func(t *testing.T) { + pd, err := decodePriceData(voidScVal()) + require.NoError(t, err) + assert.Nil(t, pd) + }) + + t.Run("returns an error for a value that is neither a map nor void", func(t *testing.T) { + _, err := decodePriceData(u32ScVal(1)) + assert.Error(t, err) + }) + + t.Run("returns an error when the price field is missing", func(t *testing.T) { + v := xdr.ScVal{Type: xdr.ScValTypeScvMap} + m := mapScVal( + xdr.ScMapEntry{Key: symScVal("timestamp"), Val: u64ScVal(1_700_000_000)}, + ) + v.Map = &m + + _, err := decodePriceData(v) + assert.Error(t, err) + }) + + t.Run("returns an error when the timestamp field is missing", func(t *testing.T) { + v := xdr.ScVal{Type: xdr.ScValTypeScvMap} + m := mapScVal( + xdr.ScMapEntry{Key: symScVal("price"), Val: i128ScVal(1_000_000)}, + ) + v.Map = &m + + _, err := decodePriceData(v) + assert.Error(t, err) + }) + + t.Run("returns an error when timestamp is not a u64", func(t *testing.T) { + v := xdr.ScVal{Type: xdr.ScValTypeScvMap} + m := mapScVal( + xdr.ScMapEntry{Key: symScVal("price"), Val: i128ScVal(1_000_000)}, + xdr.ScMapEntry{Key: symScVal("timestamp"), Val: i128ScVal(1_700_000_000)}, + ) + v.Map = &m + + _, err := decodePriceData(v) + assert.Error(t, err) + }) + + t.Run("returns an error when price is not an i128", func(t *testing.T) { + v := xdr.ScVal{Type: xdr.ScValTypeScvMap} + m := mapScVal( + xdr.ScMapEntry{Key: symScVal("price"), Val: u32ScVal(1)}, + xdr.ScMapEntry{Key: symScVal("timestamp"), Val: u64ScVal(1_700_000_000)}, + ) + v.Map = &m + + _, err := decodePriceData(v) + assert.Error(t, err) + }) +} + +func TestI128ToBigInt(t *testing.T) { + t.Run("decodes zero", func(t *testing.T) { + got := i128ToBigInt(xdr.Int128Parts{Hi: 0, Lo: 0}) + assert.Equal(t, big.NewInt(0), got) + }) + + t.Run("decodes a small positive value", func(t *testing.T) { + got := i128ToBigInt(xdr.Int128Parts{Hi: 0, Lo: 42}) + assert.Equal(t, big.NewInt(42), got) + }) + + t.Run("decodes a value greater than 2^64 (Hi > 0)", func(t *testing.T) { + got := i128ToBigInt(xdr.Int128Parts{Hi: 1, Lo: 0}) + want, ok := new(big.Int).SetString("18446744073709551616", 10) // 2^64 + require.True(t, ok) + assert.Equal(t, want, got) + }) + + t.Run("decodes -1 (Hi=-1, Lo=MaxUint64 two's-complement encoding)", func(t *testing.T) { + got := i128ToBigInt(xdr.Int128Parts{Hi: -1, Lo: xdr.Uint64(math.MaxUint64)}) + assert.Equal(t, big.NewInt(-1), got) + }) + + t.Run("decodes a negative value with Hi < -1", func(t *testing.T) { + got := i128ToBigInt(xdr.Int128Parts{Hi: -2, Lo: 0}) + want, ok := new(big.Int).SetString("-36893488147419103232", 10) // -2 * 2^64 + require.True(t, ok) + assert.Equal(t, want, got) + }) + + t.Run("round-trips against independently computed big.Int arithmetic", func(t *testing.T) { + parts := xdr.Int128Parts{Hi: 5, Lo: 12345} + got := i128ToBigInt(parts) + + want := new(big.Int).Lsh(big.NewInt(int64(parts.Hi)), 64) + want.Add(want, new(big.Int).SetUint64(uint64(parts.Lo))) + assert.Equal(t, want, got) + }) +} diff --git a/internal/services/ingest.go b/internal/services/ingest.go index 294853949..75a504307 100644 --- a/internal/services/ingest.go +++ b/internal/services/ingest.go @@ -64,6 +64,12 @@ type IngestServiceConfig struct { // === Live Mode Dependencies === TokenIngestionService TokenIngestionService CheckpointService CheckpointService + // PostLockTasks are long-lived background tasks started only after live + // ingestion acquires its advisory lock, so an instance that loses the + // lock never runs them (e.g. the Blend price snapshot task, which would + // otherwise duplicate RPC load and writes). Each runs in its own + // goroutine for the life of the ingestion context. + PostLockTasks []func(context.Context) // === Protocol Processors === ProtocolProcessors []ProtocolProcessor // nil means no protocol state production @@ -121,6 +127,7 @@ type ingestService struct { ledgerBackendFactory LedgerBackendFactory tokenIngestionService TokenIngestionService checkpointService CheckpointService + postLockTasks []func(context.Context) appMetrics *metrics.Metrics networkPassphrase string getLedgersLimit int @@ -174,6 +181,7 @@ func NewIngestService(cfg IngestServiceConfig) (*ingestService, error) { ledgerBackendFactory: cfg.LedgerBackendFactory, tokenIngestionService: cfg.TokenIngestionService, checkpointService: cfg.CheckpointService, + postLockTasks: cfg.PostLockTasks, appMetrics: cfg.Metrics, networkPassphrase: cfg.NetworkPassphrase, getLedgersLimit: cfg.GetLedgersLimit, diff --git a/internal/services/ingest_live.go b/internal/services/ingest_live.go index 6bcc2121a..48a98bd58 100644 --- a/internal/services/ingest_live.go +++ b/internal/services/ingest_live.go @@ -283,6 +283,13 @@ func (m *ingestService) startLiveIngestion(ctx context.Context) error { } }() + // Background tasks gated on the advisory lock: an instance that failed to + // acquire it must never run them, or a crash-looping/standby pod would + // duplicate their RPC load and writes. + for _, task := range m.postLockTasks { + go task(ctx) + } + // Get latest ingested ledger to determine DB state latestIngestedLedger, err := m.models.IngestStore.Get(ctx, data.LatestLedgerCursorName) if err != nil {