From 978aaa25cfa8ef49e6a23c7fec2dd93e1aa8e1d0 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Wed, 8 Jul 2026 15:44:39 -0400 Subject: [PATCH 1/8] feat(blend): oracle price data model with target discovery and batch upsert --- internal/data/blend/models.go | 3 +- internal/data/blend/oracle_prices.go | 151 +++++++++++++++++++ internal/data/blend/oracle_prices_test.go | 173 ++++++++++++++++++++++ 3 files changed, 326 insertions(+), 1 deletion(-) create mode 100644 internal/data/blend/oracle_prices.go create mode 100644 internal/data/blend/oracle_prices_test.go 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)) + }) +} From 98a98ae1f79e9c7e969002a8d303bce61cd74404 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Wed, 8 Jul 2026 15:59:37 -0400 Subject: [PATCH 2/8] feat(blend): SEP-40 scval encode/decode helpers --- internal/services/blend/scval.go | 111 +++++++++++++++- internal/services/blend/scval_test.go | 177 ++++++++++++++++++++++++++ 2 files changed, 286 insertions(+), 2 deletions(-) diff --git a/internal/services/blend/scval.go b/internal/services/blend/scval.go index 2d6bb916f..af116a4a6 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,100 @@ 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 + +// 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) { + raw, err := strkey.Decode(strkey.VersionByteContract, contractAddress) + if err != nil { + return xdr.ScVal{}, fmt.Errorf("blend: decoding SEP-40 asset contract address %q: %w", contractAddress, err) + } + if len(raw) != contractIDLen { + return xdr.ScVal{}, fmt.Errorf("blend: SEP-40 asset contract address %q decoded to %d bytes, want %d", contractAddress, len(raw), contractIDLen) + } + var cid xdr.ContractId + copy(cid[:], raw) + + sym := xdr.ScSymbol("Stellar") + vec := &xdr.ScVec{ + {Type: xdr.ScValTypeScvSymbol, Sym: &sym}, + { + Type: xdr.ScValTypeScvAddress, + Address: &xdr.ScAddress{ + Type: xdr.ScAddressTypeScAddressTypeContract, + ContractId: &cid, + }, + }, + } + 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) + }) +} From c8381d65adc7d35e1ea7cd3d85112ecabc12b1e8 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Wed, 8 Jul 2026 16:16:35 -0400 Subject: [PATCH 3/8] feat(blend): comet LP valuation and BLND spot derivation --- internal/services/blend/comet.go | 244 +++++++++++++++++++++ internal/services/blend/comet_test.go | 301 ++++++++++++++++++++++++++ internal/services/blend/scval.go | 42 ++-- 3 files changed, 573 insertions(+), 14 deletions(-) create mode 100644 internal/services/blend/comet.go create mode 100644 internal/services/blend/comet_test.go diff --git a/internal/services/blend/comet.go b/internal/services/blend/comet.go new file mode 100644 index 000000000..b1c713449 --- /dev/null +++ b/internal/services/blend/comet.go @@ -0,0 +1,244 @@ +// 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. All fields are +// 7-decimal fixed-point big.Ints (Comet's STROOP scale) — the same shape +// cometValuation consumes. +type cometState struct { + 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{ + 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/scval.go b/internal/services/blend/scval.go index af116a4a6..325cb11a3 100644 --- a/internal/services/blend/scval.go +++ b/internal/services/blend/scval.go @@ -176,6 +176,31 @@ func mapAddrI128(v xdr.ScVal) (map[string]string, bool) { // (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 @@ -196,26 +221,15 @@ const contractIDLen = 32 // which round-tripped to a real Some(PriceData{...}) response. See // TestBuildSep40StellarAsset for the full verification note. func buildSep40StellarAsset(contractAddress string) (xdr.ScVal, error) { - raw, err := strkey.Decode(strkey.VersionByteContract, contractAddress) + addrVal, err := contractAddressScVal(contractAddress) if err != nil { - return xdr.ScVal{}, fmt.Errorf("blend: decoding SEP-40 asset contract address %q: %w", contractAddress, err) - } - if len(raw) != contractIDLen { - return xdr.ScVal{}, fmt.Errorf("blend: SEP-40 asset contract address %q decoded to %d bytes, want %d", contractAddress, len(raw), contractIDLen) + return xdr.ScVal{}, fmt.Errorf("blend: building SEP-40 Asset::Stellar(%q): %w", contractAddress, err) } - var cid xdr.ContractId - copy(cid[:], raw) sym := xdr.ScSymbol("Stellar") vec := &xdr.ScVec{ {Type: xdr.ScValTypeScvSymbol, Sym: &sym}, - { - Type: xdr.ScValTypeScvAddress, - Address: &xdr.ScAddress{ - Type: xdr.ScAddressTypeScAddressTypeContract, - ContractId: &cid, - }, - }, + addrVal, } return xdr.ScVal{Type: xdr.ScValTypeScvVec, Vec: &vec}, nil } From ff506a30738d92bcc97c267ba5c0eb483f1d8513 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Wed, 8 Jul 2026 16:29:08 -0400 Subject: [PATCH 4/8] feat(blend): periodic oracle price snapshot service with metrics --- internal/metrics/blend.go | 42 +++ internal/metrics/blend_test.go | 47 ++++ internal/metrics/metrics.go | 34 +-- internal/metrics/metrics_test.go | 1 + internal/services/blend/comet.go | 11 +- internal/services/blend/prices.go | 267 ++++++++++++++++++ internal/services/blend/prices_test.go | 363 +++++++++++++++++++++++++ 7 files changed, 746 insertions(+), 19 deletions(-) create mode 100644 internal/metrics/blend.go create mode 100644 internal/metrics/blend_test.go create mode 100644 internal/services/blend/prices.go create mode 100644 internal/services/blend/prices_test.go diff --git a/internal/metrics/blend.go b/internal/metrics/blend.go new file mode 100644 index 000000000..379aff54a --- /dev/null +++ b/internal/metrics/blend.go @@ -0,0 +1,42 @@ +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, or none (SEP-40 Option::None — no price recorded yet). + // 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 +} + +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).", + }, []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.", + }), + } + reg.MustRegister(m.SnapshotDuration, m.FetchesTotal, m.PricesTracked) + 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 index b1c713449..baf7fb75c 100644 --- a/internal/services/blend/comet.go +++ b/internal/services/blend/comet.go @@ -52,10 +52,14 @@ const cometPriceDecimals = 7 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. All fields are -// 7-decimal fixed-point big.Ints (Comet's STROOP scale) — the same shape -// cometValuation consumes. +// 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 @@ -235,6 +239,7 @@ func fetchCometState(ctx context.Context, metadata services.ContractMetadataServ usdcIdx := 1 - blndIdx return &cometState{ + BLNDAddress: tokens[blndIdx], BLNDBalance: balances[blndIdx], USDCBalance: balances[usdcIdx], BLNDWeight: weights[blndIdx], diff --git a/internal/services/blend/prices.go b/internal/services/blend/prices.go new file mode 100644 index 000000000..928b5ad3d --- /dev/null +++ b/internal/services/blend/prices.go @@ -0,0 +1,267 @@ +// 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 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" +) + +// 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 every s.interval until ctx is +// cancelled. A failed pass is logged, never fatal — the next tick tries +// again. Mirrors protocol_migrate.go's refreshTargetTip ticker-loop shape. +func (s *PriceSnapshotService) Run(ctx context.Context) { + ticker := time.NewTicker(s.interval) + defer ticker.Stop() + for { + if ctx.Err() != nil { + return + } + if err := s.SnapshotOnce(ctx); err != nil { + log.Ctx(ctx).Warnf("blend: price snapshot pass failed: %v", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} + +// 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) + } + + var rows []blenddata.OraclePrice + var errs []error + for oracle, oracleTargets := range byOracle { + oracleRows, oracleErr := s.snapshotOracle(ctx, oracle, oracleTargets) + if oracleErr != nil { + errs = append(errs, fmt.Errorf("blend: price snapshot: oracle %s: %w", oracle, oracleErr)) + } + rows = append(rows, oracleRows...) + } + + 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 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. +func (s *PriceSnapshotService) snapshotOracle(ctx context.Context, oracle string, targets []blenddata.PriceTarget) ([]blenddata.OraclePrice, error) { + decimalsVal, err := s.metadata.FetchSingleField(ctx, oracle, "decimals") + if err != nil { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + return nil, fmt.Errorf("fetching decimals: %w", err) + } + decimals, ok := u32Val(decimalsVal) + if !ok { + s.metrics.FetchesTotal.WithLabelValues("error").Inc() + return nil, fmt.Errorf("decimals: expected u32, got %v", decimalsVal.Type) + } + + rows := make([]blenddata.OraclePrice, 0, len(targets)) + var errs []error + 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 + } + + 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, 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..a90612a00 --- /dev/null +++ b/internal/services/blend/prices_test.go @@ -0,0 +1,363 @@ +// 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) + + 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, 1_700_000_000), nil).Once() + meta.On("FetchSingleField", mock.Anything, oracle1, "lastprice", []xdr.ScVal{assetYArg}). + Return(priceDataScVal(200_000_000, 1_700_000_001), 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, 1_700_000_002), 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(1_700_000_000), 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)) +} + +// 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_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, 1_700_000_000), 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") +} From b55aef904b37dd46cf242853e946555353aade07 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Wed, 8 Jul 2026 16:37:11 -0400 Subject: [PATCH 5/8] feat(ingest): wire blend price snapshot task behind BLEND_PRICE_INTERVAL --- cmd/ingest.go | 2 ++ cmd/utils/global_options.go | 28 ++++++++++++++++++++++++++++ docker-compose.yaml | 4 ++++ internal/ingest/ingest.go | 24 +++++++++++++++++++++++- 4 files changed, 57 insertions(+), 1 deletion(-) 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/ingest/ingest.go b/internal/ingest/ingest.go index eeaf0aff9..8b79eca69 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,22 @@ func setupDeps(cfg Configs) (services.IngestService, error) { return nil, fmt.Errorf("instantiating contract metadata service: %w", err) } + // Start 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. + 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) + } + go blendPrices.Run(ctx) + } + // 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 From 9e980c49d6d8ac0b2b9c4a1634b8675198899099 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Wed, 8 Jul 2026 20:28:20 -0400 Subject: [PATCH 6/8] fix(blend): wait the full interval after each price snapshot pass --- internal/services/blend/prices.go | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/internal/services/blend/prices.go b/internal/services/blend/prices.go index 928b5ad3d..ed7220a37 100644 --- a/internal/services/blend/prices.go +++ b/internal/services/blend/prices.go @@ -95,24 +95,25 @@ func NewPriceSnapshotService(cfg PriceSnapshotConfig) (*PriceSnapshotService, er }, nil } -// Run snapshots once immediately, then every s.interval until ctx is -// cancelled. A failed pass is logged, never fatal — the next tick tries -// again. Mirrors protocol_migrate.go's refreshTargetTip ticker-loop shape. +// 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) { - ticker := time.NewTicker(s.interval) - defer ticker.Stop() + timer := time.NewTimer(0) // fires immediately: the first pass runs at startup + defer timer.Stop() for { - if ctx.Err() != nil { + 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) } - select { - case <-ctx.Done(): - return - case <-ticker.C: - } + timer.Reset(s.interval) } } From b07b54aef88c3eaef9c64bb245f03aa6fa799d29 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Thu, 9 Jul 2026 09:54:44 -0400 Subject: [PATCH 7/8] fix(ingest): start post-lock background tasks only after acquiring the live advisory lock The blend price snapshot task started in setupDeps, so an instance that never wins the advisory lock (rolling deploy, standby pod) still ran snapshot passes, duplicating oracle RPC load and writes. Long-lived background tasks now register as PostLockTasks and launch inside startLiveIngestion after the lock is held. --- internal/ingest/ingest.go | 8 ++++++-- internal/services/ingest.go | 8 ++++++++ internal/services/ingest_live.go | 7 +++++++ 3 files changed, 21 insertions(+), 2 deletions(-) diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 8b79eca69..b9bd69ecf 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -175,8 +175,9 @@ func setupDeps(cfg Configs) (services.IngestService, error) { return nil, fmt.Errorf("instantiating contract metadata service: %w", err) } - // Start the Blend v2 oracle price snapshot task. Live-only: backfill replays historical + // 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, @@ -188,7 +189,9 @@ func setupDeps(cfg Configs) (services.IngestService, error) { if err != nil { return nil, fmt.Errorf("instantiating blend price snapshot service: %w", err) } - go blendPrices.Run(ctx) + // 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 @@ -292,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/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 { From adec3b61da8e98d73a1cf9b675890132617dbfe1 Mon Sep 17 00:00:00 2001 From: Aditya Vyas Date: Thu, 9 Jul 2026 15:42:12 -0400 Subject: [PATCH 8/8] fix(blend): skip stale and non-positive oracle prices at snapshot time The pool contract itself (pool.rs::load_price) rejects prices older than 24h or <= 0, so persisting them only lets the API price positions the pool would refuse to. Skip both at snapshot time, counted as new "stale"/"invalid" fetch outcomes, and add a wallet_blend_price_oldest_age_seconds gauge covering every decoded price including skipped-as-stale ones, so a dead oracle surfaces as unbounded gauge growth. --- internal/metrics/blend.go | 17 +++- internal/services/blend/prices.go | 51 +++++++++--- internal/services/blend/prices_test.go | 104 +++++++++++++++++++++++-- 3 files changed, 155 insertions(+), 17 deletions(-) diff --git a/internal/metrics/blend.go b/internal/metrics/blend.go index 379aff54a..b0cab73fd 100644 --- a/internal/metrics/blend.go +++ b/internal/metrics/blend.go @@ -13,12 +13,19 @@ type BlendPriceMetrics struct { 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, or none (SEP-40 Option::None — no price recorded yet). + // 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 { @@ -30,13 +37,17 @@ func newBlendPriceMetrics(reg prometheus.Registerer) *BlendPriceMetrics { }), FetchesTotal: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "wallet_blend_price_fetches_total", - Help: "Blend v2 oracle price fetch outcomes, labeled by result (success, error, none).", + 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) + reg.MustRegister(m.SnapshotDuration, m.FetchesTotal, m.PricesTracked, m.OldestPriceAge) return m } diff --git a/internal/services/blend/prices.go b/internal/services/blend/prices.go index ed7220a37..c5c2ec8e5 100644 --- a/internal/services/blend/prices.go +++ b/internal/services/blend/prices.go @@ -19,6 +19,11 @@ // 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 @@ -40,6 +45,12 @@ import ( "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 @@ -139,15 +150,21 @@ func (s *PriceSnapshotService) SnapshotOnce(ctx context.Context) error { 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, oracleErr := s.snapshotOracle(ctx, oracle, oracleTargets) + 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) @@ -173,24 +190,29 @@ func (s *PriceSnapshotService) SnapshotOnce(ctx context.Context) error { // snapshotOracle fetches decimals() once for oracle, then lastprice(asset) // for each of targets, skipping (not erroring on) a SEP-40 Option::None -// response. 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. -func (s *PriceSnapshotService) snapshotOracle(ctx context.Context, oracle string, targets []blenddata.PriceTarget) ([]blenddata.OraclePrice, error) { +// 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, fmt.Errorf("fetching decimals: %w", err) + return nil, 0, fmt.Errorf("fetching decimals: %w", err) } decimals, ok := u32Val(decimalsVal) if !ok { s.metrics.FetchesTotal.WithLabelValues("error").Inc() - return nil, fmt.Errorf("decimals: expected u32, got %v", decimalsVal.Type) + 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 { @@ -216,6 +238,17 @@ func (s *PriceSnapshotService) snapshotOracle(ctx context.Context, oracle string 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{ @@ -227,7 +260,7 @@ func (s *PriceSnapshotService) snapshotOracle(ctx context.Context, oracle string }) } - return rows, errors.Join(errs...) + return rows, oldestAge, errors.Join(errs...) } // snapshotComet derives the BLND and Comet LP-share USD prices from the diff --git a/internal/services/blend/prices_test.go b/internal/services/blend/prices_test.go index a90612a00..e58a74424 100644 --- a/internal/services/blend/prices_test.go +++ b/internal/services/blend/prices_test.go @@ -188,17 +188,23 @@ func TestSnapshotOnce_ReserveAssets(t *testing.T) { 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, 1_700_000_000), nil).Once() + Return(priceDataScVal(100_000_000, tsX), nil).Once() meta.On("FetchSingleField", mock.Anything, oracle1, "lastprice", []xdr.ScVal{assetYArg}). - Return(priceDataScVal(200_000_000, 1_700_000_001), nil).Once() + 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, 1_700_000_002), nil).Once() + Return(priceDataScVal(300_000_000, tsZ), nil).Once() bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices svc := newTestSnapshotService(t, oraclePrices, meta, "", bpm) @@ -209,7 +215,7 @@ func TestSnapshotOnce_ReserveAssets(t *testing.T) { require.True(t, ok) assert.Equal(t, "100000000", rowX.Price) assert.Equal(t, int32(7), rowX.PriceDecimals) - assert.Equal(t, int64(1_700_000_000), rowX.PriceTimestamp) + assert.Equal(t, int64(tsX), rowX.PriceTimestamp) rowY, ok := getOraclePriceRow(t, ctx, pool, oracle1, assetY) require.True(t, ok) @@ -224,6 +230,8 @@ func TestSnapshotOnce_ReserveAssets(t *testing.T) { 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 @@ -260,6 +268,92 @@ func TestSnapshotOnce_NonePriceSkipped(t *testing.T) { 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) { @@ -287,7 +381,7 @@ func TestSnapshotOnce_OracleErrorIsolated(t *testing.T) { 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, 1_700_000_000), nil).Once() + Return(priceDataScVal(500_000_000, uint64(time.Now().Unix())), nil).Once() bpm := metrics.NewMetrics(prometheus.NewRegistry()).BlendPrices svc := newTestSnapshotService(t, oraclePrices, meta, "", bpm)