diff --git a/Cargo.lock b/Cargo.lock index 9eaf8737..f3911bc7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -218,6 +218,7 @@ checksum = "de3b431b4e72cd8bd0ec7a50b4be18e73dab74de0dba180eef171055e5d5926e" dependencies = [ "alloy-rlp", "arbitrary", + "borsh", "bytes", "cfg-if", "const-hex", @@ -828,12 +829,6 @@ dependencies = [ "serde", ] -[[package]] -name = "ascii" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d92bec98840b8f03a5ff5413de5293bfcd8bf96467cf5452609f939ec6f5de16" - [[package]] name = "astral-tokio-tar" version = "0.6.1" @@ -1670,12 +1665,6 @@ dependencies = [ "windows-link", ] -[[package]] -name = "chunked_transfer" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" - [[package]] name = "cid" version = "0.11.1" @@ -5010,6 +4999,32 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "price-oracle" +version = "1.0.0" +dependencies = [ + "alloy-primitives", + "borsh", + "bytes", + "sov-evm", + "sov-modules-api", + "sov-rollup-interface", + "sov-test-utils", + "tracing", +] + +[[package]] +name = "price-oracle-ipc" +version = "0.3.0" +dependencies = [ + "alloy-primitives", + "borsh", + "socket2 0.5.10", + "thiserror 1.0.69", + "tokio", + "tracing", +] + [[package]] name = "primeorder" version = "0.13.6" @@ -5136,20 +5151,6 @@ dependencies = [ "syn 2.0.117", ] -[[package]] -name = "prometheus" -version = "0.13.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3d33c28a30771f7f96db69893f78b857f7450d7e0237e9c8fc6427a81bae7ed1" -dependencies = [ - "cfg-if", - "fnv", - "lazy_static", - "memchr", - "parking_lot", - "thiserror 1.0.69", -] - [[package]] name = "prometheus" version = "0.14.0" @@ -5164,20 +5165,6 @@ dependencies = [ "thiserror 2.0.18", ] -[[package]] -name = "prometheus_exporter" -version = "0.8.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "caf17cbebe0bfdf4f279ef84eeefe0d50468b0b7116f078acf41d456e48fe81a" -dependencies = [ - "ascii", - "lazy_static", - "log", - "prometheus 0.13.4", - "thiserror 1.0.69", - "tiny_http", -] - [[package]] name = "proptest" version = "1.11.0" @@ -5995,7 +5982,7 @@ source = "git+https://github.com/sovereign-labs/rockbound?rev=d8b270d399d0d355a1 dependencies = [ "anyhow", "parking_lot", - "prometheus 0.14.0", + "prometheus", "proptest", "proptest-derive 0.5.1", "quick_cache", @@ -6019,6 +6006,7 @@ dependencies = [ name = "rollup-starter" version = "0.3.0" dependencies = [ + "alloy-primitives", "anyhow", "async-trait", "axum", @@ -6026,7 +6014,9 @@ dependencies = [ "bytes", "clap", "futures", - "prometheus_exporter", + "hex", + "price-oracle", + "price-oracle-ipc", "rand 0.8.5", "rustls", "serde", @@ -6044,6 +6034,7 @@ dependencies = [ "sov-hyperlane-integration", "sov-kernels", "sov-ledger-apis", + "sov-metrics", "sov-mock-zkvm", "sov-modules-api", "sov-modules-rollup-blueprint", @@ -6059,6 +6050,7 @@ dependencies = [ "stf-starter", "tempfile", "tokio", + "toml", "tracing", "tracing-panic", "tracing-subscriber 0.3.23", @@ -6113,6 +6105,7 @@ dependencies = [ "ark-ff 0.3.0", "ark-ff 0.4.2", "ark-ff 0.5.0", + "borsh", "bytes", "fastrlp 0.3.1", "fastrlp 0.4.0", @@ -8312,6 +8305,8 @@ version = "0.3.0" dependencies = [ "anyhow", "borsh", + "bytes", + "price-oracle", "serde_json", "sov-address", "sov-bank", @@ -8323,6 +8318,7 @@ dependencies = [ "sov-evm", "sov-hyperlane-integration", "sov-kernels", + "sov-mock-da", "sov-mock-zkvm", "sov-modules-api", "sov-modules-stf-blueprint", @@ -8344,6 +8340,7 @@ dependencies = [ "borsh", "clap", "jsonrpsee", + "price-oracle", "serde", "sov-accounts", "sov-address", @@ -8709,19 +8706,6 @@ dependencies = [ "time-core", ] -[[package]] -name = "tiny_http" -version = "0.10.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c5f8734c6d6943ad6df6b588d228a87b4af184998bcffa268ceddf05c2055a8c" -dependencies = [ - "ascii", - "chunked_transfer", - "log", - "time", - "url", -] - [[package]] name = "tinystr" version = "0.8.2" diff --git a/Cargo.toml b/Cargo.toml index 74fdccce..0be7cda0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,8 @@ members = [ "crates/stf", "crates/stf/stf-declaration", "crates/utils/node-discovery", + "crates/precompiles/price-oracle", + "crates/shared/price-oracle-ipc", "examples/value-setter", ] default-members = ["crates/rollup", "crates/utils/node-discovery"] @@ -16,6 +18,7 @@ version = "0.3.0" edition = "2021" license = "Sovereign Permissionless Commercial License" authors = ["Sovereign Labs "] +repository = "https://github.com/Sovereign-Labs/rollup-starter" publish = false rust-version = "1.93" @@ -71,6 +74,8 @@ sov-metrics = { git = "https://github.com/Sovereign-Labs/sovereign-sdk.git", rev stf-starter = { path = "./crates/stf", default-features = false } stf-starter-declaration = { path = "./crates/stf/stf-declaration", default-features = false } node-discovery = { path = "./crates/utils/node-discovery", default-features = false } +price-oracle = { path = "./crates/precompiles/price-oracle", default-features = false } +price-oracle-ipc = { path = "./crates/shared/price-oracle-ipc", default-features = false } value-setter = { path = "./examples/value-setter" } strum = { version = "0.28", features = ["derive"] } @@ -83,7 +88,6 @@ axum = { version = "0.8.8", default-features = false } serde = { version = "1.0.192", features = ["derive", "rc"] } serde_json = { version = "1.0" } -prometheus_exporter = "0.8.5" sha2 = { version = "=0.10.9", default-features = false } schemars = { version = "1.2", features = ["derive"] } @@ -91,6 +95,7 @@ anyhow = { version = "1.0.95" } clap = { version = "4.4.10", features = ["derive"] } async-trait = "0.1.71" borsh = { version = "1", features = ["rc", "bytes"] } +bytes = { version = "1", default-features = false } tracing = { version = "0.1.40", default-features = false } tracing-subscriber = { version = "0.3.23", features = ["env-filter"] } tracing-appender = "0.2" @@ -98,6 +103,7 @@ tokio = { version = "1", features = ["full"] } lazy_static = "1.5.0" tempfile = "3.20" jsonrpsee = { version = "0.26", features = ["jsonrpsee-types"] } +toml = { version = "0.8", default-features = false, features = ["parse"] } # [patch."https://github.com/paradigmxyz/reth"] # reth-rpc-eth-types = { path = "/home/gentoo/work/sovereign/reth/crates/rpc/rpc-eth-types" } diff --git a/configs/celestia/oracle.toml b/configs/celestia/oracle.toml new file mode 100644 index 00000000..207d6247 --- /dev/null +++ b/configs/celestia/oracle.toml @@ -0,0 +1,18 @@ +[oracle] +enabled = true +require_sources = false + +[[oracle.source]] +name = "chainlink-1" +socket_address = "127.0.0.1:9801" +transport = "tcp" + +[[oracle.source]] +name = "chainlink-2" +socket_address = "127.0.0.1:9802" +transport = "tcp" + +[[oracle.source]] +name = "chainlink-3" +socket_address = "127.0.0.1:9803" +transport = "tcp" diff --git a/crates/precompiles/price-oracle/Cargo.toml b/crates/precompiles/price-oracle/Cargo.toml new file mode 100644 index 00000000..1ceae039 --- /dev/null +++ b/crates/precompiles/price-oracle/Cargo.toml @@ -0,0 +1,27 @@ +[package] +description = "Relay price oracle EVM precompile for the Relay Chain" +edition = { workspace = true } +license = { workspace = true } +name = "price-oracle" +publish = false +repository = { workspace = true } +rust-version = { workspace = true } +version = "1.0.0" + +[dependencies] +alloy-primitives = { workspace = true, features = ["borsh"] } +borsh = { workspace = true } +bytes = { workspace = true } +sov-evm = { workspace = true } +sov-modules-api = { workspace = true } +tracing = { workspace = true, optional = true } + +[dev-dependencies] +sov-evm = { workspace = true, features = ["native"] } +sov-modules-api = { workspace = true, features = ["native"] } +sov-rollup-interface = { workspace = true } +sov-test-utils = { workspace = true } + +[features] +default = [] +native = ["sov-evm/native", "sov-modules-api/native", "dep:tracing"] diff --git a/crates/precompiles/price-oracle/src/lib.rs b/crates/precompiles/price-oracle/src/lib.rs new file mode 100644 index 00000000..752f3e2f --- /dev/null +++ b/crates/precompiles/price-oracle/src/lib.rs @@ -0,0 +1,13 @@ +pub mod precompile; +#[cfg(feature = "native")] +pub mod sequencing; +pub mod types; + +pub use alloy_primitives::B256; +pub use precompile::{ + decode_feed_request, PriceOraclePrecompile, PRICE_ORACLE_PRECOMPILE_ADDRESS, + PRICE_ORACLE_PRECOMPILE_BASE_GAS, PRICE_ORACLE_PRECOMPILE_WORD_GAS, +}; +#[cfg(feature = "native")] +pub use sequencing::prune_unused; +pub use types::{FeedKey, PriceReports, UsedFeedKeys}; diff --git a/crates/precompiles/price-oracle/src/precompile.rs b/crates/precompiles/price-oracle/src/precompile.rs new file mode 100644 index 00000000..a8680996 --- /dev/null +++ b/crates/precompiles/price-oracle/src/precompile.rs @@ -0,0 +1,134 @@ +use std::marker::PhantomData; + +use alloy_primitives::{Address, Bytes}; +use borsh::BorshDeserialize; +use sov_evm::precompiles::{ + EvmPrecompile, EvmPrecompileEnv, PrecompileError, PrecompileOutput, PrecompileResult, +}; +use sov_modules_api::{Spec, TxState}; + +use crate::types::{FeedKey, PriceReports}; + +/// Precompile address 0x0000000000000000000000000000000000010002. +pub const PRICE_ORACLE_PRECOMPILE_ADDRESS: Address = Address::new([ + 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x01, 0x00, 0x02, +]); + +pub const PRICE_ORACLE_PRECOMPILE_BASE_GAS: u64 = 3_000; +pub const PRICE_ORACLE_PRECOMPILE_WORD_GAS: u64 = 16; + +#[derive(Clone, Default)] +pub struct PriceOraclePrecompile(PhantomData); + +impl EvmPrecompile for PriceOraclePrecompile { + const ADDRESS: Address = PRICE_ORACLE_PRECOMPILE_ADDRESS; + + fn execute>( + &self, + input: &[u8], + gas_limit: u64, + env: &mut EvmPrecompileEnv<'_, S, ST>, + ) -> PrecompileResult { + if PRICE_ORACLE_PRECOMPILE_BASE_GAS > gas_limit { + return Err(PrecompileError::OutOfGas); + } + + let (provider_id, feed_id) = decode_feed_request(input)?; + + let context = env + .sov_context + .ok_or_else(|| PrecompileError::State("missing transaction context".to_string()))?; + let sequencing_data = context.sequencing_data().as_ref().ok_or_else(|| { + PrecompileError::State("no sequencing data attached to transaction".to_string()) + })?; + let reports = PriceReports::try_from_slice(sequencing_data).map_err(|err| { + PrecompileError::State(format!("could not decode sequencing data: {err}")) + })?; + + let payload = reports + .get(&FeedKey::new(provider_id, feed_id)) + .ok_or_else(|| { + PrecompileError::InvalidInput(format!( + "no price report for provider {provider_id} feed {feed_id}" + )) + })?; + + // Record the feed before the gas check. The payload length affects gas, + // so a feed read here must be kept even if the call then runs out of gas, + // otherwise replay from the DA layer would diverge. + #[cfg(feature = "native")] + crate::sequencing::record_used_feed_key(context, FeedKey::new(provider_id, feed_id)) + .map_err(|err| { + PrecompileError::State(format!("could not record used feed key: {err}")) + })?; + + let words = payload.len().div_ceil(32) as u64; + let gas_used = PRICE_ORACLE_PRECOMPILE_BASE_GAS + PRICE_ORACLE_PRECOMPILE_WORD_GAS * words; + if gas_used > gas_limit { + return Err(PrecompileError::OutOfGas); + } + + Ok(PrecompileOutput { + gas_used, + bytes: Bytes::from(payload.clone()), + }) + } +} + +pub fn decode_feed_request( + input: &[u8], +) -> Result<(alloy_primitives::B256, alloy_primitives::B256), PrecompileError> { + if input.len() != 64 { + return Err(PrecompileError::InvalidInput(format!( + "expected 64 byte input with provider id and feed id, got {}", + input.len() + ))); + } + let provider_id = alloy_primitives::B256::from_slice(&input[0..32]); + let feed_id = alloy_primitives::B256::from_slice(&input[32..64]); + Ok((provider_id, feed_id)) +} + +#[cfg(test)] +mod tests { + use std::sync::LazyLock; + + use alloy_primitives::{keccak256, B256}; + + use super::*; + + static PROVIDER_ID: LazyLock = LazyLock::new(|| keccak256("chainlink")); + + fn feed_id(suffix: u8) -> B256 { + let mut bytes = [0u8; 32]; + bytes[1] = 0x03; + bytes[31] = suffix; + B256::from(bytes) + } + + fn request(provider_id: B256, feed_id: B256) -> Vec { + let mut bytes = Vec::with_capacity(64); + bytes.extend_from_slice(provider_id.as_slice()); + bytes.extend_from_slice(feed_id.as_slice()); + bytes + } + + #[test] + fn decode_splits_provider_and_feed() { + let (provider, feed) = + decode_feed_request(&request(*PROVIDER_ID, feed_id(1))).expect("decode"); + assert_eq!(provider, *PROVIDER_ID); + assert_eq!(feed, feed_id(1)); + } + + #[test] + fn decode_rejects_non_64_byte_input() { + for len in [0usize, 3, 32, 63, 65, 68, 96] { + let err = decode_feed_request(&vec![0u8; len]).unwrap_err(); + assert!( + matches!(err, PrecompileError::InvalidInput(_)), + "input length {len} should be rejected as invalid" + ); + } + } +} diff --git a/crates/precompiles/price-oracle/src/sequencing.rs b/crates/precompiles/price-oracle/src/sequencing.rs new file mode 100644 index 00000000..c5056b60 --- /dev/null +++ b/crates/precompiles/price-oracle/src/sequencing.rs @@ -0,0 +1,99 @@ +//! Integration with the per-transaction sequencing data and scratchpad. + +use borsh::BorshDeserialize; +use bytes::Bytes; +use sov_modules_api::{Context, Spec}; + +use crate::types::{FeedKey, PriceReports, UsedFeedKeys}; + +pub fn prune_unused(mut data: PriceReports, scratchpad: Option) -> PriceReports { + let Some(scratchpad) = scratchpad else { + // No scratchpad means the precompile recorded no reads, so every report is unused and safe to drop. + return PriceReports::default(); + }; + let used = match UsedFeedKeys::try_from_slice(&scratchpad) { + Ok(used) => used, + Err(err) => { + // A malformed scratchpad hides which feeds were read, we keep all reports because a superset is safer. + tracing::error!(%err, "sequencing scratchpad is malformed, publishing full sequencing data"); + return data; + } + }; + data.retain_keys(&used.0); + data +} + +pub(crate) fn record_used_feed_key( + context: &Context, + key: FeedKey, +) -> Result<(), std::io::Error> { + context.sequencing_scratchpad().with_value(|slot| { + let mut used = match slot.as_deref() { + Some(bytes) => UsedFeedKeys::try_from_slice(bytes)?, + None => UsedFeedKeys::default(), + }; + if !used.0.insert(key) { + return Ok(()); + } + let bytes = borsh::to_vec(&used).expect("in-memory borsh serialization is infallible"); + *slot = Some(Bytes::from(bytes)); + Ok(()) + }) +} + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use alloy_primitives::B256; + + use super::*; + + fn feed_key(suffix: u8) -> FeedKey { + FeedKey::new(B256::repeat_byte(0xc1), B256::repeat_byte(suffix)) + } + + fn prices(keys: &[FeedKey]) -> PriceReports { + PriceReports( + keys.iter() + .map(|k| (*k, Bytes::from_static(b"payload"))) + .collect::>(), + ) + } + + fn scratchpad(used: &[FeedKey]) -> Bytes { + Bytes::from(borsh::to_vec(&UsedFeedKeys(used.iter().copied().collect())).unwrap()) + } + + #[test] + fn none_scratchpad_drops_all_feeds() { + let pruned = prune_unused(prices(&[feed_key(1), feed_key(2)]), None); + assert_eq!(pruned, PriceReports::default()); + } + + #[test] + fn malformed_scratchpad_keeps_full_data() { + let data = prices(&[feed_key(1), feed_key(2)]); + let pruned = prune_unused(data.clone(), Some(Bytes::from(vec![0xff, 0xff, 0xff]))); + assert_eq!(pruned, data); + } + + #[test] + fn prunes_to_used_subset() { + let kept = feed_key(1); + let dropped = feed_key(2); + let pruned = prune_unused( + prices(&[kept, dropped, feed_key(3)]), + Some(scratchpad(&[kept])), + ); + assert_eq!(pruned.0.len(), 1); + assert_eq!(pruned.get(&kept), Some(&Bytes::from_static(b"payload"))); + assert_eq!(pruned.get(&dropped), None); + } + + #[test] + fn empty_used_set_drops_all_feeds() { + let pruned = prune_unused(prices(&[feed_key(1), feed_key(2)]), Some(scratchpad(&[]))); + assert_eq!(pruned, PriceReports::default()); + } +} diff --git a/crates/precompiles/price-oracle/src/types.rs b/crates/precompiles/price-oracle/src/types.rs new file mode 100644 index 00000000..2bcda0bc --- /dev/null +++ b/crates/precompiles/price-oracle/src/types.rs @@ -0,0 +1,52 @@ +use std::collections::{BTreeMap, BTreeSet}; + +use alloy_primitives::B256; +use borsh::{BorshDeserialize, BorshSerialize}; +use bytes::Bytes; +use sov_modules_api::capabilities::SequencingDataTrait; +use sov_modules_api::HDTimestamp; + +#[derive( + Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, BorshSerialize, BorshDeserialize, +)] +pub struct FeedKey { + pub provider_id: B256, + pub feed_id: B256, +} + +impl FeedKey { + pub const fn new(provider_id: B256, feed_id: B256) -> Self { + Self { + provider_id, + feed_id, + } + } +} + +/// Latest opaque signed report per feed. +/// Reports are bytes so cloning the map for each transaction only bumps refcounts +/// instead of copying large blobs. +#[derive(Clone, Debug, Default, PartialEq, Eq, BorshSerialize, BorshDeserialize)] +pub struct PriceReports(pub BTreeMap); + +impl PriceReports { + pub fn get(&self, key: &FeedKey) -> Option<&Bytes> { + self.0.get(key) + } + + pub fn retain_keys(&mut self, keep: &BTreeSet) { + self.0.retain(|k, _| keep.contains(k)); + } +} + +impl SequencingDataTrait for PriceReports { + // Carries no sequencer timestamp hence this is always None. + fn get_maybe_timestamp(self) -> Option { + None + } +} + +/// Feed keys a transaction actually read. +/// Recorded in the sequencing scratchpad during execution. +#[derive(Clone, Debug, Default, PartialEq, Eq, BorshSerialize, BorshDeserialize)] +pub struct UsedFeedKeys(pub BTreeSet); diff --git a/crates/precompiles/price-oracle/tests/execute.rs b/crates/precompiles/price-oracle/tests/execute.rs new file mode 100644 index 00000000..47e2b637 --- /dev/null +++ b/crates/precompiles/price-oracle/tests/execute.rs @@ -0,0 +1,269 @@ +#![cfg(feature = "native")] + +use std::sync::LazyLock; + +use alloy_primitives::{keccak256, B256}; +use borsh::BorshDeserialize; +use bytes::Bytes; +use price_oracle::{ + FeedKey, PriceOraclePrecompile, PriceReports, UsedFeedKeys, PRICE_ORACLE_PRECOMPILE_BASE_GAS, + PRICE_ORACLE_PRECOMPILE_WORD_GAS, +}; +use sov_evm::precompiles::{EvmPrecompile, EvmPrecompileEnv, PrecompileError, PrecompileResult}; +use sov_modules_api::capabilities::mocks::MockKernel; +use sov_modules_api::transaction::Credentials; +use sov_modules_api::{Context, DaSpec, ExecutionContext, SequencerType, Spec, StateCheckpoint}; +use sov_test_utils::storage::SimpleStorageManager; +use sov_test_utils::{TestSpec, TestStorageSpec}; + +type S = TestSpec; + +const GAS_LIMIT: u64 = 1_000_000; + +static PROVIDER_ID: LazyLock = LazyLock::new(|| keccak256("chainlink")); + +fn feed_id(suffix: u8) -> B256 { + let mut bytes = [0u8; 32]; + bytes[1] = 0x03; + bytes[31] = suffix; + B256::from(bytes) +} + +fn request(provider_id: B256, feed_id: B256) -> Vec { + let mut bytes = Vec::with_capacity(64); + bytes.extend_from_slice(provider_id.as_slice()); + bytes.extend_from_slice(feed_id.as_slice()); + bytes +} + +fn reports_with(entries: &[(FeedKey, &[u8])]) -> PriceReports { + PriceReports( + entries + .iter() + .map(|(k, p)| (*k, Bytes::copy_from_slice(p))) + .collect(), + ) +} + +fn expected_gas(payload_len: usize) -> u64 { + PRICE_ORACLE_PRECOMPILE_BASE_GAS + + PRICE_ORACLE_PRECOMPILE_WORD_GAS * payload_len.div_ceil(32) as u64 +} + +fn context(reports: Option<&PriceReports>) -> Context { + let sequencing_data = reports.map(|u| { + sov_rollup_interface::Bytes::from(borsh::to_vec(u).expect("encode sequencing data")) + }); + context_with_raw_sequencing_data(sequencing_data) +} + +fn context_with_raw_sequencing_data( + sequencing_data: Option, +) -> Context { + let addr = ::Address::from([7u8; 28]); + let da_addr = <::Da as DaSpec>::Address::from([9u8; 32]); + Context::::new( + addr, + Credentials::default(), + addr, + da_addr, + sequencing_data, + ExecutionContext::Node, + SequencerType::Preferred, + ) +} + +fn run(sov_context: Option<&Context>, input: &[u8], gas_limit: u64) -> PrecompileResult { + let storage_manager = SimpleStorageManager::::new(); + let storage = storage_manager.create_storage(); + let mut state = + StateCheckpoint::::new(storage, &MockKernel::::default()).to_working_set_unmetered(); + let mut env = EvmPrecompileEnv { + state: &mut state, + sov_context, + }; + PriceOraclePrecompile::::default().execute(input, gas_limit, &mut env) +} + +fn used_feed_keys(ctx: &Context) -> Vec { + ctx.sequencing_scratchpad().with_value(|scratchpad| { + let bytes = scratchpad + .as_deref() + .expect("scratchpad should be populated"); + UsedFeedKeys::try_from_slice(bytes) + .expect("decode used feed keys") + .0 + .into_iter() + .collect() + }) +} + +#[test] +fn present_feed_returns_payload_and_gas() { + let payload = b"signed-update-bytes".to_vec(); + let reports = reports_with(&[(FeedKey::new(*PROVIDER_ID, feed_id(1)), &payload)]); + let ctx = context(Some(&reports)); + + let output = run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT) + .expect("present feed should resolve"); + + assert_eq!(output.bytes.as_ref(), payload.as_slice()); + assert_eq!(output.gas_used, expected_gas(payload.len())); +} + +#[test] +fn missing_feed_is_invalid_input() { + let reports = reports_with(&[(FeedKey::new(*PROVIDER_ID, feed_id(1)), b"present")]); + let ctx = context(Some(&reports)); + + let err = run(Some(&ctx), &request(*PROVIDER_ID, feed_id(2)), GAS_LIMIT).unwrap_err(); + assert!(matches!(err, PrecompileError::InvalidInput(_))); +} + +#[test] +fn empty_reports_is_invalid_input() { + let ctx = context(Some(&reports_with(&[]))); + + let err = run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT).unwrap_err(); + assert!(matches!(err, PrecompileError::InvalidInput(_))); +} + +#[test] +fn wrong_length_request_is_invalid_input() { + let reports = reports_with(&[(FeedKey::new(*PROVIDER_ID, feed_id(1)), b"present")]); + let ctx = context(Some(&reports)); + + let err = run(Some(&ctx), &[0u8; 10], GAS_LIMIT).unwrap_err(); + assert!(matches!(err, PrecompileError::InvalidInput(_))); +} + +#[test] +fn missing_context_is_state_error() { + let err = run(None, &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT).unwrap_err(); + assert!(matches!(err, PrecompileError::State(_))); +} + +#[test] +fn missing_sequencing_data_is_state_error() { + let ctx = context(None); + + let err = run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT).unwrap_err(); + assert!(matches!(err, PrecompileError::State(_))); +} + +#[test] +fn undecodable_sequencing_data_is_state_error() { + let ctx = + context_with_raw_sequencing_data(Some(sov_rollup_interface::Bytes::from(vec![0xff; 8]))); + + let err = run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT).unwrap_err(); + assert!(matches!(err, PrecompileError::State(_))); +} + +#[test] +fn insufficient_base_gas_is_out_of_gas() { + let reports = reports_with(&[(FeedKey::new(*PROVIDER_ID, feed_id(1)), b"present")]); + let ctx = context(Some(&reports)); + + let err = run( + Some(&ctx), + &request(*PROVIDER_ID, feed_id(1)), + PRICE_ORACLE_PRECOMPILE_BASE_GAS - 1, + ) + .unwrap_err(); + assert!(matches!(err, PrecompileError::OutOfGas)); +} + +#[test] +fn payload_gas_over_limit_is_out_of_gas() { + let payload = vec![0xab; 64]; + let reports = reports_with(&[(FeedKey::new(*PROVIDER_ID, feed_id(1)), &payload)]); + let ctx = context(Some(&reports)); + + let one_word_short = expected_gas(payload.len()) - PRICE_ORACLE_PRECOMPILE_WORD_GAS; + let err = run( + Some(&ctx), + &request(*PROVIDER_ID, feed_id(1)), + one_word_short, + ) + .unwrap_err(); + assert!(matches!(err, PrecompileError::OutOfGas)); +} + +#[test] +fn exact_gas_limit_succeeds() { + let payload = b"signed-update-bytes".to_vec(); + let reports = reports_with(&[(FeedKey::new(*PROVIDER_ID, feed_id(1)), &payload)]); + let ctx = context(Some(&reports)); + + let exact = expected_gas(payload.len()); + let output = run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), exact) + .expect("exact gas limit should resolve"); + assert_eq!(output.gas_used, exact); +} + +#[test] +fn charges_gas_per_word() { + for (len, words) in [(0usize, 0u64), (32, 1), (33, 2)] { + let payload = vec![0xab; len]; + let reports = reports_with(&[(FeedKey::new(*PROVIDER_ID, feed_id(1)), &payload)]); + let ctx = context(Some(&reports)); + + let output = run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT) + .expect("present feed should resolve"); + + assert_eq!( + output.gas_used, + PRICE_ORACLE_PRECOMPILE_BASE_GAS + PRICE_ORACLE_PRECOMPILE_WORD_GAS * words, + "payload of {len} bytes should cost {words} word(s)" + ); + } +} + +#[test] +fn records_used_feed_key() { + let key = FeedKey::new(*PROVIDER_ID, feed_id(1)); + let reports = reports_with(&[(key, b"present")]); + let ctx = context(Some(&reports)); + + run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT) + .expect("present feed should resolve"); + + assert_eq!(used_feed_keys(&ctx), vec![key]); +} + +#[test] +fn accumulates_used_feed_keys() { + let key1 = FeedKey::new(*PROVIDER_ID, feed_id(1)); + let key2 = FeedKey::new(*PROVIDER_ID, feed_id(2)); + let reports = reports_with(&[(key1, b"first"), (key2, b"second")]); + let ctx = context(Some(&reports)); + + run(Some(&ctx), &request(*PROVIDER_ID, feed_id(1)), GAS_LIMIT) + .expect("first feed should resolve"); + run(Some(&ctx), &request(*PROVIDER_ID, feed_id(2)), GAS_LIMIT) + .expect("second feed should resolve"); + + assert_eq!(used_feed_keys(&ctx), vec![key1, key2]); +} + +#[test] +fn records_used_feed_key_even_when_out_of_gas() { + // A feed read here still runs out of gas, but it must be recorded so the + // sequencer keeps it and replay from the DA layer reaches the same outcome. + let payload = vec![0xab; 64]; + let key = FeedKey::new(*PROVIDER_ID, feed_id(1)); + let reports = reports_with(&[(key, &payload)]); + let ctx = context(Some(&reports)); + + let one_word_short = expected_gas(payload.len()) - PRICE_ORACLE_PRECOMPILE_WORD_GAS; + let err = run( + Some(&ctx), + &request(*PROVIDER_ID, feed_id(1)), + one_word_short, + ) + .unwrap_err(); + assert!(matches!(err, PrecompileError::OutOfGas)); + + assert_eq!(used_feed_keys(&ctx), vec![key]); +} diff --git a/crates/precompiles/price-oracle/tests/sequencing_data.rs b/crates/precompiles/price-oracle/tests/sequencing_data.rs new file mode 100644 index 00000000..73a30393 --- /dev/null +++ b/crates/precompiles/price-oracle/tests/sequencing_data.rs @@ -0,0 +1,78 @@ +use std::collections::BTreeMap; +use std::collections::BTreeSet; +use std::sync::LazyLock; + +use alloy_primitives::{keccak256, B256}; +use borsh::BorshDeserialize; +use bytes::Bytes; +use price_oracle::{FeedKey, PriceReports}; + +static PROVIDER_ID: LazyLock = LazyLock::new(|| keccak256("chainlink")); + +fn feed_id(suffix: u8) -> B256 { + let mut bytes = [0u8; 32]; + bytes[1] = 0x03; + bytes[31] = suffix; + B256::from(bytes) +} + +fn reports_with(entries: &[(FeedKey, &[u8])]) -> PriceReports { + let map: BTreeMap = entries + .iter() + .map(|(k, p)| (*k, Bytes::copy_from_slice(p))) + .collect(); + PriceReports(map) +} + +#[test] +fn retain_keys_drops_unused_entries() { + let kept = FeedKey::new(*PROVIDER_ID, feed_id(1)); + let dropped = FeedKey::new(*PROVIDER_ID, feed_id(2)); + let mut reports = reports_with(&[(kept, b"keep-me"), (dropped, b"drop-me")]); + + reports.retain_keys(&[kept].into_iter().collect()); + + assert_eq!(reports.0.len(), 1); + assert!(reports.get(&kept).is_some()); + assert!(reports.get(&dropped).is_none()); +} + +#[test] +fn retain_keys_empty_set_clears_map() { + let key = FeedKey::new(*PROVIDER_ID, feed_id(1)); + let mut reports = reports_with(&[(key, b"payload")]); + + reports.retain_keys(&BTreeSet::new()); + assert!(reports.0.is_empty()); +} + +#[test] +fn retain_keys_ignores_unknown_keys() { + let present = FeedKey::new(*PROVIDER_ID, feed_id(1)); + let absent = FeedKey::new(*PROVIDER_ID, feed_id(99)); + let mut reports = reports_with(&[(present, b"present")]); + + reports.retain_keys(&[present, absent].into_iter().collect()); + + assert_eq!(reports.0.len(), 1); + assert!(reports.get(&present).is_some()); +} + +#[test] +fn borsh_round_trip_is_canonical() { + let mut entries = BTreeMap::new(); + entries.insert( + FeedKey::new(*PROVIDER_ID, feed_id(2)), + Bytes::from_static(b"second"), + ); + entries.insert( + FeedKey::new(*PROVIDER_ID, feed_id(1)), + Bytes::from_static(b"first"), + ); + let reports = PriceReports(entries); + + let bytes = borsh::to_vec(&reports).expect("borsh encode"); + let decoded = PriceReports::try_from_slice(&bytes).expect("borsh decode"); + assert_eq!(decoded, reports); + assert_eq!(borsh::to_vec(&decoded).expect("borsh re-encode"), bytes); +} diff --git a/crates/rollup/Cargo.toml b/crates/rollup/Cargo.toml index f023ce5f..f7a14363 100644 --- a/crates/rollup/Cargo.toml +++ b/crates/rollup/Cargo.toml @@ -46,10 +46,17 @@ tracing = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true, features = ["rt", "rt-multi-thread", "macros"] } sha2 = { workspace = true } -prometheus_exporter = { workspace = true } +sov-metrics = { workspace = true } stf-starter = { workspace = true, default-features = false, features = ["native"] } +price-oracle-ipc = { workspace = true } +price-oracle = { workspace = true } +alloy-primitives = { workspace = true } +toml = { workspace = true } +futures = { version = "0.3", default-features = false } +hex = { version = "0.4", default-features = false, features = ["alloc"] } + # binary dependencies tracing-subscriber = { workspace = true } tracing-panic = "0.1.2" diff --git a/crates/rollup/src/bin/rollup.rs b/crates/rollup/src/bin/rollup.rs index 045004c3..6facab2f 100644 --- a/crates/rollup/src/bin/rollup.rs +++ b/crates/rollup/src/bin/rollup.rs @@ -3,6 +3,7 @@ use anyhow::Context; use clap::Parser; use rollup_starter::da::DaService; +use rollup_starter::oracle; use rollup_starter::rollup::StarterRollup; use sov_modules_rollup_blueprint::logging::initialize_logging; use sov_modules_rollup_blueprint::FullNodeBlueprint; @@ -39,10 +40,6 @@ struct Args { #[arg(long, default_value = default_genesis_path().into_os_string())] genesis_path: PathBuf, - // UDP port on 127.0.0.1 where Telegraf service suppose to listen. - #[arg(long, default_value_t = 9845)] - metrics: u64, - /// Start the rollup at a given height. #[arg(long, default_value = None)] start_at_rollup_height: Option, @@ -55,6 +52,10 @@ struct Args { /// it issues a fresh outer proof that replaces any previous one. #[arg(long, default_value_t = false)] start_fresh_outer_proof_on_resync: bool, + + /// The path to the price oracle config. + #[arg(long, default_value = None)] + oracle_config_path: Option, } #[tokio::main] @@ -68,14 +69,13 @@ async fn main() { let _guard = initialize_logging(); - let metrics_port = args.metrics; - let address = format!("127.0.0.1:{metrics_port}"); - prometheus_exporter::start(address.parse().unwrap()) - .expect("Could not start prometheus server"); - let prover_config = parse_prover_config().expect("Malformed prover_config"); tracing::info!(?prover_config, "Running demo rollup with prover config"); + spawn_oracle_clients(&args) + .await + .expect("failed to start oracle price-feed clients"); + let rollup = new_rollup( args.genesis_path, args.rollup_config_path, @@ -110,6 +110,13 @@ fn parse_prover_config() -> anyhow::Result { } } +async fn spawn_oracle_clients(args: &Args) -> anyhow::Result<()> { + let path = + oracle::resolve_config_path(args.oracle_config_path.clone(), &args.rollup_config_path); + let config = oracle::load_or_create_config(&path)?; + oracle::spawn_clients(config, path).await +} + async fn new_rollup( genesis_path: PathBuf, rollup_config_path: PathBuf, diff --git a/crates/rollup/src/lib.rs b/crates/rollup/src/lib.rs index 82a604b0..0e013055 100644 --- a/crates/rollup/src/lib.rs +++ b/crates/rollup/src/lib.rs @@ -1,3 +1,4 @@ pub mod da; +pub mod oracle; pub mod rollup; pub mod zkvm; diff --git a/crates/rollup/src/oracle.rs b/crates/rollup/src/oracle.rs new file mode 100644 index 00000000..5f1e4382 --- /dev/null +++ b/crates/rollup/src/oracle.rs @@ -0,0 +1,1027 @@ +use std::collections::{HashMap, HashSet}; +use std::fs; +use std::future; +use std::io::ErrorKind; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use anyhow::{anyhow, Context}; +use price_oracle_ipc::{ + connect, read_frame_with_timeout, Backoff, IpcError, OracleFrame, OracleStream, B256, + PROTOCOL_VERSION, +}; +use serde::Deserialize; +use tokio::signal::unix::{signal, SignalKind}; +use tokio::sync::{mpsc, watch}; +use tokio::task::JoinHandle; +use tracing::{debug, error, info, trace, warn}; + +use stf_starter::prices; + +const ORACLE_CONFIG_FILE: &str = "oracle.toml"; +const ORACLE_CONFIG_CONTENT: &str = "[oracle]\nenabled = false\n"; +const DEADLINE_HEARTBEAT_MULTIPLIER: u32 = 3; +const BOOTSTRAP_DEADLINE: Duration = Duration::from_secs(30); +const SUPERVISOR_GUARD_MIN: Duration = Duration::from_secs(1); +const SUPERVISOR_GUARD_MAX: Duration = Duration::from_secs(30); +const HEALTHY_CONNECTION_THRESHOLD: Duration = Duration::from_secs(10); +const STALENESS_WARN_SEC: u64 = 30; +const REQUIRE_SOURCES_TIMEOUT: Duration = Duration::from_secs(5); +const METRICS_REPORT_INTERVAL: Duration = Duration::from_secs(15); +const PROVIDER_FEEDS_MAX: usize = 512; +const HEARTBEAT_INTERVAL_MAX_SEC: u32 = 300; +const THROTTLE_WARN_SEC: u64 = 60; + +static LAST_DROPPED_WARN_AT: AtomicU64 = AtomicU64::new(0); +static LAST_DIVERGENCE_WARN_AT: AtomicU64 = AtomicU64::new(0); + +fn warn_allowed(last: &AtomicU64) -> bool { + let now = now_unix(); + let prev = last.load(Ordering::Relaxed); + now.saturating_sub(prev) >= THROTTLE_WARN_SEC + && last + .compare_exchange(prev, now, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() +} + +#[derive(Debug, Clone, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OracleConfigFile { + pub oracle: OracleConfig, +} + +#[derive(Debug, Clone, Default, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct OracleConfig { + #[serde(default)] + pub enabled: bool, + #[serde(default)] + pub require_sources: bool, + #[serde(default, rename = "source")] + pub sources: Vec, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Transport { + #[default] + Tcp, +} + +#[derive(Debug, Clone, PartialEq, Eq, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SourceConfig { + pub name: String, + #[serde(default)] + pub transport: Transport, + #[serde(default)] + pub socket_address: Option, +} + +impl SourceConfig { + fn address(&self) -> anyhow::Result { + match self.transport { + Transport::Tcp => self.socket_address.clone().ok_or_else(|| { + anyhow!( + "oracle source '{}': transport = \"tcp\" requires socket_address", + self.name + ) + }), + } + } +} + +pub fn resolve_config_path(explicit: Option, rollup_config_path: &Path) -> PathBuf { + match explicit { + Some(path) => path, + None => { + let dir = rollup_config_path + .parent() + .unwrap_or_else(|| Path::new(".")); + dir.join(ORACLE_CONFIG_FILE) + } + } +} + +fn validate(config: &OracleConfig) -> anyhow::Result<()> { + let mut seen = HashSet::new(); + for source in &config.sources { + if !seen.insert(source.name.as_str()) { + return Err(anyhow!("duplicate oracle source name '{}'", source.name)); + } + source.address()?; + } + Ok(()) +} + +pub fn load_config(path: &Path) -> anyhow::Result> { + match fs::read_to_string(path) { + Ok(contents) => { + let file: OracleConfigFile = toml::from_str(&contents) + .with_context(|| format!("failed to parse oracle config {}", path.display()))?; + validate(&file.oracle) + .with_context(|| format!("invalid oracle config {}", path.display()))?; + Ok(Some(file.oracle)) + } + Err(e) if e.kind() == ErrorKind::NotFound => Ok(None), + Err(e) => Err(anyhow!( + "failed to read oracle config {}: {e}", + path.display() + )), + } +} + +pub fn load_or_create_config(path: &Path) -> anyhow::Result { + if let Some(config) = load_config(path)? { + return Ok(config); + } + match fs::write(path, ORACLE_CONFIG_CONTENT) { + Ok(()) => info!( + path = %path.display(), + "Oracle config not found, created default disabled config" + ), + Err(e) => error!( + path = %path.display(), + error = %e, + "Failed to create default oracle config, proceeding with oracle disabled" + ), + } + Ok(OracleConfig::default()) +} + +struct SourceHandle { + config: SourceConfig, + cancel: watch::Sender<()>, + join: JoinHandle<()>, +} + +impl SourceHandle { + async fn shutdown(self) { + let _ = self.cancel.send(()); + let _ = self.join.await; + } +} + +pub async fn spawn_clients(config: OracleConfig, path: PathBuf) -> anyhow::Result<()> { + if !config.enabled { + info!("Price oracle disabled via config, not connecting to any source"); + tokio::spawn(ignore_sighup()); + return Ok(()); + } + + let startup_config = config.clone(); + + // When require_sources is set, each supervisor reports its name on first connect and we block startup until all report or time out. + let (ready_tx, ready_rx) = if config.require_sources { + let (tx, rx) = mpsc::unbounded_channel::(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + + let mut registry: HashMap = HashMap::new(); + for source in config.sources { + let name = source.name.clone(); + if let Some(handle) = spawn_source(source, ready_tx.clone()) { + registry.insert(name, handle); + } + } + drop(ready_tx); + + if let Some(ready_rx) = ready_rx { + let pending: HashSet = registry.keys().cloned().collect(); + await_sources_ready(pending, ready_rx, REQUIRE_SOURCES_TIMEOUT).await?; + info!("All required oracle sources connected"); + } + + tokio::spawn(metrics::run_reporter(METRICS_REPORT_INTERVAL)); + tokio::spawn(reload_manager(path, startup_config, registry)); + Ok(()) +} + +async fn await_sources_ready( + mut pending: HashSet, + mut ready_rx: mpsc::UnboundedReceiver, + timeout: Duration, +) -> anyhow::Result<()> { + let deadline = tokio::time::Instant::now() + timeout; + while !pending.is_empty() { + match tokio::time::timeout_at(deadline, ready_rx.recv()).await { + Ok(Some(name)) => { + pending.remove(&name); + } + Ok(None) | Err(_) => break, + } + } + if pending.is_empty() { + return Ok(()); + } + let mut unconnected: Vec = pending.into_iter().collect(); + unconnected.sort(); + Err(anyhow!( + "require_sources is set, but {} source(s) did not connect within {}s: {}", + unconnected.len(), + timeout.as_secs(), + unconnected.join(", ") + )) +} + +fn spawn_source( + source: SourceConfig, + ready: Option>, +) -> Option { + let address = match source.address() { + Ok(address) => address, + Err(e) => { + error!(source = %source.name, error = %e, "Skipping oracle source with invalid address"); + return None; + } + }; + info!(source = %source.name, address = %address, "Starting oracle source client"); + let (cancel_tx, cancel_rx) = watch::channel(()); + let join = tokio::spawn(supervise_source(source.clone(), address, ready, cancel_rx)); + Some(SourceHandle { + config: source, + cancel: cancel_tx, + join, + }) +} + +async fn ignore_sighup() { + let mut sighup = match signal(SignalKind::hangup()) { + Ok(sighup) => sighup, + Err(e) => { + error!(error = %e, "Failed to install SIGHUP handler"); + return; + } + }; + loop { + sighup.recv().await; + info!("Received SIGHUP, but the price oracle is disabled, ignoring reload"); + } +} + +async fn reload_manager( + path: PathBuf, + startup_config: OracleConfig, + mut registry: HashMap, +) { + let mut sighup = match signal(SignalKind::hangup()) { + Ok(sighup) => sighup, + Err(e) => { + error!(error = %e, "Failed to install SIGHUP handler, oracle config reload disabled"); + // Park forever so the running sources (held in `registry`) are not dropped. + future::pending::<()>().await; + return; + } + }; + + loop { + sighup.recv().await; + info!("Received SIGHUP, reloading oracle config"); + + let new_config = match load_config(&path) { + Ok(Some(config)) => config, + Ok(None) => { + warn!("Oracle config not found on reload, keeping current sources"); + continue; + } + Err(e) => { + error!(error = %format!("{e:#}"), "Oracle config reload failed, keeping current sources"); + continue; + } + }; + + if new_config.enabled != startup_config.enabled + || new_config.require_sources != startup_config.require_sources + { + warn!("Oracle top-level config is startup-only, ignoring changed values until restart"); + } + + apply_source_diff(&mut registry, new_config.sources).await; + } +} + +struct SourceDiff { + to_add: Vec, + to_restart: Vec, + to_remove: Vec, +} + +fn diff_sources( + current: &HashMap, + new_sources: Vec, +) -> SourceDiff { + let mut diff = SourceDiff { + to_add: Vec::new(), + to_restart: Vec::new(), + to_remove: Vec::new(), + }; + let mut new_names = HashSet::new(); + for source in new_sources { + new_names.insert(source.name.clone()); + match current.get(&source.name) { + None => diff.to_add.push(source), + Some(existing) if *existing != source => diff.to_restart.push(source), + Some(_) => {} + } + } + for name in current.keys() { + if !new_names.contains(name) { + diff.to_remove.push(name.clone()); + } + } + diff +} + +async fn apply_source_diff( + registry: &mut HashMap, + new_sources: Vec, +) { + let current: HashMap = registry + .iter() + .map(|(name, handle)| (name.clone(), handle.config.clone())) + .collect(); + let diff = diff_sources(¤t, new_sources); + let (added, restarted, removed) = ( + diff.to_add.len(), + diff.to_restart.len(), + diff.to_remove.len(), + ); + + for name in diff.to_remove { + if let Some(handle) = registry.remove(&name) { + info!(source = %name, "Removing oracle source (config reload)"); + handle.shutdown().await; + let evicted = prices::remove_source(&name); + if evicted > 0 { + info!(source = %name, evicted, "Evicted feeds for removed oracle source"); + } + } + } + for source in diff.to_restart { + let name = source.name.clone(); + if let Some(handle) = registry.remove(&name) { + info!(source = %name, "Restarting oracle source (config changed)"); + handle.shutdown().await; + } + if let Some(handle) = spawn_source(source, None) { + registry.insert(name, handle); + } + } + for source in diff.to_add { + let name = source.name.clone(); + info!(source = %name, "Adding oracle source (config reload)"); + if let Some(handle) = spawn_source(source, None) { + registry.insert(name, handle); + } + } + + debug!(added, restarted, removed, "Applied oracle config reload"); +} + +async fn supervise_source( + source: SourceConfig, + address: String, + ready: Option>, + mut cancel: watch::Receiver<()>, +) { + let mut guard = Backoff::new(SUPERVISOR_GUARD_MIN, SUPERVISOR_GUARD_MAX); + loop { + let started = Instant::now(); + let mut child = tokio::spawn(run_source(source.clone(), address.clone(), ready.clone())); + + tokio::select! { + biased; + _ = cancel.changed() => { + child.abort(); + let _ = child.await; + metrics::set_connected(&source.name, false); + info!(source = %source.name, "Oracle source stopped (config reload)"); + return; + } + result = &mut child => match result { + Ok(()) => { + warn!(source = %source.name, "Oracle client task exited unexpectedly, restarting"); + } + Err(join_error) => { + metrics::set_connected(&source.name, false); + error!(source = %source.name, error = %join_error, "Oracle client task panicked, restarting"); + } + }, + } + + if started.elapsed() >= HEALTHY_CONNECTION_THRESHOLD { + guard.reset(); + } + + tokio::select! { + biased; + _ = cancel.changed() => { + info!(source = %source.name, "Oracle source stopped (config reload)"); + return; + } + _ = tokio::time::sleep(guard.next_delay()) => {} + } + } +} + +fn heartbeat_deadline(heartbeat_interval_sec: u32) -> Duration { + if heartbeat_interval_sec == 0 { + return BOOTSTRAP_DEADLINE; + } + Duration::from_secs( + u64::from(heartbeat_interval_sec) * u64::from(DEADLINE_HEARTBEAT_MULTIPLIER), + ) +} + +async fn run_source( + source: SourceConfig, + address: String, + mut ready: Option>, +) { + let mut backoff = Backoff::default(); + let mut reconnect = false; + loop { + let stream = match connect(&address).await { + Ok(stream) => { + if reconnect { + metrics::inc_reconnects(&source.name); + info!(source = %source.name, "Connected to oracle source"); + } + stream + } + Err(e) => { + let delay = backoff.next_delay(); + warn!( + source = %source.name, + error = %e, + retry_in_secs = delay.as_secs(), + "Oracle source connect failed, retrying" + ); + tokio::time::sleep(delay).await; + continue; + } + }; + reconnect = true; + + metrics::set_connected(&source.name, true); + let started = Instant::now(); + let outcome = consume(&source, stream, &mut ready).await; + metrics::set_connected(&source.name, false); + + if outcome.hello && started.elapsed() >= HEALTHY_CONNECTION_THRESHOLD { + backoff.reset(); + } + + match outcome.error { + IpcError::Closed => { + info!(source = %source.name, "Oracle source disconnected, reconnecting") + } + other => { + warn!(source = %source.name, error = %other, "Oracle source connection error, reconnecting") + } + } + tokio::time::sleep(backoff.next_delay()).await; + } +} + +struct SessionOutcome { + hello: bool, + error: IpcError, +} + +async fn consume( + source: &SourceConfig, + mut stream: OracleStream, + ready: &mut Option>, +) -> SessionOutcome { + let mut read_deadline = BOOTSTRAP_DEADLINE; + let mut stale = false; + let mut hello = false; + let mut session_provider: Option = None; + loop { + let frame = match read_frame_with_timeout(&mut stream, read_deadline).await { + Ok(frame) => frame, + Err(error) => return SessionOutcome { hello, error }, + }; + metrics::inc_frames(&source.name); + metrics::set_last_frame(&source.name, now_unix()); + if !hello && !matches!(frame, OracleFrame::Hello { .. }) { + warn!( + source = %source.name, + "Oracle source sent a frame before Hello, dropping connection" + ); + return SessionOutcome { + hello, + error: IpcError::Closed, + }; + } + match frame { + OracleFrame::Hello { + protocol_version, + provider_id, + feeds, + heartbeat_interval_sec, + } => { + if protocol_version != PROTOCOL_VERSION { + warn!( + source = %source.name, + theirs = protocol_version, + ours = PROTOCOL_VERSION, + "Oracle source protocol version mismatch, dropping connection" + ); + return SessionOutcome { + hello, + error: IpcError::Closed, + }; + } + let feed_count = feeds.len(); + if feed_count > PROVIDER_FEEDS_MAX { + warn!( + source = %source.name, + feed_count, + max = PROVIDER_FEEDS_MAX, + "Oracle source advertised too many feeds, dropping connection" + ); + return SessionOutcome { + hello, + error: IpcError::Closed, + }; + } + if heartbeat_interval_sec > HEARTBEAT_INTERVAL_MAX_SEC { + warn!( + source = %source.name, + heartbeat_interval_sec, + max = HEARTBEAT_INTERVAL_MAX_SEC, + "Oracle source advertised an excessive heartbeat interval, dropping connection" + ); + return SessionOutcome { + hello, + error: IpcError::Closed, + }; + } + let registration = prices::register_feeds(&source.name, provider_id, feeds); + if registration.feeds_diverged { + metrics::inc_divergent_feeds(&source.name); + if warn_allowed(&LAST_DIVERGENCE_WARN_AT) { + warn!( + source = %source.name, + %provider_id, + "Oracle provider replicas advertise divergent feed sets" + ); + } + } + hello = true; + session_provider = Some(provider_id); + read_deadline = heartbeat_deadline(heartbeat_interval_sec); + info!( + source = %source.name, + %provider_id, + feed_count, + evicted = registration.evicted, + heartbeat_interval_sec, + "Oracle source handshake" + ); + if let Some(ready) = ready.take() { + let _ = ready.send(source.name.clone()); + } + } + OracleFrame::PriceUpdate { + provider_id, + feed_id, + payload, + ingested_at, + source_time, + } => { + if Some(provider_id) != session_provider { + metrics::inc_dropped_unexpected(&source.name); + if warn_allowed(&LAST_DROPPED_WARN_AT) { + warn!( + source = %source.name, + %provider_id, + "Oracle source sent an update for an unadvertised provider, dropping" + ); + } + continue; + } + let age = now_unix().saturating_sub(ingested_at); + trace!( + target: "oracle::frames", + source = %source.name, + %feed_id, + age_secs = age, + bytes = payload.len(), + "Received price update" + ); + if age > STALENESS_WARN_SEC { + if !stale { + stale = true; + warn!( + source = %source.name, + %feed_id, + age_secs = age, + "Oracle source data is stale (payload older than threshold)" + ); + } + } else if stale { + stale = false; + info!(source = %source.name, "Oracle source data freshness recovered"); + } + let order_time = if source_time != 0 { + source_time + } else { + ingested_at + }; + match prices::insert_if_newer(provider_id, feed_id, payload, order_time) { + prices::InsertOutcome::Inserted | prices::InsertOutcome::Stale => {} + prices::InsertOutcome::Unexpected => { + metrics::inc_dropped_unexpected(&source.name); + if warn_allowed(&LAST_DROPPED_WARN_AT) { + warn!( + source = %source.name, + %feed_id, + "Oracle source sent an update for an unexpected feed, dropping" + ); + } + } + } + } + OracleFrame::Heartbeat { sent_at_unix } => { + trace!( + target: "oracle::frames", + source = %source.name, + sent_at_unix, + "Received heartbeat" + ); + } + } + } +} + +fn now_unix() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + +mod metrics { + use std::collections::BTreeMap; + use std::io::{self, Write}; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + use std::sync::{Arc, LazyLock, Mutex, PoisonError}; + use std::time::Duration; + + use sov_metrics::Metric; + use tracing::info; + + #[derive(Default)] + struct SourceMetrics { + connected: AtomicBool, + reconnects: AtomicU64, + frames: AtomicU64, + last_frame_unix: AtomicU64, + dropped_unexpected: AtomicU64, + divergent_feeds: AtomicU64, + } + + static SOURCES: LazyLock>>> = + LazyLock::new(|| Mutex::new(BTreeMap::new())); + + fn source(name: &str) -> Arc { + SOURCES + .lock() + .unwrap_or_else(PoisonError::into_inner) + .entry(name.to_owned()) + .or_default() + .clone() + } + + pub fn set_connected(name: &str, connected: bool) { + source(name).connected.store(connected, Ordering::Relaxed); + } + + pub fn inc_reconnects(name: &str) { + source(name).reconnects.fetch_add(1, Ordering::Relaxed); + } + + pub fn inc_frames(name: &str) { + source(name).frames.fetch_add(1, Ordering::Relaxed); + } + + pub fn set_last_frame(name: &str, unix_secs: u64) { + source(name) + .last_frame_unix + .store(unix_secs, Ordering::Relaxed); + } + + pub fn inc_dropped_unexpected(name: &str) { + source(name) + .dropped_unexpected + .fetch_add(1, Ordering::Relaxed); + } + + pub fn inc_divergent_feeds(name: &str) { + source(name).divergent_feeds.fetch_add(1, Ordering::Relaxed); + } + + // One InfluxDB measurement per source, tagged by source name. + #[derive(Debug)] + struct OracleMetric { + source: String, + connected: bool, + reconnects: u64, + frames: u64, + last_frame_unix: u64, + dropped_unexpected: u64, + divergent_feeds: u64, + } + + impl Metric for OracleMetric { + fn measurement_name(&self) -> &'static str { + "oracle" + } + + fn serialize_for_telegraf(&self, buffer: &mut Vec) -> io::Result<()> { + write!( + buffer, + "{},source={} connected={},reconnects={},frames={},last_frame={},dropped_unexpected={},divergent_feeds={}", + self.measurement_name(), + sov_metrics::safe_telegraf_string(&self.source), + self.connected as u8, + self.reconnects, + self.frames, + self.last_frame_unix, + self.dropped_unexpected, + self.divergent_feeds, + ) + } + } + + /// Submits a snapshot of every source's counters to sov-metrics and logs it. + pub async fn run_reporter(interval: Duration) { + let mut ticker = tokio::time::interval(interval); + ticker.tick().await; + loop { + ticker.tick().await; + report_once(); + } + } + + fn report_once() { + let snapshot: Vec = { + let sources = SOURCES.lock().unwrap_or_else(PoisonError::into_inner); + sources + .iter() + .map(|(name, metrics)| OracleMetric { + source: name.clone(), + connected: metrics.connected.load(Ordering::Relaxed), + reconnects: metrics.reconnects.load(Ordering::Relaxed), + frames: metrics.frames.load(Ordering::Relaxed), + last_frame_unix: metrics.last_frame_unix.load(Ordering::Relaxed), + dropped_unexpected: metrics.dropped_unexpected.load(Ordering::Relaxed), + divergent_feeds: metrics.divergent_feeds.load(Ordering::Relaxed), + }) + .collect() + }; + + for metric in snapshot { + info!( + source = %metric.source, + connected = metric.connected, + reconnects = metric.reconnects, + frames = metric.frames, + last_frame = metric.last_frame_unix, + dropped_unexpected = metric.dropped_unexpected, + divergent_feeds = metric.divergent_feeds, + "oracle source metrics" + ); + sov_metrics::track_metrics(|tracker| tracker.submit(metric)); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn require_sources_defaults_to_false() { + let config = toml::from_str::("[oracle]") + .unwrap() + .oracle; + assert!(!config.require_sources); + } + + #[test] + fn transport_defaults_to_tcp() { + let toml = r#" + [oracle] + [[oracle.source]] + name = "chainlink" + socket_address = "127.0.0.1:9801" + "#; + let config = toml::from_str::(toml).unwrap().oracle; + assert_eq!(config.sources[0].transport, Transport::Tcp); + } + + fn source(toml: &str) -> SourceConfig { + toml::from_str::(&format!("[oracle]\n[[oracle.source]]\n{toml}")) + .unwrap() + .oracle + .sources + .remove(0) + } + + #[test] + fn parses_full_config() { + let toml = r#" + [oracle] + enabled = true + require_sources = true + [[oracle.source]] + name = "chainlink-1" + transport = "tcp" + socket_address = "127.0.0.1:9801" + [[oracle.source]] + name = "chainlink-2" + socket_address = "127.0.0.1:9802" + "#; + let config = toml::from_str::(toml).unwrap().oracle; + assert!(config.enabled); + assert!(config.require_sources); + assert_eq!(config.sources.len(), 2); + assert_eq!(config.sources[0].name, "chainlink-1"); + assert_eq!(config.sources[0].transport, Transport::Tcp); + assert_eq!(config.sources[1].transport, Transport::Tcp); + } + + #[test] + fn tcp_source_resolves_to_address() { + let s = source("name = \"a\"\ntransport = \"tcp\"\nsocket_address = \"127.0.0.1:9802\""); + assert_eq!(s.address().unwrap(), "127.0.0.1:9802"); + } + + #[test] + fn default_transport_resolves_to_address() { + let s = source("name = \"a\"\nsocket_address = \"127.0.0.1:9802\""); + assert_eq!(s.address().unwrap(), "127.0.0.1:9802"); + } + + #[test] + fn tcp_without_socket_address_is_an_error() { + let s = source("name = \"a\"\ntransport = \"tcp\""); + assert!(s.address().is_err()); + } + + #[test] + fn default_transport_without_address_is_an_error() { + let s = source("name = \"a\""); + assert!(s.address().is_err()); + } + + #[test] + fn applies_defaults() { + let config = toml::from_str::("[oracle]") + .unwrap() + .oracle; + assert!(!config.enabled); + assert!(config.sources.is_empty()); + } + + #[test] + fn resolve_derives_path() { + let path = resolve_config_path(None, Path::new("configs/mock/rollup.toml")); + assert_eq!(path, PathBuf::from("configs/mock/oracle.toml")); + } + + #[test] + fn resolve_uses_explicit_path() { + let path = resolve_config_path( + Some(PathBuf::from("/etc/relay/oracle.toml")), + Path::new("configs/mock/rollup.toml"), + ); + assert_eq!(path, PathBuf::from("/etc/relay/oracle.toml")); + } + + #[test] + fn missing_file_loads_as_none() { + let result = load_config(Path::new("/nonexistent/oracle.toml")).unwrap(); + assert!(result.is_none()); + } + + #[test] + fn load_or_create_writes_default() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("oracle.toml"); + + let config = load_or_create_config(&path).unwrap(); + assert!(!config.enabled); + assert!(config.sources.is_empty()); + assert!(path.exists()); + + let reloaded = load_config(&path).unwrap().unwrap(); + assert!(!reloaded.enabled); + } + + #[test] + fn malformed_file_is_an_error() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("oracle.toml"); + fs::write(&path, "this is not = valid toml [[[").unwrap(); + assert!(load_config(&path).is_err()); + } + + fn tcp_src(name: &str, address: &str) -> SourceConfig { + SourceConfig { + name: name.to_string(), + transport: Transport::Tcp, + socket_address: Some(address.to_string()), + } + } + + #[test] + fn duplicate_source_names_are_rejected() { + let config = OracleConfig { + sources: vec![tcp_src("dup", "127.0.0.1:1"), tcp_src("dup", "127.0.0.1:2")], + ..Default::default() + }; + assert!(validate(&config).is_err()); + } + + #[test] + fn unique_source_names_pass_validation() { + let config = OracleConfig { + sources: vec![tcp_src("a", "127.0.0.1:1"), tcp_src("b", "127.0.0.1:2")], + ..Default::default() + }; + assert!(validate(&config).is_ok()); + } + + #[test] + fn diff_add_remove_restart() { + let mut current = HashMap::new(); + current.insert("keep".to_string(), tcp_src("keep", "127.0.0.1:1")); + current.insert("change".to_string(), tcp_src("change", "127.0.0.1:2")); + current.insert("drop".to_string(), tcp_src("drop", "127.0.0.1:3")); + + let new_sources = vec![ + tcp_src("keep", "127.0.0.1:1"), + tcp_src("change", "127.0.0.1:9"), + tcp_src("add", "127.0.0.1:4"), + ]; + let diff = diff_sources(¤t, new_sources); + + assert_eq!( + diff.to_add + .iter() + .map(|s| s.name.as_str()) + .collect::>(), + ["add"] + ); + assert_eq!( + diff.to_restart + .iter() + .map(|s| s.name.as_str()) + .collect::>(), + ["change"] + ); + assert_eq!(diff.to_remove, ["drop"]); + } + + #[test] + fn heartbeat_deadline_scales() { + assert_eq!( + heartbeat_deadline(10), + Duration::from_secs(10 * u64::from(DEADLINE_HEARTBEAT_MULTIPLIER)) + ); + } + + #[test] + fn heartbeat_deadline_zero_interval() { + assert_eq!(heartbeat_deadline(0), BOOTSTRAP_DEADLINE); + } + + #[tokio::test] + async fn require_sources_ready() { + let (tx, rx) = mpsc::unbounded_channel(); + tx.send("a".to_string()).unwrap(); + tx.send("b".to_string()).unwrap(); + let pending: HashSet = ["a", "b"].iter().map(|s| s.to_string()).collect(); + assert!(await_sources_ready(pending, rx, Duration::from_secs(5)) + .await + .is_ok()); + } + + #[tokio::test] + async fn require_sources_times_out() { + let (tx, rx) = mpsc::unbounded_channel(); + tx.send("a".to_string()).unwrap(); + let pending: HashSet = ["a", "b"].iter().map(|s| s.to_string()).collect(); + let result = await_sources_ready(pending, rx, Duration::from_millis(50)).await; + assert!(result.is_err()); + drop(tx); + } +} diff --git a/crates/shared/price-oracle-ipc/Cargo.toml b/crates/shared/price-oracle-ipc/Cargo.toml new file mode 100644 index 00000000..bb582970 --- /dev/null +++ b/crates/shared/price-oracle-ipc/Cargo.toml @@ -0,0 +1,20 @@ +[package] +description = "Provider-agnostic IPC protocol for oracle provider services" +edition = { workspace = true } +license = { workspace = true } +name = "price-oracle-ipc" +publish = false +repository = { workspace = true } +rust-version = { workspace = true } +version = { workspace = true } + +[dependencies] +alloy-primitives = { workspace = true, features = ["borsh"] } +borsh = { workspace = true } +socket2 = { version = "0.5", default-features = false, features = ["all"] } +thiserror = { version = "1", default-features = false } +tokio = { workspace = true } +tracing = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["test-util"] } diff --git a/crates/shared/price-oracle-ipc/README.md b/crates/shared/price-oracle-ipc/README.md new file mode 100644 index 00000000..f2289662 --- /dev/null +++ b/crates/shared/price-oracle-ipc/README.md @@ -0,0 +1,54 @@ +# Protocol + +An IPC protocol library which is shared between price oracle provider services and +the Relay Chain sequencer. It defines the wire format and transport that an oracle +provider service uses to stream signed price reports to a local subscriber (i.e. +the sequencer), over a TCP connection. The protocol is provider-agnostic, the price +payload is carried as an opaque byte blob and this library does not decode or validate it. + +## Frames + +Messages are `OracleFrame` values, serialized with Borsh. +There are three variants. + +- `Hello` - Sent first by the server, carries the protocol version, provider id, the list of subscribed feeds, and the server's heartbeat interval in seconds. +- `PriceUpdate` - A price report for one feed, with provider id, feed id, opaque payload, an ingested-at timestamp, and a source timestamp taken from the upstream report. +- `Heartbeat` - A liveness ping carrying a Unix timestamp in seconds. + +A feed is identified by a `FeedKey`, the pair of provider id and feed id, both 32-byte values. + +Each frame is length-prefixed. +The wire layout is a 4-byte little-endian length followed by the Borsh-encoded frame body. + +- `write_frame` - Encodes a frame and writes the length prefix and body. +- `read_frame` - Reads the prefix, then the body, and decodes it. +- `write_frame_with_timeout` - `write_frame` bounded by a deadline, returning `IpcError::WriteTimeout` on expiry. +- `read_frame_with_timeout` - `read_frame` bounded by a deadline, returning `IpcError::ReadTimeout` on expiry. + +Frames larger than `MAX_FRAME_LEN` (16 MiB) are rejected on both read and write. +A clean end of stream while reading the length prefix is reported as `IpcError::Closed` +rather than an I/O error. The codec works over any `AsyncRead` or `AsyncWrite` +the socket type is not assumed. + +## Transport + +Helpers wrap a local TCP connection (no TLS). + +- `connect` - Opens an `OracleStream` to a `host:port` address, bounded by `DEFAULT_CONNECT_TIMEOUT` (10s). It sets `TCP_NODELAY` and `SO_KEEPALIVE`. +- `bind` - Binds an `OracleListener` to a `host:port` address. +- `OracleStream` - A connected TCP stream, implementing `AsyncRead` and `AsyncWrite`. +- `OracleListener` - A bound listener, with `accept` and `local_addr`. Accepted connections get `TCP_NODELAY` and `SO_KEEPALIVE`. +- `BoundListener` - Owns a bound listener and its resolved address, exposing it through `address()` (resolving an ephemeral `:0` port). +- `Backoff` - A reusable exponential backoff helper, doubling from a minimum up to a maximum, defaulting to 1s through 30s. + +## Errors + +All fallible operations return `IpcError`. + +- `Closed` - The peer closed the connection. +- `FrameTooLarge` - A frame exceeded `MAX_FRAME_LEN`. +- `ConnectTimeout` - A connect attempt exceeded `DEFAULT_CONNECT_TIMEOUT`. +- `ReadTimeout` - A `read_frame_with_timeout` call exceeded its deadline. +- `WriteTimeout` - A `write_frame_with_timeout` call exceeded its deadline. +- `Io` - An underlying I/O error. +- `Codec` - A Borsh serialization or deserialization error. diff --git a/crates/shared/price-oracle-ipc/src/codec.rs b/crates/shared/price-oracle-ipc/src/codec.rs new file mode 100644 index 00000000..2c867c8a --- /dev/null +++ b/crates/shared/price-oracle-ipc/src/codec.rs @@ -0,0 +1,158 @@ +use std::time::Duration; + +use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; + +use crate::error::IpcError; +use crate::types::OracleFrame; + +pub const MAX_FRAME_LEN: u32 = 16 * 1024 * 1024; + +pub async fn write_frame(writer: &mut W, frame: &OracleFrame) -> Result<(), IpcError> +where + W: AsyncWrite + Unpin, +{ + let body = borsh::to_vec(frame).map_err(IpcError::Codec)?; + let len = u32::try_from(body.len()).map_err(|_| IpcError::FrameTooLarge(body.len()))?; + if len > MAX_FRAME_LEN { + return Err(IpcError::FrameTooLarge(body.len())); + } + writer.write_all(&len.to_le_bytes()).await?; + writer.write_all(&body).await?; + Ok(()) +} + +pub async fn read_frame(reader: &mut R) -> Result +where + R: AsyncRead + Unpin, +{ + let mut len_buf = [0u8; 4]; + match reader.read_exact(&mut len_buf).await { + Ok(_) => {} + Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Err(IpcError::Closed), + Err(e) => return Err(e.into()), + } + let len = u32::from_le_bytes(len_buf); + if len > MAX_FRAME_LEN { + return Err(IpcError::FrameTooLarge(len as usize)); + } + let mut body = vec![0u8; len as usize]; + reader.read_exact(&mut body).await?; + borsh::from_slice(&body).map_err(IpcError::Codec) +} + +pub async fn read_frame_with_timeout( + reader: &mut R, + timeout: Duration, +) -> Result +where + R: AsyncRead + Unpin, +{ + tokio::time::timeout(timeout, read_frame(reader)) + .await + .map_err(|_| IpcError::ReadTimeout)? +} + +pub async fn write_frame_with_timeout( + writer: &mut W, + frame: &OracleFrame, + timeout: Duration, +) -> Result<(), IpcError> +where + W: AsyncWrite + Unpin, +{ + tokio::time::timeout(timeout, write_frame(writer, frame)) + .await + .map_err(|_| IpcError::WriteTimeout)? +} + +#[cfg(test)] +mod tests { + use super::*; + use alloy_primitives::B256; + + fn sample_update() -> OracleFrame { + OracleFrame::PriceUpdate { + provider_id: B256::repeat_byte(0xab), + feed_id: B256::repeat_byte(0xcd), + payload: vec![0x12, 0x34, 0x56, 0x78], + ingested_at: 0x1234_5678, + source_time: 0x1234_5670, + } + } + + #[tokio::test] + async fn roundtrips_a_frame_through_a_pipe() { + let (mut a, mut b) = tokio::io::duplex(1024); + let frame = sample_update(); + write_frame(&mut a, &frame).await.unwrap(); + let got = read_frame(&mut b).await.unwrap(); + assert_eq!(got, frame); + } + + #[tokio::test] + async fn roundtrips_multiple_frames_in_order() { + let (mut a, mut b) = tokio::io::duplex(4096); + let hello = OracleFrame::Hello { + protocol_version: crate::PROTOCOL_VERSION, + provider_id: B256::repeat_byte(0x01), + feeds: vec![B256::repeat_byte(0x02), B256::repeat_byte(0x03)], + heartbeat_interval_sec: 10, + }; + write_frame(&mut a, &hello).await.unwrap(); + write_frame(&mut a, &sample_update()).await.unwrap(); + assert_eq!(read_frame(&mut b).await.unwrap(), hello); + assert_eq!(read_frame(&mut b).await.unwrap(), sample_update()); + } + + #[tokio::test] + async fn clean_eof_reports_closed() { + let (a, mut b) = tokio::io::duplex(64); + drop(a); + assert!(matches!(read_frame(&mut b).await, Err(IpcError::Closed))); + } + + #[tokio::test] + async fn oversized_length_is_rejected() { + let (mut a, mut b) = tokio::io::duplex(64); + a.write_all(&(MAX_FRAME_LEN + 1).to_le_bytes()) + .await + .unwrap(); + assert!(matches!( + read_frame(&mut b).await, + Err(IpcError::FrameTooLarge(_)) + )); + } + + #[tokio::test] + async fn read_with_timeout_elapses_on_idle_stream() { + let (_a, mut b) = tokio::io::duplex(64); + let result = read_frame_with_timeout(&mut b, Duration::from_millis(20)).await; + assert!(matches!(result, Err(IpcError::ReadTimeout))); + } + + #[tokio::test] + async fn read_with_timeout_returns_frame_when_in_time() { + let (mut a, mut b) = tokio::io::duplex(1024); + write_frame(&mut a, &sample_update()).await.unwrap(); + let got = read_frame_with_timeout(&mut b, Duration::from_secs(5)) + .await + .unwrap(); + assert_eq!(got, sample_update()); + } + + #[tokio::test] + async fn write_with_timeout_elapses_when_peer_never_reads() { + let (_a, mut b) = tokio::io::duplex(16); + let result = + write_frame_with_timeout(&mut b, &sample_update(), Duration::from_millis(20)).await; + assert!(matches!(result, Err(IpcError::WriteTimeout))); + } + + #[tokio::test] + async fn invalid_frame_body_is_codec_error() { + let (mut a, mut b) = tokio::io::duplex(64); + a.write_all(&1u32.to_le_bytes()).await.unwrap(); + a.write_all(&[0xff]).await.unwrap(); + assert!(matches!(read_frame(&mut b).await, Err(IpcError::Codec(_)))); + } +} diff --git a/crates/shared/price-oracle-ipc/src/error.rs b/crates/shared/price-oracle-ipc/src/error.rs new file mode 100644 index 00000000..e7b789f8 --- /dev/null +++ b/crates/shared/price-oracle-ipc/src/error.rs @@ -0,0 +1,23 @@ +#[derive(Debug, thiserror::Error)] +pub enum IpcError { + #[error("connection closed by peer")] + Closed, + + #[error("frame too large: {0} bytes")] + FrameTooLarge(usize), + + #[error("connect timed out")] + ConnectTimeout, + + #[error("read timed out")] + ReadTimeout, + + #[error("write timed out")] + WriteTimeout, + + #[error("io error: {0}")] + Io(#[from] std::io::Error), + + #[error("frame codec error: {0}")] + Codec(std::io::Error), +} diff --git a/crates/shared/price-oracle-ipc/src/lib.rs b/crates/shared/price-oracle-ipc/src/lib.rs new file mode 100644 index 00000000..9a462b9e --- /dev/null +++ b/crates/shared/price-oracle-ipc/src/lib.rs @@ -0,0 +1,15 @@ +pub mod codec; +pub mod error; +pub mod transport; +pub mod types; + +pub use codec::{ + read_frame, read_frame_with_timeout, write_frame, write_frame_with_timeout, MAX_FRAME_LEN, +}; +pub use error::IpcError; +pub use transport::{ + bind, connect, Backoff, BoundListener, OracleListener, OracleStream, DEFAULT_CONNECT_TIMEOUT, +}; +pub use types::{FeedKey, OracleFrame, PROTOCOL_VERSION}; + +pub use alloy_primitives::B256; diff --git a/crates/shared/price-oracle-ipc/src/transport.rs b/crates/shared/price-oracle-ipc/src/transport.rs new file mode 100644 index 00000000..53bb0bd3 --- /dev/null +++ b/crates/shared/price-oracle-ipc/src/transport.rs @@ -0,0 +1,181 @@ +use std::pin::Pin; +use std::task::{Context, Poll}; +use std::time::Duration; + +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +use tokio::net::{TcpListener, TcpStream}; + +use crate::error::IpcError; + +pub const DEFAULT_BACKOFF_MIN: Duration = Duration::from_secs(1); +pub const DEFAULT_BACKOFF_MAX: Duration = Duration::from_secs(30); +pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +pub const DEFAULT_KEEPALIVE_IDLE: Duration = Duration::from_secs(15); +pub const DEFAULT_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(5); +pub const DEFAULT_KEEPALIVE_RETRIES: u32 = 3; + +fn apply_tcp_keepalive(stream: &TcpStream) -> Result<(), IpcError> { + let keepalive = socket2::TcpKeepalive::new() + .with_time(DEFAULT_KEEPALIVE_IDLE) + .with_interval(DEFAULT_KEEPALIVE_INTERVAL) + .with_retries(DEFAULT_KEEPALIVE_RETRIES); + socket2::SockRef::from(stream).set_tcp_keepalive(&keepalive)?; + Ok(()) +} + +#[derive(Debug)] +pub struct OracleStream(TcpStream); + +impl AsyncRead for OracleStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_read(cx, buf) + } +} + +impl AsyncWrite for OracleStream { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_shutdown(cx) + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[std::io::IoSlice<'_>], + ) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_write_vectored(cx, bufs) + } + + fn is_write_vectored(&self) -> bool { + self.0.is_write_vectored() + } +} + +pub async fn connect(address: &str) -> Result { + let stream = tokio::time::timeout(DEFAULT_CONNECT_TIMEOUT, TcpStream::connect(address)) + .await + .map_err(|_| IpcError::ConnectTimeout)??; + stream.set_nodelay(true)?; + apply_tcp_keepalive(&stream)?; + Ok(OracleStream(stream)) +} + +pub struct OracleListener(TcpListener); + +impl OracleListener { + pub async fn accept(&self) -> Result { + let (stream, _addr) = self.0.accept().await?; + stream.set_nodelay(true)?; + apply_tcp_keepalive(&stream)?; + Ok(OracleStream(stream)) + } + + pub fn local_addr(&self) -> Option { + self.0.local_addr().ok() + } +} + +pub async fn bind(address: &str) -> Result { + Ok(OracleListener(TcpListener::bind(address).await?)) +} + +pub struct BoundListener { + listener: OracleListener, + address: String, +} + +impl BoundListener { + pub async fn bind(address: impl Into) -> Result { + let requested = address.into(); + let listener = bind(&requested).await?; + let address = listener + .local_addr() + .map(|addr| addr.to_string()) + .unwrap_or(requested); + Ok(Self { listener, address }) + } + + pub async fn accept(&self) -> Result { + self.listener.accept().await + } + + pub fn address(&self) -> &str { + &self.address + } +} + +#[derive(Debug, Clone)] +pub struct Backoff { + current: Duration, + min: Duration, + max: Duration, +} + +impl Backoff { + pub fn new(min: Duration, max: Duration) -> Self { + Self { + current: min, + min, + max, + } + } + + pub fn next_delay(&mut self) -> Duration { + let delay = self.current; + self.current = (self.current * 2).min(self.max); + delay + } + + pub fn reset(&mut self) { + self.current = self.min; + } +} + +impl Default for Backoff { + fn default() -> Self { + Self::new(DEFAULT_BACKOFF_MIN, DEFAULT_BACKOFF_MAX) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backoff_doubles_then_saturates() { + let mut b = Backoff::new(Duration::from_secs(1), Duration::from_secs(8)); + assert_eq!(b.next_delay(), Duration::from_secs(1)); + assert_eq!(b.next_delay(), Duration::from_secs(2)); + assert_eq!(b.next_delay(), Duration::from_secs(4)); + assert_eq!(b.next_delay(), Duration::from_secs(8)); + assert_eq!(b.next_delay(), Duration::from_secs(8)); + b.reset(); + assert_eq!(b.next_delay(), Duration::from_secs(1)); + } + + #[tokio::test] + async fn tcp_bind_then_connect_roundtrip() { + let listener = BoundListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.address().to_string(); + + let client = tokio::spawn(async move { connect(&address).await.is_ok() }); + let accepted = listener.accept().await.is_ok(); + assert!(accepted); + assert!(client.await.unwrap()); + } +} diff --git a/crates/shared/price-oracle-ipc/src/types.rs b/crates/shared/price-oracle-ipc/src/types.rs new file mode 100644 index 00000000..88875c94 --- /dev/null +++ b/crates/shared/price-oracle-ipc/src/types.rs @@ -0,0 +1,41 @@ +use alloy_primitives::B256; +use borsh::{BorshDeserialize, BorshSerialize}; + +pub const PROTOCOL_VERSION: u16 = 1; + +#[derive( + Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, BorshSerialize, BorshDeserialize, +)] +pub struct FeedKey { + pub provider_id: B256, + pub feed_id: B256, +} + +impl FeedKey { + pub const fn new(provider_id: B256, feed_id: B256) -> Self { + Self { + provider_id, + feed_id, + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, BorshSerialize, BorshDeserialize)] +pub enum OracleFrame { + Hello { + protocol_version: u16, + provider_id: B256, + feeds: Vec, + heartbeat_interval_sec: u32, + }, + PriceUpdate { + provider_id: B256, + feed_id: B256, + payload: Vec, + ingested_at: u64, + source_time: u64, + }, + Heartbeat { + sent_at_unix: u64, + }, +} diff --git a/crates/shared/price-oracle-ipc/tests/common/mod.rs b/crates/shared/price-oracle-ipc/tests/common/mod.rs new file mode 100644 index 00000000..69d88025 --- /dev/null +++ b/crates/shared/price-oracle-ipc/tests/common/mod.rs @@ -0,0 +1,68 @@ +use std::time::Duration; + +use price_oracle_ipc::{ + connect, read_frame_with_timeout, write_frame, Backoff, BoundListener, OracleFrame, B256, + PROTOCOL_VERSION, +}; + +pub async fn listener() -> BoundListener { + BoundListener::bind("127.0.0.1:0").await.unwrap() +} + +pub fn hello() -> OracleFrame { + OracleFrame::Hello { + protocol_version: PROTOCOL_VERSION, + provider_id: B256::repeat_byte(0x11), + feeds: vec![B256::repeat_byte(0x01)], + heartbeat_interval_sec: 1, + } +} + +pub fn update(feed: u8, payload: &[u8]) -> OracleFrame { + OracleFrame::PriceUpdate { + provider_id: B256::repeat_byte(0x11), + feed_id: B256::repeat_byte(feed), + payload: payload.to_vec(), + ingested_at: 1_700_000_000, + source_time: 1_700_000_000, + } +} + +pub fn heartbeat(sent_at_unix: u64) -> OracleFrame { + OracleFrame::Heartbeat { sent_at_unix } +} + +pub fn serve_once( + listener: BoundListener, + frames: Vec, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut stream = listener.accept().await.unwrap(); + for frame in &frames { + write_frame(&mut stream, frame).await.unwrap(); + } + }) +} + +pub async fn run_consumer(address: String, deadline: Duration, want: usize) -> Vec { + let mut backoff = Backoff::new(Duration::from_millis(1), Duration::from_millis(20)); + let mut collected = Vec::new(); + loop { + let mut stream = match connect(&address).await { + Ok(stream) => { + backoff.reset(); + stream + } + Err(_) => { + tokio::time::sleep(backoff.next_delay()).await; + continue; + } + }; + while let Ok(frame) = read_frame_with_timeout(&mut stream, deadline).await { + collected.push(frame); + if collected.len() >= want { + return collected; + } + } + } +} diff --git a/crates/shared/price-oracle-ipc/tests/integration.rs b/crates/shared/price-oracle-ipc/tests/integration.rs new file mode 100644 index 00000000..83a08ebe --- /dev/null +++ b/crates/shared/price-oracle-ipc/tests/integration.rs @@ -0,0 +1,142 @@ +mod common; + +use std::time::Duration; + +use price_oracle_ipc::{ + connect, read_frame_with_timeout, write_frame, write_frame_with_timeout, IpcError, OracleFrame, +}; +use tokio::sync::oneshot; + +#[tokio::test] +async fn session_round_trip() { + let listener = common::listener().await; + let address = listener.address().to_string(); + let expected = vec![ + common::hello(), + common::update(0x01, b"snap-1"), + common::update(0x02, b"snap-2"), + common::heartbeat(7), + common::update(0x01, b"live-1"), + ]; + let server = common::serve_once(listener, expected.clone()); + + let mut client = connect(&address).await.unwrap(); + let mut got = Vec::new(); + for _ in 0..expected.len() { + got.push( + read_frame_with_timeout(&mut client, Duration::from_secs(5)) + .await + .unwrap(), + ); + } + + assert_eq!(got, expected); + server.await.unwrap(); +} + +#[tokio::test] +async fn silent_server_trips_read_deadline() { + let listener = common::listener().await; + let address = listener.address().to_string(); + let (tx, rx) = oneshot::channel::<()>(); + let server = tokio::spawn(async move { + let mut stream = listener.accept().await.unwrap(); + write_frame(&mut stream, &common::hello()).await.unwrap(); + let _ = rx.await; + }); + + let mut client = connect(&address).await.unwrap(); + let hello = read_frame_with_timeout(&mut client, Duration::from_secs(5)) + .await + .unwrap(); + assert!(matches!(hello, OracleFrame::Hello { .. })); + + let timed_out = read_frame_with_timeout(&mut client, Duration::from_millis(50)).await; + assert!(matches!(timed_out, Err(IpcError::ReadTimeout))); + + drop(tx); + server.await.unwrap(); +} + +#[tokio::test] +async fn server_disconnect_reports_closed() { + let listener = common::listener().await; + let address = listener.address().to_string(); + let server = common::serve_once(listener, vec![common::hello()]); + + let mut client = connect(&address).await.unwrap(); + let hello = read_frame_with_timeout(&mut client, Duration::from_secs(5)) + .await + .unwrap(); + assert!(matches!(hello, OracleFrame::Hello { .. })); + + let closed = read_frame_with_timeout(&mut client, Duration::from_secs(5)).await; + assert!(matches!(closed, Err(IpcError::Closed))); + server.await.unwrap(); +} + +#[tokio::test] +async fn write_trips_deadline_when_peer_never_reads() { + let listener = common::listener().await; + let address = listener.address().to_string(); + let server = tokio::spawn(async move { + let mut stream = listener.accept().await.unwrap(); + let big = common::update(0x01, &vec![0u8; 1024 * 1024]); + for _ in 0..1000 { + match write_frame_with_timeout(&mut stream, &big, Duration::from_millis(100)).await { + Ok(()) => continue, + Err(e) => return e, + } + } + panic!("expected a write to time out, but all writes succeeded"); + }); + + let _client = connect(&address).await.unwrap(); + let err = server.await.unwrap(); + assert!(matches!(err, IpcError::WriteTimeout)); +} + +#[tokio::test] +async fn consumer_reconnects_after_drop() { + let listener = common::listener().await; + let address = listener.address().to_string(); + let server = tokio::spawn(async move { + let mut first = listener.accept().await.unwrap(); + write_frame(&mut first, &common::hello()).await.unwrap(); + drop(first); + + let mut second = listener.accept().await.unwrap(); + write_frame(&mut second, &common::hello()).await.unwrap(); + write_frame(&mut second, &common::update(0x09, b"after-reconnect")) + .await + .unwrap(); + }); + + let collected = common::run_consumer(address, Duration::from_secs(5), 3).await; + assert_eq!(collected.len(), 3); + assert_eq!(collected[2], common::update(0x09, b"after-reconnect")); + server.await.unwrap(); +} + +#[tokio::test] +async fn streams_many_frames_in_order() { + let listener = common::listener().await; + let address = listener.address().to_string(); + let expected: Vec = (0u32..100) + .map(|i| common::update(0x01, &i.to_le_bytes())) + .collect(); + let server = common::serve_once(listener, expected.clone()); + + let mut client = connect(&address).await.unwrap(); + let mut got = Vec::with_capacity(expected.len()); + for _ in 0..expected.len() { + got.push( + read_frame_with_timeout(&mut client, Duration::from_secs(5)) + .await + .unwrap(), + ); + } + + assert_eq!(got, expected); + server.await.unwrap(); +} diff --git a/crates/stf/Cargo.toml b/crates/stf/Cargo.toml index 9711471f..9e19891d 100644 --- a/crates/stf/Cargo.toml +++ b/crates/stf/Cargo.toml @@ -27,9 +27,11 @@ sov-test-state-consistency = { workspace = true } sov-address = { workspace = true } sov-evm = { workspace = true } sov-eip712-auth = { workspace = true } +price-oracle = { workspace = true } anyhow = { workspace = true } borsh = { workspace = true } +bytes = { workspace = true, optional = true } [build-dependencies] anyhow = { workspace = true } @@ -37,6 +39,7 @@ stf-starter-declaration = { workspace = true } sov-address = { workspace = true, features = ["evm"] } sov-modules-api = { workspace = true } sov-mock-zkvm = { workspace = true } +sov-mock-da = { workspace = true } sov-celestia-adapter = { workspace = true, optional = true } serde_json = { workspace = true } sov-universal-wallet = { workspace = true, features = ["serde"] } @@ -60,12 +63,15 @@ native = [ "sov-bank/native", "sov-address/native", "sov-celestia-adapter?/native", + "sov-mock-da/native", "sov-mock-zkvm/native", "sov-evm/native", "sov-test-state-consistency/native", "sov-eip712-auth/native", "sov-state/native", - "sov-chain-state/native" + "sov-chain-state/native", + "price-oracle/native", + "dep:bytes" ] bench = [ diff --git a/crates/stf/build.rs b/crates/stf/build.rs index d917c1d3..96905132 100644 --- a/crates/stf/build.rs +++ b/crates/stf/build.rs @@ -3,7 +3,10 @@ use sov_mock_zkvm::MockZkvm; use sov_modules_api::configurable_spec::ConfigurableSpec; use stf_starter_declaration::Runtime; -pub use sov_celestia_adapter::verifier::CelestiaSpec as DaSpec; +#[cfg(feature = "celestia_da")] +use sov_celestia_adapter::verifier::CelestiaSpec as DaSpec; +#[cfg(not(feature = "celestia_da"))] +use sov_mock_da::MockDaSpec as DaSpec; #[cfg(feature = "native")] type ExecMode = sov_modules_api::execution_mode::Native; diff --git a/crates/stf/src/delegation.rs b/crates/stf/src/delegation.rs index 8839490a..6fdd724a 100644 --- a/crates/stf/src/delegation.rs +++ b/crates/stf/src/delegation.rs @@ -1,14 +1,42 @@ -//! This is a technical only module to forward all necessary implementations to inner, non-authenticated Runtime +//! Forwards the required trait implementations to the inner non-authenticated runtime. +use price_oracle::PriceReports; use sov_address::{EthereumAddress, FromVmAddress}; +use sov_bank::Amount; use sov_capabilities::StandardProvenRollupCapabilities as StandardCapabilities; use sov_eip712_auth::{Eip712AuthenticatorTrait, Secp256k1CryptoSpec}; use sov_evm::EthereumAuthenticator; use sov_hyperlane_integration::HyperlaneAddress; use sov_kernels::soft_confirmations::SoftConfirmationsKernel; +use sov_modules_api::capabilities::AuthorizationData; +use sov_modules_api::capabilities::GasEnforcer; #[cfg(feature = "native")] use sov_modules_api::capabilities::KernelWithSlotMapping; -use sov_modules_api::capabilities::TransactionAuthenticator; +use sov_modules_api::capabilities::ProofProcessor; +use sov_modules_api::capabilities::SequencerAuthorization; +use sov_modules_api::capabilities::SequencerRemuneration; +use sov_modules_api::capabilities::TransactionAuthorizer; use sov_modules_api::capabilities::{Guard, HasCapabilities, HasKernel}; +use sov_modules_api::capabilities::{SequencingDataHandler, TransactionAuthenticator}; +use sov_modules_api::transaction::ProverReward; +use sov_modules_api::transaction::RemainingFunds; +use sov_modules_api::transaction::SequencerReward; +use sov_modules_api::AggregatedProofPublicData; +use sov_modules_api::ExecutionContext; +use sov_modules_api::Gas; +use sov_modules_api::GetGasPrice; +use sov_modules_api::InfallibleStateAccessor; +use sov_modules_api::InvalidProofError; +use sov_modules_api::OperatingMode; +use sov_modules_api::Rewards; +use sov_modules_api::SequencerType; +use sov_modules_api::SerializedAggregatedProof; +use sov_modules_api::SerializedAttestation; +use sov_modules_api::SerializedChallenge; +use sov_modules_api::SovAttestation; +use sov_modules_api::SovStateTransitionPublicData; +use sov_modules_api::StateReader; +use sov_modules_api::StateWriter; +use sov_modules_api::VersionReader; use sov_modules_api::{prelude::*, RawTx}; use sov_modules_api::{ AuthenticatedTransactionData, BlockHooks, DispatchCall, EncodeCall, Genesis, GenesisState, @@ -16,6 +44,9 @@ use sov_modules_api::{ }; use sov_modules_api::{ModuleError, ModuleId, ModuleInfo, NestedEnumUtils}; use sov_rollup_interface::da::DaSpec; +use sov_state::Kernel; +use sov_state::User; +use std::convert::Infallible; use crate::authentication::EvmAndEip712AuthenticatorInput; use crate::Runtime; @@ -201,20 +232,22 @@ impl HasCapabilities for Runtime where S::Address: HyperlaneAddress + FromVmAddress, { - type Capabilities<'a> = StandardCapabilities<'a, S, &'a mut sov_paymaster::Paymaster>; - type SequencingData = sov_modules_api::HDTimestamp; + type Capabilities<'a> = RelayChainCapabilities<'a, S>; + type SequencingData = PriceReports; fn capabilities(&mut self) -> Guard> { - Guard::new(StandardCapabilities { - bank: &mut self.0.bank, - sequencer_registry: &mut self.0.sequencer_registry, - accounts: &mut self.0.accounts, - uniqueness: &mut self.0.uniqueness, - gas_payer: &mut self.0.paymaster, - chain_state: &mut self.0.chain_state, - operator_incentives: &mut self.0.operator_incentives, - attester_incentives: &mut self.0.attester_incentives, - prover_incentives: &mut self.0.prover_incentives, + Guard::new(RelayChainCapabilities { + standard: StandardCapabilities { + bank: &mut self.0.bank, + sequencer_registry: &mut self.0.sequencer_registry, + accounts: &mut self.0.accounts, + uniqueness: &mut self.0.uniqueness, + gas_payer: &mut self.0.paymaster, + chain_state: &mut self.0.chain_state, + operator_incentives: &mut self.0.operator_incentives, + attester_incentives: &mut self.0.attester_incentives, + prover_incentives: &mut self.0.prover_incentives, + }, }) } } @@ -273,3 +306,259 @@ where EvmAndEip712AuthenticatorInput::Eip712(tx) } } + +pub struct RelayChainCapabilities<'a, S: Spec> { + standard: StandardCapabilities<'a, S, &'a mut sov_paymaster::Paymaster>, +} + +impl SequencingDataHandler for RelayChainCapabilities<'_, S> { + type SequencingData = PriceReports; + + fn handle_sequencing_data( + &mut self, + _data: Self::SequencingData, + _context: &Context, + _state: &mut impl TxState, + ) -> anyhow::Result<()> { + Ok(()) + } + + #[cfg(feature = "native")] + fn create_sequencing_data(&self) -> Self::SequencingData { + // This snapshot is attached to every transaction and pruned later by + // finalize_sequencing_data before it reaches the DA layer. + // The SDK sizes transactions for batch limits using this pre-pruned size. + // Its best to keep the snapshot bounded to avoid reducing batch throughput. + crate::prices::snapshot_prices() + } + + #[cfg(feature = "native")] + fn finalize_sequencing_data( + &mut self, + data: Self::SequencingData, + scratchpad: Option, + ) -> Self::SequencingData { + price_oracle::prune_unused(data, scratchpad) + } +} + +impl GasEnforcer for RelayChainCapabilities<'_, S> { + fn try_reserve_gas( + &mut self, + tx: &AuthenticatedTransactionData, + gas_price: ::Price, + ctx: &mut Context, + state: &mut impl StateAccessor, + ) -> anyhow::Result<()> { + self.standard.try_reserve_gas(tx, gas_price, ctx, state) + } + + fn try_reserve_gas_for_proof( + &mut self, + tx: &AuthenticatedTransactionData, + gas_price: ::Price, + sender: &S::Address, + state: &mut impl StateAccessor, + ) -> anyhow::Result<()> { + self.standard + .try_reserve_gas_for_proof(tx, gas_price, sender, state) + } + + fn reward_prover( + &mut self, + prover_rewards: &ProverReward, + operating_mode: OperatingMode, + state: &mut impl InfallibleStateAccessor, + ) { + self.standard + .reward_prover(prover_rewards, operating_mode, state); + } + + fn refund_remaining_gas( + &mut self, + recipient: &S::Address, + remaining_funds: &RemainingFunds, + state: &mut impl InfallibleStateAccessor, + ) { + self.standard + .refund_remaining_gas(recipient, remaining_funds, state); + } + + fn reward_prover_from_sequencer_balance( + &mut self, + amount: Amount, + sequencer: &S::Address, + operating_mode: OperatingMode, + state: &mut impl InfallibleStateAccessor, + ) -> anyhow::Result<()> { + self.standard + .reward_prover_from_sequencer_balance(amount, sequencer, operating_mode, state) + } + + fn return_escrowed_funds_to_sequencer< + Accessor: StateReader + + StateWriter + + StateWriter + + StateReader + + VersionReader, + >( + &mut self, + bond_amount: Amount, + reward: Rewards, + sequencer: &::Address, + state: &mut Accessor, + ) { + self.standard + .return_escrowed_funds_to_sequencer(bond_amount, reward, sequencer, state); + } +} + +impl SequencerAuthorization for RelayChainCapabilities<'_, S> { + fn is_preferred_sequencer( + &self, + sequencer: &::Address, + state: &mut impl InfallibleStateAccessor, + ) -> bool { + self.standard.is_preferred_sequencer(sequencer, state) + } +} + +impl TransactionAuthorizer for RelayChainCapabilities<'_, S> { + fn resolve_context( + &mut self, + auth_data: &AuthorizationData, + sequencer: &::Address, + sequencer_rollup_address: S::Address, + state: &mut impl StateAccessor, + sequencing_data: Option, + execution_context: ExecutionContext, + sequencer_type: SequencerType, + ) -> anyhow::Result> { + self.standard.resolve_context( + auth_data, + sequencer, + sequencer_rollup_address, + state, + sequencing_data, + execution_context, + sequencer_type, + ) + } + + fn resolve_unregistered_context( + &mut self, + auth_data: &AuthorizationData, + sequencer: &::Address, + state: &mut impl StateAccessor, + execution_context: ExecutionContext, + ) -> anyhow::Result> { + self.standard + .resolve_unregistered_context(auth_data, sequencer, state, execution_context) + } + + fn check_uniqueness( + &self, + auth_data: &AuthorizationData, + context: &Context, + execution_context: &ExecutionContext, + state: &mut impl StateAccessor, + ) -> anyhow::Result<()> { + self.standard + .check_uniqueness(auth_data, context, execution_context, state) + } + + fn mark_tx_attempted( + &mut self, + auth_data: &AuthorizationData, + sequencer: &::Address, + state: &mut impl StateAccessor, + ) -> anyhow::Result<()> { + self.standard.mark_tx_attempted(auth_data, sequencer, state) + } +} + +impl<'a, S: Spec> ProofProcessor for RelayChainCapabilities<'a, S> { + #[cfg(feature = "native")] + type BondingProofService> = , + > as ProofProcessor>::BondingProofService; + + #[cfg(feature = "native")] + fn create_bonding_proof_service>( + &self, + attester_address: S::Address, + storage_receiver: tokio::sync::watch::Receiver, + ) -> Self::BondingProofService { + self.standard + .create_bonding_proof_service::(attester_address, storage_receiver) + } + + fn process_aggregated_proof + GetGasPrice>( + &mut self, + proof: SerializedAggregatedProof, + prover_address: &S::Address, + execution_context: ExecutionContext, + state: &mut ST, + ) -> Result< + ( + AggregatedProofPublicData::Root>, + SerializedAggregatedProof, + ), + InvalidProofError, + > { + self.standard + .process_aggregated_proof(proof, prover_address, execution_context, state) + } + + fn process_attestation + GetGasPrice>( + &mut self, + proof: SerializedAttestation, + prover_address: &S::Address, + state: &mut ST, + ) -> Result, InvalidProofError> { + self.standard + .process_attestation(proof, prover_address, state) + } + + fn process_challenge + GetGasPrice>( + &mut self, + proof: SerializedChallenge, + rollup_height: sov_rollup_interface::common::SlotNumber, + prover_address: &S::Address, + state: &mut ST, + ) -> Result, InvalidProofError> { + self.standard + .process_challenge(proof, rollup_height, prover_address, state) + } +} + +impl SequencerRemuneration for RelayChainCapabilities<'_, S> { + fn reward_sequencer_or_refund< + Accessor: StateReader + + StateWriter + + StateWriter + + StateReader, + >( + &mut self, + sequencer: &::Address, + sequencer_rollup_address: &S::Address, + reward: SequencerReward, + state: &mut Accessor, + ) { + self.standard.reward_sequencer_or_refund( + sequencer, + sequencer_rollup_address, + reward, + state, + ); + } + + fn preferred_sequencer( + &self, + state: &mut impl InfallibleStateAccessor, + ) -> Option<::Address> { + self.standard.preferred_sequencer(state) + } +} diff --git a/crates/stf/src/lib.rs b/crates/stf/src/lib.rs index 43d1f3db..ab892b78 100644 --- a/crates/stf/src/lib.rs +++ b/crates/stf/src/lib.rs @@ -2,6 +2,8 @@ pub mod authentication; mod delegation; +#[cfg(feature = "native")] +pub mod prices; pub mod runtime; pub use runtime::*; diff --git a/crates/stf/src/prices.rs b/crates/stf/src/prices.rs new file mode 100644 index 00000000..2b9986f5 --- /dev/null +++ b/crates/stf/src/prices.rs @@ -0,0 +1,261 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::sync::{LazyLock, Mutex}; + +use bytes::Bytes; +use price_oracle::{FeedKey, PriceReports, B256}; + +static ORACLE_STORE: LazyLock> = + LazyLock::new(|| Mutex::new(OracleStore::default())); + +#[derive(Debug, PartialEq, Eq)] +pub enum InsertOutcome { + Inserted, + Stale, + Unexpected, +} + +#[derive(Debug, Default, PartialEq, Eq)] +pub struct RegisterOutcome { + pub evicted: usize, + pub feeds_diverged: bool, +} + +struct SourceFeeds { + provider_id: B256, + feeds: BTreeSet, +} + +#[derive(Default)] +struct OracleStore { + reports: BTreeMap, + source_feeds: BTreeMap, + allowed_feeds: BTreeMap>, +} + +impl OracleStore { + fn snapshot(&self) -> PriceReports { + PriceReports( + self.reports + .iter() + .map(|(key, (payload, _))| (*key, payload.clone())) + .collect(), + ) + } + + fn insert_if_newer( + &mut self, + provider_id: B256, + feed_id: B256, + payload: Vec, + order_time: u64, + ) -> InsertOutcome { + let allowed = self + .allowed_feeds + .get(&provider_id) + .is_some_and(|feeds| feeds.contains(&feed_id)); + if !allowed { + return InsertOutcome::Unexpected; + } + let key = FeedKey::new(provider_id, feed_id); + if let Some((_, existing_time)) = self.reports.get(&key) { + if *existing_time >= order_time { + return InsertOutcome::Stale; + } + } + self.reports.insert(key, (Bytes::from(payload), order_time)); + InsertOutcome::Inserted + } + + fn register( + &mut self, + source_name: &str, + provider_id: B256, + feeds: Vec, + ) -> RegisterOutcome { + self.source_feeds.insert( + source_name.to_owned(), + SourceFeeds { + provider_id, + feeds: feeds.into_iter().collect(), + }, + ); + self.recompute_allowed_feeds(provider_id); + let evicted = self.retain_allowed_feeds(); + RegisterOutcome { + evicted, + feeds_diverged: self.feeds_diverged(provider_id), + } + } + + fn remove_source(&mut self, source_name: &str) -> usize { + let Some(removed) = self.source_feeds.remove(source_name) else { + return 0; + }; + self.recompute_allowed_feeds(removed.provider_id); + self.retain_allowed_feeds() + } + + fn recompute_allowed_feeds(&mut self, provider_id: B256) { + let union: BTreeSet = self + .source_feeds + .values() + .filter(|entry| entry.provider_id == provider_id) + .flat_map(|entry| entry.feeds.iter().copied()) + .collect(); + if union.is_empty() { + self.allowed_feeds.remove(&provider_id); + } else { + self.allowed_feeds.insert(provider_id, union); + } + } + + fn retain_allowed_feeds(&mut self) -> usize { + let keep: BTreeSet = self + .allowed_feeds + .iter() + .flat_map(|(provider_id, feeds)| { + feeds + .iter() + .map(|feed_id| FeedKey::new(*provider_id, *feed_id)) + }) + .collect(); + let before = self.reports.len(); + self.reports.retain(|key, _| keep.contains(key)); + before - self.reports.len() + } + + fn feeds_diverged(&self, provider_id: B256) -> bool { + let mut sets = self + .source_feeds + .values() + .filter(|entry| entry.provider_id == provider_id) + .map(|entry| &entry.feeds); + let Some(first) = sets.next() else { + return false; + }; + sets.any(|set| set != first) + } +} + +fn store() -> std::sync::MutexGuard<'static, OracleStore> { + ORACLE_STORE + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) +} + +pub fn snapshot_prices() -> PriceReports { + store().snapshot() +} + +pub fn insert_if_newer( + provider_id: B256, + feed_id: B256, + payload: Vec, + order_time: u64, +) -> InsertOutcome { + store().insert_if_newer(provider_id, feed_id, payload, order_time) +} + +pub fn register_feeds(source_name: &str, provider_id: B256, feeds: Vec) -> RegisterOutcome { + store().register(source_name, provider_id, feeds) +} + +pub fn remove_source(source_name: &str) -> usize { + store().remove_source(source_name) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn provider(byte: u8) -> B256 { + B256::repeat_byte(byte) + } + + fn feed(byte: u8) -> B256 { + B256::repeat_byte(byte) + } + + #[test] + fn rejects_unknown_feeds() { + let mut store = OracleStore::default(); + store.register("chainlink-1", provider(0x01), vec![feed(0xaa)]); + assert_eq!( + store.insert_if_newer(provider(0x01), feed(0xbb), vec![1], 100), + InsertOutcome::Unexpected + ); + assert!(store.reports.is_empty()); + } + + #[test] + fn keeps_newest() { + let mut store = OracleStore::default(); + store.register("chainlink-1", provider(0x01), vec![feed(0xaa)]); + assert_eq!( + store.insert_if_newer(provider(0x01), feed(0xaa), vec![1], 100), + InsertOutcome::Inserted + ); + assert_eq!( + store.insert_if_newer(provider(0x01), feed(0xaa), vec![2], 90), + InsertOutcome::Stale + ); + assert_eq!( + store.insert_if_newer(provider(0x01), feed(0xaa), vec![3], 100), + InsertOutcome::Stale + ); + assert_eq!( + store.insert_if_newer(provider(0x01), feed(0xaa), vec![4], 110), + InsertOutcome::Inserted + ); + let snapshot = store.snapshot(); + assert_eq!( + snapshot + .get(&FeedKey::new(provider(0x01), feed(0xaa))) + .unwrap(), + &Bytes::from(vec![4]) + ); + } + + #[test] + fn unions_replica_feeds() { + let mut store = OracleStore::default(); + let out = store.register("chainlink-1", provider(0x01), vec![feed(0xaa), feed(0xbb)]); + assert!(!out.feeds_diverged); + let out = store.register("chainlink-2", provider(0x01), vec![feed(0xaa)]); + assert!(out.feeds_diverged); + assert_eq!( + store.insert_if_newer(provider(0x01), feed(0xbb), vec![1], 100), + InsertOutcome::Inserted + ); + } + + #[test] + fn evicts_when_all_replicas_drop() { + let mut store = OracleStore::default(); + store.register("chainlink-1", provider(0x01), vec![feed(0xaa), feed(0xbb)]); + store.register("chainlink-2", provider(0x01), vec![feed(0xaa), feed(0xbb)]); + store.insert_if_newer(provider(0x01), feed(0xbb), vec![1], 100); + + let out = store.register("chainlink-1", provider(0x01), vec![feed(0xaa)]); + assert_eq!(out.evicted, 0); + assert!(store + .reports + .contains_key(&FeedKey::new(provider(0x01), feed(0xbb)))); + + let out = store.register("chainlink-2", provider(0x01), vec![feed(0xaa)]); + assert_eq!(out.evicted, 1); + assert!(!store + .reports + .contains_key(&FeedKey::new(provider(0x01), feed(0xbb)))); + } + + #[test] + fn remove_source_evicts() { + let mut store = OracleStore::default(); + store.register("chainlink-1", provider(0x01), vec![feed(0xaa)]); + store.insert_if_newer(provider(0x01), feed(0xaa), vec![1], 100); + assert_eq!(store.remove_source("chainlink-1"), 1); + assert!(store.reports.is_empty()); + assert!(store.allowed_feeds.is_empty()); + } +} diff --git a/crates/stf/stf-declaration/Cargo.toml b/crates/stf/stf-declaration/Cargo.toml index 4f3828f8..c010eb15 100644 --- a/crates/stf/stf-declaration/Cargo.toml +++ b/crates/stf/stf-declaration/Cargo.toml @@ -31,6 +31,7 @@ sov-rollup-apis = { workspace = true, optional = true } sov-hyperlane-integration = { workspace = true } sov-evm = { workspace = true } sov-address = { workspace = true } +price-oracle = { workspace = true } tracing = { workspace = true } value-setter = { workspace = true } @@ -61,8 +62,6 @@ native = [ "sov-rollup-interface/native", "sov-state/native", "sov-sequencer-registry/native", - "sov-capabilities/native", - "sov-state/native", "value-setter/native", "sov-rollup-apis", "sov-kernels/native", @@ -72,7 +71,8 @@ native = [ "sov-revenue-share/native", "sov-address/native", "sov-evm/native", - "sov-test-state-consistency/native" + "sov-test-state-consistency/native", + "price-oracle/native" ] bench = [ "sov-modules-api/bench", diff --git a/crates/stf/stf-declaration/src/lib.rs b/crates/stf/stf-declaration/src/lib.rs index c07df655..cb7dc604 100644 --- a/crates/stf/stf-declaration/src/lib.rs +++ b/crates/stf/stf-declaration/src/lib.rs @@ -16,6 +16,12 @@ use sov_modules_api::{DispatchCall, Event, Genesis, Hooks, MessageCodec, Spec}; pub type Mailbox = RawMailbox>; +sov_evm::generate_precompile_set! { + pub struct RelayEvmPrecompiles { + price_oracle: price_oracle::PriceOraclePrecompile, + } +} + /// The runtime defines the logic of the rollup. /// /// At a high level, the rollup node receives serialized "call messages" from the DA layer and executes them as atomic transactions. @@ -34,14 +40,14 @@ pub type Mailbox = RawMailbox>; /// Runtime lifecycle: /// /// 1. Initialization: -/// When a rollup is deployed for the first time, it needs to set its genesis state. -/// The `#[derive(Genesis)]` macro will generate a `Runtime::genesis(config)` method which returns -/// `Storage` with the initialized state. +/// When a rollup is deployed for the first time, it needs to set its genesis state. +/// The `#[derive(Genesis)]` macro will generate a `Runtime::genesis(config)` method which returns +/// `Storage` with the initialized state. /// /// 2. Calls: -/// The `Module` interface defines a `call` method which accepts a module-defined type and triggers the specific `module logic.` -/// In general, the point of a call is to change the module state, but if the call throws an error, -/// no state is updated (the transaction is reverted). +/// The `Module` interface defines a `call` method which accepts a module-defined type and triggers the specific `module logic.` +/// In general, the point of a call is to change the module state, but if the call throws an error, +/// no state is updated (the transaction is reverted). /// /// `#[derive(MessageCodec)]` adds deserialization capabilities to the `Runtime` (by implementing the `decode_call` method). /// `Runtime::decode_call` accepts a serialized call message and returns a type that implements the `DispatchCall` trait. @@ -89,5 +95,5 @@ where pub state_consistency: sov_test_state_consistency::StateConsistency, #[cfg_attr(feature = "native", cli_skip)] /// The EVM module. - pub evm: sov_evm::Evm, + pub evm: sov_evm::Evm>, }