From 458c734350f1d1cdfcb4a5e27b665071f4175718 Mon Sep 17 00:00:00 2001 From: ByteYue Date: Mon, 3 Aug 2026 17:16:08 +0800 Subject: [PATCH] feat(oracle): add deterministic Binance index feed Add the source type 3 Binance USD-M indexPriceKlines adapter with deterministic nonce-to-bucket mapping, exact closed-bucket validation, bounded HTTP responses, and fixed-point payload encoding. Reject task history mismatches during registration and runtime reconciliation, retain the existing 500,000 callback gas policy, and document the canonical PriceFeedResolver payload. Tests cover URI/config validation, restart mapping, response parsing, precision bounds, localhost HTTP transport, canonical wrapper encoding, manager reconciliation, and execution calldata. --- .../execute/src/onchain_config/jwk_oracle.rs | 44 +- .../pipe-exec-layer-ext-v2/relayer/README.md | 51 +- .../relayer/src/data_source.rs | 25 +- .../pipe-exec-layer-ext-v2/relayer/src/lib.rs | 4 + .../relayer/src/oracle_manager.rs | 49 +- .../relayer/src/price_feed_source.rs | 950 ++++++++++++++++++ .../relayer/src/uri_parser.rs | 24 +- 7 files changed, 1135 insertions(+), 12 deletions(-) create mode 100644 crates/pipe-exec-layer-ext-v2/relayer/src/price_feed_source.rs diff --git a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs index a8d61da6b1..cfa75c4852 100644 --- a/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs +++ b/crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/jwk_oracle.rs @@ -52,7 +52,7 @@ fn parse_source_from_issuer(issuer: &[u8]) -> Option<(u32, u64)> { fn callback_gas_limit(source_type: u32) -> Result { match source_type { - source_types::BLOCKCHAIN => Ok(CALLBACK_GAS_LIMIT), + source_types::BLOCKCHAIN | source_types::PRICE_FEED => Ok(CALLBACK_GAS_LIMIT), _ => Err(format!("Unsupported oracle source type: {source_type}")), } } @@ -190,6 +190,17 @@ fn construct_unsupported_oracle_batch_transaction( mod tests { use super::*; use alloy_consensus::Transaction; + use alloy_primitives::I256; + + sol! { + struct PricePayloadForTest { + uint256 feedId; + uint64 roundId; + uint64 resolvedAt; + uint8 decimals; + int256 price; + } + } fn wrapped_jwk(nonce: u128, position: U256, payload: &[u8]) -> JWKStruct { JWKStruct { @@ -242,14 +253,39 @@ mod tests { assert_eq!(call.callbackGasLimits, vec![U256::from(CALLBACK_GAS_LIMIT)]); } + #[test] + fn preserves_price_feed_coordinates_payload_and_callback_gas() { + let payload = PricePayloadForTest { + feedId: U256::from(2001), + roundId: 28_500_000, + resolvedAt: 1_710_000_059_999, + decimals: 8, + price: "40067545000".parse::().unwrap(), + } + .abi_encode(); + let provider = provider( + b"gravity://3/2001/price_feed?provider=binance_index_kline_v1", + vec![wrapped_jwk(1, U256::from(1_710_000_059_999u64), &payload)], + ); + + let tx = construct_oracle_record_transaction(provider, 0, 0).unwrap(); + let call = recordBatchCall::abi_decode(tx.input()).unwrap(); + assert_eq!(call.sourceType, source_types::PRICE_FEED); + assert_eq!(call.sourceId, U256::from(2001)); + assert_eq!(call.nonces, vec![1]); + assert_eq!(call.blockNumbers, vec![U256::from(1_710_000_059_999u64)]); + assert_eq!(call.payloads, vec![Bytes::from(payload)]); + assert_eq!(call.callbackGasLimits, vec![U256::from(CALLBACK_GAS_LIMIT)]); + } + #[test] fn rejects_provider_source_types_not_implemented_by_core() { let provider = provider( - b"gravity://3/1001/price_feed", - vec![wrapped_jwk(1, U256::from(60_000), b"price")], + b"gravity://6/1001/settlement", + vec![wrapped_jwk(1, U256::from(60_000), b"settlement")], ); let error = construct_oracle_record_transaction(provider, 0, 0).unwrap_err(); - assert_eq!(error, "Unsupported oracle source type: 3"); + assert_eq!(error, "Unsupported oracle source type: 6"); } #[test] diff --git a/crates/pipe-exec-layer-ext-v2/relayer/README.md b/crates/pipe-exec-layer-ext-v2/relayer/README.md index 379da6deb8..6f57da31e7 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/README.md +++ b/crates/pipe-exec-layer-ext-v2/relayer/README.md @@ -2,8 +2,8 @@ This crate converts finalized external-source observations into the UnsupportedJWK payloads used by Gravity validator consensus. This core slice -implements source type `0` (`GravityPortal.MessageSent`). Provider-specific -source types are added in separate modules and PRs. +implements source type `0` (`GravityPortal.MessageSent`) and source type `3` +(Binance USD-M index-price klines). Other providers remain separate slices. ## Task Identity @@ -24,9 +24,56 @@ Source type `0` example: gravity://0/1/events?portal=0x0000000000000000000000000000000000000001&fromBlock=19000000 ``` +Source type `3` example: + +```text +gravity://3/2001/price_feed?provider=binance_index_kline_v1&pair=TSLAUSDT&interval=1m&bucketStartMs=1710000000000&decimals=8&graceMs=120000 +``` + RPC URLs are local validator configuration. They are not stored in the task URI or committed on-chain. +For Binance, the local URL is the USD-M Futures base URL. The adapter appends +`/fapi/v1/indexPriceKlines` and the deterministic query parameters. The public +index-kline endpoint does not require API credentials. `baseUrl` and unknown +query parameters are rejected in the on-chain URI. + +## Binance Index Price Delivery + +For one `(sourceType=3, feedId)` task, delivery nonce `n` maps to exactly one +bucket: + +```text +bucketStart(n) = configuredBucketStart + (n - 1) * interval +bucketEnd(n) = bucketStart(n) + interval - 1 +sourcePosition = bucketEnd(n) +roundId = bucketStart(n) / interval +resolvedAt = bucketEnd(n) +``` + +The adapter waits until `bucketEnd + graceMs`, requests exactly that bucket, +and accepts exactly one response row whose open and close timestamps match. +The callback payload is: + +```solidity +abi.encode( + uint256 feedId, + uint64 roundId, + uint64 resolvedAt, + uint8 decimals, + int256 price +) +``` + +The decimal parser rejects negative, zero, malformed, overflowing, or +non-representable prices. HTTP connect/request timeouts and a 64 KiB response +limit bound validator resource use. Supported fixed intervals are `1m`, `3m`, +`5m`, `15m`, `30m`, `1h`, `2h`, `4h`, `6h`, `8h`, `12h`, `1d`, and `3d`. + +Changing the interval or bucket origin for an active feed changes the +nonce-to-position history. Registration and runtime reconciliation reject that +mismatch; deploy the changed task under a new `feedId`. + ## Canonical Delivery Each source observation becomes an `OracleData` value: diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs index 12a052b502..8aef63daa9 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/data_source.rs @@ -7,7 +7,7 @@ use alloy_primitives::{Bytes, U256}; use anyhow::Result; use async_trait::async_trait; -use crate::blockchain_source::BlockchainEventSource; +use crate::{blockchain_source::BlockchainEventSource, price_feed_source::PriceFeedSource}; /// Data returned by oracle data sources /// @@ -16,6 +16,7 @@ use crate::blockchain_source::BlockchainEventSource; pub struct OracleData { /// Strictly increasing nonce for this (sourceType, sourceId) pair /// - For Blockchain: MessageSent.nonce + /// - For `PriceFeed`: deterministic delivery sequence pub nonce: u128, /// Source-defined restart position committed with this delivery @@ -33,6 +34,7 @@ pub struct OracleData { pub trait OracleDataSource: Send + Sync { /// Get the source type (corresponds to NativeOracle.sourceType) /// - 0: BLOCKCHAIN + /// - 3: `PRICE_FEED` fn source_type(&self) -> u32; /// Get the source ID (corresponds to NativeOracle.sourceId) @@ -50,6 +52,9 @@ pub trait OracleDataSource: Send + Sync { pub mod source_types { /// Blockchain cross-chain events (e.g., GravityPortal.MessageSent) pub const BLOCKCHAIN: u32 = 0; + + /// Deterministic external price buckets + pub const PRICE_FEED: u32 = 3; } /// Extensible enum for runtime dispatch of data sources @@ -62,36 +67,47 @@ pub mod source_types { pub enum DataSourceKind { /// Blockchain cross-chain events (sourceType=0) Blockchain(BlockchainEventSource), + + /// Deterministic external price buckets (sourceType=3) + PriceFeed(PriceFeedSource), } impl DataSourceKind { pub(crate) async fn last_nonce(&self) -> Option { match self { Self::Blockchain(source) => source.last_nonce().await, + Self::PriceFeed(source) => source.last_nonce().await, } } pub(crate) async fn last_nonce_position(&self) -> Option { match self { Self::Blockchain(source) => source.last_nonce_block().await, + Self::PriceFeed(source) => source.last_nonce_position().await, } } - pub(crate) async fn reconcile_progress(&self, nonce: u128, position: u64) { + pub(crate) async fn reconcile_progress(&self, nonce: u128, position: u64) -> Result<()> { match self { - Self::Blockchain(source) => source.reconcile_progress(nonce, position).await, + Self::Blockchain(source) => { + source.reconcile_progress(nonce, position).await; + Ok(()) + } + Self::PriceFeed(source) => source.reconcile_progress(nonce, position).await, } } pub(crate) fn cursor(&self) -> u64 { match self { Self::Blockchain(source) => source.cursor(), + Self::PriceFeed(source) => source.cursor(), } } pub(crate) const fn source_id_u64(&self) -> u64 { match self { Self::Blockchain(source) => source.chain_id(), + Self::PriceFeed(source) => source.feed_id(), } } } @@ -101,18 +117,21 @@ impl OracleDataSource for DataSourceKind { fn source_type(&self) -> u32 { match self { Self::Blockchain(_) => source_types::BLOCKCHAIN, + Self::PriceFeed(_) => source_types::PRICE_FEED, } } fn source_id(&self) -> U256 { match self { Self::Blockchain(s) => s.source_id(), + Self::PriceFeed(s) => s.source_id(), } } async fn poll(&self) -> Result> { match self { Self::Blockchain(s) => s.poll().await, + Self::PriceFeed(s) => s.poll().await, } } } diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/lib.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/lib.rs index deed4d3596..c6012c32cd 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/lib.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/lib.rs @@ -20,6 +20,9 @@ pub mod data_source; /// Blockchain event source (GravityPortal.MessageSent) pub mod blockchain_source; +/// Binance index-price feed source +pub mod price_feed_source; + /// Factory for creating data sources pub mod factory; @@ -44,4 +47,5 @@ pub use data_source::{source_types, DataSourceKind, OracleData, OracleDataSource pub use eth_client::EthHttpCli; pub use factory::DataSourceFactory; pub use oracle_manager::{JWKStruct, OracleRelayerManager, PollResult}; +pub use price_feed_source::PriceFeedSource; pub use uri_parser::{parse_oracle_uri, ParsedOracleTask}; diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs index 0894759622..a29b1d624b 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/oracle_manager.rs @@ -6,6 +6,7 @@ use crate::{ blockchain_source::BlockchainEventSource, data_source::{source_types, DataSourceKind, OracleDataSource}, persistence::{load_state_if_exists, state_file_path, RelayerState, SourceState}, + price_feed_source::PriceFeedSource, uri_parser::{parse_oracle_uri, ParsedOracleTask}, }; use anyhow::{anyhow, Result}; @@ -309,6 +310,15 @@ impl OracleRelayerManager { .await?; Ok(DataSourceKind::Blockchain(source)) } + source_types::PRICE_FEED => { + let source = PriceFeedSource::from_task_with_progress( + task, + latest_onchain_nonce, + latest_onchain_position, + Some(rpc_url), + )?; + Ok(DataSourceKind::PriceFeed(source)) + } _ => Err(anyhow!("Unknown source type: {}", task.source_type)), } } @@ -346,7 +356,7 @@ impl OracleRelayerManager { onchain_position, "Reconciling local source with confirmed on-chain progress" ); - source.reconcile_progress(onchain_nonce, onchain_position).await; + source.reconcile_progress(onchain_nonce, onchain_position).await?; } } @@ -461,6 +471,8 @@ mod tests { const URI: &str = "gravity://0/1/events?portal=0x0000000000000000000000000000000000000001&fromBlock=100"; + const PRICE_URI: &str = + "gravity://3/2001/price_feed?provider=binance_index_kline_v1&pair=TSLAUSDT&interval=1m&bucketStartMs=1710000000000&decimals=8"; fn state(last_nonce: u128, last_position: u64, cursor: u64) -> RelayerState { let mut state = RelayerState::new(); @@ -542,4 +554,39 @@ mod tests { let position = u128::from(u64::MAX) + 1; assert!(u64::try_from(position).is_err()); } + + #[tokio::test] + async fn adds_binance_price_feed_without_network_access() { + let datadir = tempfile::tempdir().unwrap(); + let manager = OracleRelayerManager::new(datadir.path().to_path_buf()); + + manager.add_uri(PRICE_URI, "https://fapi.binance.com", 0, 0).await.unwrap(); + + assert!(manager.has_uri(PRICE_URI).await); + } + + #[tokio::test] + async fn rejects_binance_history_mismatch_during_registration() { + let datadir = tempfile::tempdir().unwrap(); + let manager = OracleRelayerManager::new(datadir.path().to_path_buf()); + + let error = manager + .add_uri(PRICE_URI, "https://fapi.binance.com", 2, 1_710_000_059_999) + .await + .unwrap_err(); + + assert!(error.to_string().contains("task history mismatch")); + } + + #[tokio::test] + async fn rejects_binance_history_mismatch_during_runtime_reconcile() { + let datadir = tempfile::tempdir().unwrap(); + let manager = OracleRelayerManager::new(datadir.path().to_path_buf()); + manager.add_uri(PRICE_URI, "https://fapi.binance.com", 1, 1_710_000_059_999).await.unwrap(); + + let error = + manager.poll_uri(PRICE_URI, Some(2), Some(1_710_000_060_000)).await.unwrap_err(); + + assert!(error.to_string().contains("task history mismatch")); + } } diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/price_feed_source.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/price_feed_source.rs new file mode 100644 index 0000000000..02b1c8a468 --- /dev/null +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/price_feed_source.rs @@ -0,0 +1,950 @@ +//! Binance index-price feed source. +//! +//! Each delivery nonce maps to one immutable, closed Binance USD-M +//! `indexPriceKlines` bucket. The adapter verifies the exact bucket timestamps +//! and emits one close price through the existing UnsupportedJWK consensus path. + +use crate::{ + data_source::{source_types, OracleData, OracleDataSource}, + uri_parser::ParsedOracleTask, +}; +use alloy_primitives::{Bytes, I256, U256}; +use alloy_sol_macro::sol; +use alloy_sol_types::SolValue; +use anyhow::{anyhow, Context, Result}; +use async_trait::async_trait; +use reqwest::Client; +use serde_json::Value; +use std::{ + sync::atomic::{AtomicU64, Ordering}, + time::{Duration, SystemTime, UNIX_EPOCH}, +}; +use tokio::sync::Mutex; +use tracing::info; +use url::Url; + +const PROVIDER_BINANCE_INDEX_KLINE: &str = "binance_index_kline_v1"; +const DEFAULT_BINANCE_GRACE_MS: u64 = 120_000; +const BINANCE_HTTP_TIMEOUT: Duration = Duration::from_secs(15); +const MAX_BINANCE_RESPONSE_BYTES: usize = 64 * 1024; +const MAX_PRICE_DECIMALS: u8 = 18; +const BINANCE_TASK_PARAMETERS: &[&str] = + &["provider", "pair", "interval", "bucketStartMs", "decimals", "graceMs"]; + +sol! { + struct PricePayloadSol { + uint256 feedId; + uint64 roundId; + uint64 resolvedAt; + uint8 decimals; + int256 price; + } +} + +#[derive(Debug, Clone, Copy, Default)] +struct LastPriceRound { + nonce: u128, + source_position: u64, +} + +impl LastPriceRound { + const fn is_initialized(self) -> bool { + self.nonce > 0 + } +} + +#[derive(Debug, Clone)] +struct BinanceIndexKlineConfig { + base_url: String, + pair: String, + interval: String, + interval_ms: u64, + bucket_start_ms: u64, + grace_ms: u64, + decimals: u8, +} + +#[derive(Debug, Clone)] +struct BinanceIndexKlineRound { + endpoint_url: String, + bucket_start_ms: u64, + bucket_end_ms: u64, + round_id: u64, + resolved_at: u64, + delivery_nonce: u128, + source_position: u64, +} + +impl BinanceIndexKlineConfig { + fn round_for_delivery_nonce(&self, delivery_nonce: u128) -> Result { + let offset = delivery_nonce + .checked_sub(1) + .ok_or_else(|| anyhow!("Binance delivery nonce must start at 1"))?; + let offset_ms = u128::from(self.interval_ms) + .checked_mul(offset) + .ok_or_else(|| anyhow!("Binance bucket offset overflow"))?; + let bucket_start_ms = u128::from(self.bucket_start_ms) + .checked_add(offset_ms) + .ok_or_else(|| anyhow!("Binance bucket start overflow"))?; + let bucket_start_ms = u64::try_from(bucket_start_ms) + .map_err(|_| anyhow!("Binance bucket start exceeds u64"))?; + let bucket_end_ms = bucket_start_ms + .checked_add(self.interval_ms) + .and_then(|value| value.checked_sub(1)) + .ok_or_else(|| anyhow!("Binance bucket end overflow"))?; + let round_id = bucket_start_ms / self.interval_ms; + let endpoint_url = build_binance_index_kline_url( + &self.base_url, + &self.pair, + &self.interval, + bucket_start_ms, + bucket_end_ms, + )?; + + Ok(BinanceIndexKlineRound { + endpoint_url, + bucket_start_ms, + bucket_end_ms, + round_id, + resolved_at: bucket_end_ms, + delivery_nonce, + source_position: bucket_end_ms, + }) + } +} + +/// Price feed data source for `sourceType=3`. +#[derive(Debug)] +pub struct PriceFeedSource { + feed_id: u64, + config: BinanceIndexKlineConfig, + client: Client, + cursor: AtomicU64, + last_round: Mutex, +} + +impl PriceFeedSource { + /// Create a source without a validator-local endpoint mapping. + /// + /// Binance sources require a URL mapping, so production callers use + /// [`Self::from_task_with_rpc`] or [`Self::from_task_with_progress`]. + pub fn from_task(task: &ParsedOracleTask, latest_onchain_nonce: u128) -> Result { + Self::from_task_with_rpc(task, latest_onchain_nonce, None) + } + + /// Create a source with the validator-local Binance base URL. + pub fn from_task_with_rpc( + task: &ParsedOracleTask, + latest_onchain_nonce: u128, + rpc_url: Option<&str>, + ) -> Result { + Self::from_task_with_progress(task, latest_onchain_nonce, 0, rpc_url) + } + + pub(crate) fn from_task_with_progress( + task: &ParsedOracleTask, + latest_onchain_nonce: u128, + latest_onchain_position: u64, + rpc_url: Option<&str>, + ) -> Result { + if task.source_type != source_types::PRICE_FEED { + return Err(anyhow!("PriceFeedSource requires sourceType={}", source_types::PRICE_FEED)); + } + if task.task_type != "price_feed" { + return Err(anyhow!( + "PriceFeedSource requires task type 'price_feed', got '{}'", + task.task_type + )); + } + + match task.params.get("provider").map(String::as_str) { + Some(PROVIDER_BINANCE_INDEX_KLINE) => {} + Some(provider) => return Err(anyhow!("Unsupported price feed provider '{provider}'")), + None => return Err(anyhow!("Missing 'provider' parameter for price feed")), + } + + if task.params.contains_key("baseUrl") { + return Err(anyhow!( + "Binance baseUrl must be validator-local relayer config, not an on-chain URI parameter" + )); + } + for parameter in task.params.keys() { + if !BINANCE_TASK_PARAMETERS.contains(¶meter.as_str()) { + return Err(anyhow!("Unsupported Binance index kline parameter '{parameter}'")); + } + } + + let pair = task + .params + .get("pair") + .ok_or_else(|| anyhow!("Missing 'pair' parameter for Binance index kline price feed"))? + .clone(); + validate_binance_pair(&pair)?; + let interval = task.params.get("interval").cloned().unwrap_or_else(|| "1m".to_string()); + let interval_ms = binance_interval_ms(&interval)?; + let bucket_start_ms = parse_required::(task, "bucketStartMs")?; + if bucket_start_ms % interval_ms != 0 { + return Err(anyhow!( + "Binance index kline bucketStartMs {} is not aligned to interval {}", + bucket_start_ms, + interval + )); + } + let decimals = parse_required::(task, "decimals")?; + if decimals > MAX_PRICE_DECIMALS { + return Err(anyhow!( + "price feed decimals {} exceeds maximum {}", + decimals, + MAX_PRICE_DECIMALS + )); + } + let grace_ms = parse_optional(task, "graceMs")?.unwrap_or(DEFAULT_BINANCE_GRACE_MS); + let base_url = binance_base_url(rpc_url)?; + let client = Client::builder() + .no_proxy() + .use_rustls_tls() + .connect_timeout(Duration::from_secs(5)) + .timeout(BINANCE_HTTP_TIMEOUT) + .build() + .context("failed to build Binance index kline HTTP client")?; + let config = BinanceIndexKlineConfig { + base_url, + pair, + interval, + interval_ms, + bucket_start_ms, + grace_ms, + decimals, + }; + + if latest_onchain_nonce == 0 && latest_onchain_position != 0 { + return Err(anyhow!("Binance task has zero nonce with nonzero source position")); + } + let previous_position = if latest_onchain_nonce == 0 { + None + } else { + let expected = config.round_for_delivery_nonce(latest_onchain_nonce)?.source_position; + if latest_onchain_position != 0 && latest_onchain_position != expected { + return Err(anyhow!( + "Binance task history mismatch: nonce {} implies source position {}, confirmed position is {}; use a new feedId for a new bucket origin or interval", + latest_onchain_nonce, + expected, + latest_onchain_position + )); + } + Some(expected) + }; + let last_round = previous_position + .map(|source_position| LastPriceRound { nonce: latest_onchain_nonce, source_position }) + .unwrap_or_default(); + let initial_cursor = previous_position.unwrap_or(0); + + info!( + target: "price_feed_source", + feed_id = task.source_id, + provider = PROVIDER_BINANCE_INDEX_KLINE, + pair = config.pair.as_str(), + interval = config.interval.as_str(), + bucket_start_ms, + latest_onchain_nonce, + "Created Binance index kline PriceFeedSource" + ); + + Ok(Self { + feed_id: task.source_id, + config, + client, + cursor: AtomicU64::new(initial_cursor), + last_round: Mutex::new(last_round), + }) + } + + /// Last delivery nonce returned or reconciled from `NativeOracle`. + pub async fn last_nonce(&self) -> Option { + let state = *self.last_round.lock().await; + state.is_initialized().then_some(state.nonce) + } + + /// Bucket close timestamp associated with the last delivery nonce. + pub async fn last_nonce_position(&self) -> Option { + let state = *self.last_round.lock().await; + state.is_initialized().then_some(state.source_position) + } + + /// Reconcile local state with a round already recorded by `NativeOracle`. + pub async fn reconcile_progress(&self, nonce: u128, source_position: u64) -> Result<()> { + if nonce == 0 { + if source_position != 0 { + return Err(anyhow!("Binance task has zero nonce with nonzero source position")); + } + *self.last_round.lock().await = LastPriceRound::default(); + self.cursor.store(0, Ordering::Relaxed); + return Ok(()); + } + + let expected = self.config.round_for_delivery_nonce(nonce)?.source_position; + if source_position != 0 && source_position != expected { + return Err(anyhow!( + "Binance task history mismatch: nonce {} implies source position {}, confirmed position is {}; use a new feedId for a new bucket origin or interval", + nonce, + expected, + source_position + )); + } + + let mut state = self.last_round.lock().await; + if nonce < state.nonce { + return Err(anyhow!( + "Binance task cannot reconcile backwards from nonce {} to {}", + state.nonce, + nonce + )); + } + *state = LastPriceRound { nonce, source_position: expected }; + self.cursor.store(expected, Ordering::Relaxed); + Ok(()) + } + + /// Current bucket-close cursor used for relayer persistence. + pub fn cursor(&self) -> u64 { + self.cursor.load(Ordering::Relaxed) + } + + /// Feed identifier used as `NativeOracle` sourceId. + pub const fn feed_id(&self) -> u64 { + self.feed_id + } +} + +#[async_trait] +impl OracleDataSource for PriceFeedSource { + fn source_type(&self) -> u32 { + source_types::PRICE_FEED + } + + fn source_id(&self) -> U256 { + U256::from(self.feed_id) + } + + async fn poll(&self) -> Result> { + let mut state = self.last_round.lock().await; + let next_delivery_nonce = state + .nonce + .checked_add(1) + .ok_or_else(|| anyhow!("Binance index kline delivery nonce overflow"))?; + let round = self.config.round_for_delivery_nonce(next_delivery_nonce)?; + if !is_binance_bucket_ready(&self.config, &round)? { + return Ok(vec![]); + } + + let price = fetch_binance_index_kline_price(&self.client, &self.config, &round).await?; + let resolver_payload = encode_price_payload( + self.feed_id, + round.round_id, + round.resolved_at, + self.config.decimals, + price, + ); + let wrapped_payload = SolValue::abi_encode(&( + round.delivery_nonce, + U256::from(round.source_position), + resolver_payload.as_slice(), + )); + + state.nonce = round.delivery_nonce; + state.source_position = round.source_position; + self.cursor.store(round.source_position, Ordering::Relaxed); + + Ok(vec![OracleData { + nonce: round.delivery_nonce, + source_position: round.source_position, + payload: Bytes::from(wrapped_payload), + }]) + } +} + +async fn fetch_binance_index_kline_price( + client: &Client, + config: &BinanceIndexKlineConfig, + round: &BinanceIndexKlineRound, +) -> Result { + ensure_binance_bucket_ready(config, round)?; + let response = client + .get(&round.endpoint_url) + .send() + .await + .context("failed to fetch Binance index price kline")? + .error_for_status() + .context("Binance index price kline endpoint returned an error")?; + let response = read_binance_response_limited(response).await?; + let response: Value = serde_json::from_slice(&response) + .context("failed to decode Binance index price kline response")?; + + binance_index_kline_price_from_response(config, round, &response) +} + +async fn read_binance_response_limited(mut response: reqwest::Response) -> Result> { + if response.content_length().is_some_and(|len| len > MAX_BINANCE_RESPONSE_BYTES as u64) { + return Err(binance_response_too_large()); + } + + let mut body = Vec::new(); + while let Some(chunk) = + response.chunk().await.context("failed to read Binance index price kline response")? + { + append_binance_response_chunk(&mut body, &chunk)?; + } + Ok(body) +} + +fn append_binance_response_chunk(body: &mut Vec, chunk: &[u8]) -> Result<()> { + let new_len = body.len().checked_add(chunk.len()).ok_or_else(binance_response_too_large)?; + if new_len > MAX_BINANCE_RESPONSE_BYTES { + return Err(binance_response_too_large()); + } + body.extend_from_slice(chunk); + Ok(()) +} + +fn binance_response_too_large() -> anyhow::Error { + anyhow!("Binance index price kline response exceeds {} bytes", MAX_BINANCE_RESPONSE_BYTES) +} + +fn binance_index_kline_price_from_response( + config: &BinanceIndexKlineConfig, + round: &BinanceIndexKlineRound, + response: &Value, +) -> Result { + let close = parse_binance_index_kline_close(round, response)?; + let price = parse_fixed_decimal(close, config.decimals)?; + if price <= I256::ZERO { + return Err(anyhow!("Binance index price kline close must be positive")); + } + Ok(price) +} + +fn parse_binance_index_kline_close<'a>( + round: &BinanceIndexKlineRound, + response: &'a Value, +) -> Result<&'a str> { + let rows = response + .as_array() + .ok_or_else(|| anyhow!("Binance index price kline response must be an array"))?; + if rows.len() != 1 { + return Err(anyhow!( + "Binance index price kline response must contain exactly one row, got {}", + rows.len() + )); + } + let row = rows[0] + .as_array() + .ok_or_else(|| anyhow!("Binance index price kline row must be an array"))?; + if row.len() < 7 { + return Err(anyhow!("Binance index price kline row has fewer than 7 fields")); + } + let open_time = row[0] + .as_u64() + .ok_or_else(|| anyhow!("Binance index price kline openTime must be a u64"))?; + let close = row[4] + .as_str() + .ok_or_else(|| anyhow!("Binance index price kline close must be a decimal string"))?; + let close_time = row[6] + .as_u64() + .ok_or_else(|| anyhow!("Binance index price kline closeTime must be a u64"))?; + if open_time != round.bucket_start_ms { + return Err(anyhow!( + "Binance index price kline openTime mismatch: expected {}, got {}", + round.bucket_start_ms, + open_time + )); + } + if close_time != round.bucket_end_ms { + return Err(anyhow!( + "Binance index price kline closeTime mismatch: expected {}, got {}", + round.bucket_end_ms, + close_time + )); + } + + Ok(close) +} + +fn ensure_binance_bucket_ready( + config: &BinanceIndexKlineConfig, + round: &BinanceIndexKlineRound, +) -> Result<()> { + if !is_binance_bucket_ready(config, round)? { + let ready_at = round + .bucket_end_ms + .checked_add(config.grace_ms) + .ok_or_else(|| anyhow!("Binance index kline ready time overflow"))?; + let now_ms = current_unix_millis()?; + return Err(anyhow!( + "Binance index kline bucket is not ready: readyAtMs={}, nowMs={}", + ready_at, + now_ms + )); + } + Ok(()) +} + +fn is_binance_bucket_ready( + config: &BinanceIndexKlineConfig, + round: &BinanceIndexKlineRound, +) -> Result { + let ready_at = round + .bucket_end_ms + .checked_add(config.grace_ms) + .ok_or_else(|| anyhow!("Binance index kline ready time overflow"))?; + Ok(current_unix_millis()? >= u128::from(ready_at)) +} + +fn current_unix_millis() -> Result { + Ok(SystemTime::now() + .duration_since(UNIX_EPOCH) + .context("system clock is before UNIX_EPOCH")? + .as_millis()) +} + +fn binance_base_url(rpc_url: Option<&str>) -> Result { + let base = rpc_url.filter(|value| !value.is_empty()).ok_or_else(|| { + anyhow!("Binance price feed requires a validator-local relayer URL mapping") + })?; + validate_http_url(base, "Binance base URL")?; + Ok(base.to_string()) +} + +fn validate_http_url(value: &str, label: &str) -> Result { + let url = Url::parse(value).map_err(|e| anyhow!("invalid {label}: {e}"))?; + if !matches!(url.scheme(), "http" | "https") || url.host_str().is_none() { + return Err(anyhow!("{label} must be an http(s) URL with a host")); + } + if !url.username().is_empty() || url.password().is_some() { + return Err(anyhow!("{label} must not contain URL userinfo")); + } + Ok(url) +} + +fn validate_binance_pair(pair: &str) -> Result<()> { + if pair.is_empty() || + pair.len() > 32 || + !pair.bytes().all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit()) + { + return Err(anyhow!("Binance pair must contain 1-32 uppercase ASCII letters or digits")); + } + Ok(()) +} + +fn build_binance_index_kline_url( + base_url: &str, + pair: &str, + interval: &str, + bucket_start_ms: u64, + bucket_end_ms: u64, +) -> Result { + let mut url = validate_http_url(base_url, "Binance base URL")?; + url.set_path("/fapi/v1/indexPriceKlines"); + url.set_query(None); + url.set_fragment(None); + url.query_pairs_mut() + .append_pair("pair", pair) + .append_pair("interval", interval) + .append_pair("startTime", &bucket_start_ms.to_string()) + .append_pair("endTime", &bucket_end_ms.to_string()) + .append_pair("limit", "1"); + Ok(url.to_string()) +} + +fn binance_interval_ms(interval: &str) -> Result { + match interval { + "1m" => Ok(60_000), + "3m" => Ok(180_000), + "5m" => Ok(300_000), + "15m" => Ok(900_000), + "30m" => Ok(1_800_000), + "1h" => Ok(3_600_000), + "2h" => Ok(7_200_000), + "4h" => Ok(14_400_000), + "6h" => Ok(21_600_000), + "8h" => Ok(28_800_000), + "12h" => Ok(43_200_000), + "1d" => Ok(86_400_000), + "3d" => Ok(259_200_000), + _ => Err(anyhow!("Unsupported Binance index kline interval '{interval}'")), + } +} + +fn encode_price_payload( + feed_id: u64, + round_id: u64, + resolved_at: u64, + decimals: u8, + price: I256, +) -> Vec { + PricePayloadSol { + feedId: U256::from(feed_id), + roundId: round_id, + resolvedAt: resolved_at, + decimals, + price, + } + .abi_encode() +} + +fn parse_fixed_decimal(value: &str, decimals: u8) -> Result { + if value.starts_with('-') { + return Err(anyhow!("price cannot be negative")); + } + let (whole, fraction) = value.split_once('.').unwrap_or((value, "")); + if whole.is_empty() && fraction.is_empty() { + return Err(anyhow!("empty decimal price")); + } + if !whole.chars().all(|c| c.is_ascii_digit()) { + return Err(anyhow!("invalid decimal price whole component")); + } + if !fraction.chars().all(|c| c.is_ascii_digit()) { + return Err(anyhow!("invalid decimal price fractional component")); + } + + let mut scaled = String::with_capacity(whole.len() + decimals as usize); + scaled.push_str(if whole.is_empty() { "0" } else { whole }); + let decimals = decimals as usize; + if fraction.len() > decimals && fraction[decimals..].bytes().any(|byte| byte != b'0') { + return Err(anyhow!("decimal price precision exceeds configured decimals")); + } + if fraction.len() >= decimals { + scaled.push_str(&fraction[..decimals]); + } else { + scaled.push_str(fraction); + scaled.extend(std::iter::repeat_n('0', decimals - fraction.len())); + } + + let scaled = scaled.trim_start_matches('0'); + let scaled = if scaled.is_empty() { "0" } else { scaled }; + scaled.parse::().map_err(|e| anyhow!("invalid scaled decimal price: {e}")) +} + +fn parse_required(task: &ParsedOracleTask, name: &str) -> Result +where + T: std::str::FromStr, + T::Err: std::fmt::Display, +{ + let value = task.params.get(name).ok_or_else(|| anyhow!("Missing '{name}' parameter"))?; + parse_str(value, name) +} + +fn parse_optional(task: &ParsedOracleTask, name: &str) -> Result> +where + T: std::str::FromStr, + T::Err: std::fmt::Display, +{ + task.params.get(name).map(|value| parse_str(value, name)).transpose() +} + +fn parse_str(value: &str, name: &str) -> Result +where + T: std::str::FromStr, + T::Err: std::fmt::Display, +{ + value.parse().map_err(|e| anyhow!("Invalid {name}: {e}")) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::uri_parser::parse_oracle_uri; + use std::{ + io::{Read, Write}, + net::TcpListener, + thread, + }; + + fn binance_uri() -> &'static str { + "gravity://3/2001/price_feed?provider=binance_index_kline_v1&pair=TSLAUSDT&interval=1m&bucketStartMs=1710000000000&decimals=8" + } + + fn serve_once(body: &'static str) -> (String, thread::JoinHandle) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let handle = thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = Vec::new(); + let mut buffer = [0u8; 1024]; + while !request.windows(4).any(|window| window == b"\r\n\r\n") { + let read = stream.read(&mut buffer).unwrap(); + if read == 0 { + break; + } + request.extend_from_slice(&buffer[..read]); + } + write!( + stream, + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ) + .unwrap(); + String::from_utf8(request).unwrap() + }); + (format!("http://{address}"), handle) + } + + #[test] + fn test_price_feed_requires_binance_provider() { + for uri in [ + "gravity://3/1/price_feed?pair=TSLAUSDT", + "gravity://3/1/price_feed?provider=hype", + "gravity://3/1/price_feed?provider=inline_fixture_v1", + ] { + let task = parse_oracle_uri(uri).unwrap(); + let err = PriceFeedSource::from_task(&task, 0).unwrap_err(); + assert!(err.to_string().contains("provider"), "unexpected error: {err}"); + } + } + + #[test] + fn test_binance_index_kline_source_config() { + let task = parse_oracle_uri(binance_uri()).unwrap(); + let source = + PriceFeedSource::from_task_with_rpc(&task, 0, Some("https://fapi.binance.com")) + .unwrap(); + + assert_eq!(source.config.pair, "TSLAUSDT"); + assert_eq!(source.config.interval, "1m"); + assert_eq!(source.config.interval_ms, 60_000); + assert_eq!(source.config.bucket_start_ms, 1_710_000_000_000); + let round = source.config.round_for_delivery_nonce(1).unwrap(); + assert_eq!(round.delivery_nonce, 1); + assert_eq!(round.bucket_end_ms, 1_710_000_059_999); + assert_eq!(round.round_id, 28_500_000); + assert_eq!( + round.endpoint_url, + "https://fapi.binance.com/fapi/v1/indexPriceKlines?pair=TSLAUSDT&interval=1m&startTime=1710000000000&endTime=1710000059999&limit=1" + ); + } + + #[test] + fn test_binance_index_kline_round_mapping() { + let task = parse_oracle_uri(binance_uri()).unwrap(); + let source = + PriceFeedSource::from_task_with_rpc(&task, 2, Some("https://fapi.binance.com")) + .unwrap(); + + assert_eq!(source.cursor(), 1_710_000_119_999); + let round = source.config.round_for_delivery_nonce(3).unwrap(); + assert_eq!(round.delivery_nonce, 3); + assert_eq!(round.bucket_start_ms, 1_710_000_120_000); + assert_eq!(round.bucket_end_ms, 1_710_000_179_999); + assert_eq!(round.round_id, 28_500_002); + assert_eq!(round.resolved_at, 1_710_000_179_999); + } + + #[test] + fn test_binance_rejects_mismatched_history() { + let task = parse_oracle_uri(binance_uri()).unwrap(); + let err = PriceFeedSource::from_task_with_progress( + &task, + 2, + 1_710_000_059_999, + Some("https://fapi.binance.com"), + ) + .unwrap_err(); + + assert!(err.to_string().contains("task history mismatch")); + } + + #[tokio::test] + async fn test_binance_reconcile_validates_history() { + let task = parse_oracle_uri(binance_uri()).unwrap(); + let source = + PriceFeedSource::from_task_with_rpc(&task, 1, Some("https://fapi.binance.com")) + .unwrap(); + + source.reconcile_progress(2, 1_710_000_119_999).await.unwrap(); + assert_eq!(source.last_nonce().await, Some(2)); + assert_eq!(source.last_nonce_position().await, Some(1_710_000_119_999)); + + let error = source.reconcile_progress(3, 1_710_000_120_000).await.unwrap_err(); + assert!(error.to_string().contains("task history mismatch")); + } + + #[test] + fn test_binance_rejects_removed_or_derived_parameters() { + for parameter in [ + "aggregationMode=2", + "weight=1", + "minSourceCount=1", + "minTotalWeight=1", + "maxStaleness=180000", + "observations=source-a:1:1:1", + "field=close", + "dataSourceLabel=binance", + "dataSourceId=0x01", + "continuous=true", + "round=1", + "resolvedAt=1", + "blockNumber=1", + ] { + let uri = format!("{}&{parameter}", binance_uri()); + let task = parse_oracle_uri(&uri).unwrap(); + let err = + PriceFeedSource::from_task_with_rpc(&task, 0, Some("https://fapi.binance.com")) + .unwrap_err(); + assert!( + err.to_string().contains("Unsupported Binance index kline parameter"), + "parameter {parameter}: {err}" + ); + } + } + + #[test] + fn test_binance_rejects_onchain_base_url() { + let uri = format!("{}&baseUrl=https://example.com", binance_uri()); + let task = parse_oracle_uri(&uri).unwrap(); + let err = PriceFeedSource::from_task_with_rpc(&task, 0, Some("https://fapi.binance.com")) + .unwrap_err(); + assert!(err.to_string().contains("validator-local")); + } + + #[test] + fn test_binance_rejects_unaligned_bucket() { + let uri = "gravity://3/2001/price_feed?provider=binance_index_kline_v1&pair=TSLAUSDT&interval=1m&bucketStartMs=1710000000001&decimals=8"; + let task = parse_oracle_uri(uri).unwrap(); + let err = PriceFeedSource::from_task_with_rpc(&task, 0, Some("https://fapi.binance.com")) + .unwrap_err(); + + assert!(err.to_string().contains("not aligned")); + } + + #[test] + fn test_binance_index_kline_price_from_response() { + let task = parse_oracle_uri(binance_uri()).unwrap(); + let source = + PriceFeedSource::from_task_with_rpc(&task, 0, Some("https://fapi.binance.com")) + .unwrap(); + let response = serde_json::json!([[ + 1710000000000u64, + "400.67293", + "400.67546", + "400.67293", + "400.67545", + "0", + 1710000059999u64, + "0", + 0, + "0", + "0", + "0" + ]]); + + let round = source.config.round_for_delivery_nonce(1).unwrap(); + let price = + binance_index_kline_price_from_response(&source.config, &round, &response).unwrap(); + assert_eq!(price, "40067545000".parse::().unwrap()); + } + + #[tokio::test] + async fn test_binance_poll_fetches_and_wraps_exact_closed_bucket() { + let (base_url, server) = serve_once( + r#"[[1710000000000,"400.67293","400.67546","400.67293","400.67545","0",1710000059999]]"#, + ); + let task = parse_oracle_uri(binance_uri()).unwrap(); + let source = PriceFeedSource::from_task_with_rpc(&task, 0, Some(&base_url)).unwrap(); + + let data = source.poll().await.unwrap(); + + assert_eq!(data.len(), 1); + assert_eq!(data[0].nonce, 1); + assert_eq!(data[0].source_position, 1_710_000_059_999); + let (nonce, position, payload) = + <(u128, U256, Bytes)>::abi_decode(&data[0].payload).unwrap(); + assert_eq!(nonce, 1); + assert_eq!(position, U256::from(1_710_000_059_999u64)); + let payload = PricePayloadSol::abi_decode(&payload).unwrap(); + assert_eq!(payload.feedId, U256::from(2001)); + assert_eq!(payload.roundId, 28_500_000); + assert_eq!(payload.resolvedAt, 1_710_000_059_999); + assert_eq!(payload.decimals, 8); + assert_eq!(payload.price, "40067545000".parse::().unwrap()); + assert_eq!(source.last_nonce().await, Some(1)); + assert_eq!(source.last_nonce_position().await, Some(1_710_000_059_999)); + + let request = server.join().unwrap(); + assert!(request.starts_with( + "GET /fapi/v1/indexPriceKlines?pair=TSLAUSDT&interval=1m&startTime=1710000000000&endTime=1710000059999&limit=1 HTTP/1.1\r\n" + )); + } + + #[test] + fn test_binance_index_kline_rejects_zero_price() { + let task = parse_oracle_uri(binance_uri()).unwrap(); + let source = + PriceFeedSource::from_task_with_rpc(&task, 0, Some("https://fapi.binance.com")) + .unwrap(); + let response = + serde_json::json!([[1710000000000u64, "0", "0", "0", "0", "0", 1710000059999u64]]); + + let round = source.config.round_for_delivery_nonce(1).unwrap(); + let err = + binance_index_kline_price_from_response(&source.config, &round, &response).unwrap_err(); + assert!(err.to_string().contains("must be positive")); + } + + #[test] + fn test_binance_index_kline_rejects_wrong_open_time() { + let task = parse_oracle_uri(binance_uri()).unwrap(); + let source = + PriceFeedSource::from_task_with_rpc(&task, 0, Some("https://fapi.binance.com")) + .unwrap(); + let response = serde_json::json!([[ + 1710000060000u64, + "400.67293", + "400.67546", + "400.67293", + "400.67545", + "0", + 1710000119999u64 + ]]); + + let round = source.config.round_for_delivery_nonce(1).unwrap(); + let err = + binance_index_kline_price_from_response(&source.config, &round, &response).unwrap_err(); + assert!(err.to_string().contains("openTime mismatch")); + } + + #[test] + fn test_price_payload_has_single_price() { + let encoded = encode_price_payload( + 2001, + 28_500_000, + 1_710_000_059_999, + 8, + "40067545000".parse::().unwrap(), + ); + let payload = PricePayloadSol::abi_decode(&encoded).unwrap(); + + assert_eq!(payload.feedId, U256::from(2001)); + assert_eq!(payload.roundId, 28_500_000); + assert_eq!(payload.resolvedAt, 1_710_000_059_999); + assert_eq!(payload.decimals, 8); + assert_eq!(payload.price, "40067545000".parse::().unwrap()); + } + + #[test] + fn test_binance_response_chunk_limit_is_enforced_before_buffer_growth() { + let mut body = vec![0; MAX_BINANCE_RESPONSE_BYTES - 1]; + append_binance_response_chunk(&mut body, &[1]).unwrap(); + assert_eq!(body.len(), MAX_BINANCE_RESPONSE_BYTES); + + let err = append_binance_response_chunk(&mut body, &[2]).unwrap_err(); + assert!(err.to_string().contains("response exceeds")); + assert_eq!(body.len(), MAX_BINANCE_RESPONSE_BYTES); + } + + #[test] + fn test_fixed_decimal_requires_exact_configured_precision() { + assert_eq!(parse_fixed_decimal("195.380", 2).unwrap(), "19538".parse::().unwrap()); + assert_eq!(parse_fixed_decimal("195", 8).unwrap(), "19500000000".parse::().unwrap()); + assert!(parse_fixed_decimal("195.389", 2).unwrap_err().to_string().contains("precision")); + } +} diff --git a/crates/pipe-exec-layer-ext-v2/relayer/src/uri_parser.rs b/crates/pipe-exec-layer-ext-v2/relayer/src/uri_parser.rs index 3171b3e471..60f721775a 100644 --- a/crates/pipe-exec-layer-ext-v2/relayer/src/uri_parser.rs +++ b/crates/pipe-exec-layer-ext-v2/relayer/src/uri_parser.rs @@ -10,6 +10,9 @@ //! //! ### Examples //! - Blockchain events: `gravity://0/1/events?portal=0x283fC6...&fromBlock=9565280` +//! - Binance index kline price feed: +//! `gravity://3/2001/price_feed?provider=binance_index_kline_v1&pair=TSLAUSDT&interval=1m& +//! bucketStartMs=1710000000000&decimals=8` use alloy_primitives::Address; use anyhow::{anyhow, Result}; @@ -22,7 +25,7 @@ pub struct ParsedOracleTask { /// Original URI string pub uri: String, - /// Source type (0=BLOCKCHAIN) + /// Source type (`0=BLOCKCHAIN`, `3=PRICE_FEED`) pub source_type: u32, /// Source identifier (chain ID, etc.) @@ -54,9 +57,14 @@ impl ParsedOracleTask { } /// Check if this is a blockchain source - pub fn is_blockchain(&self) -> bool { + pub const fn is_blockchain(&self) -> bool { self.source_type == 0 } + + /// Check if this is a price feed source + pub const fn is_price_feed(&self) -> bool { + self.source_type == 3 + } } /// Parse a gravity:// URI into task configuration @@ -128,6 +136,18 @@ mod tests { assert!(task.portal_address().is_ok()); } + #[test] + fn test_parse_price_feed_uri() { + let uri = "gravity://3/2001/price_feed?provider=binance_index_kline_v1&pair=TSLAUSDT&interval=1m&bucketStartMs=1710000000000&decimals=8"; + let task = parse_oracle_uri(uri).unwrap(); + + assert_eq!(task.source_type, 3); + assert_eq!(task.source_id, 2001); + assert_eq!(task.task_type, "price_feed"); + assert!(task.is_price_feed()); + assert_eq!(task.params.get("decimals").map(String::as_str), Some("8")); + } + #[test] fn test_invalid_scheme() { let uri = "http://0/1/events";