diff --git a/modules/twap-monitor/module.toml b/modules/twap-monitor/module.toml index e53a15e4..519410fb 100644 --- a/modules/twap-monitor/module.toml +++ b/modules/twap-monitor/module.toml @@ -23,6 +23,11 @@ optional = [] # `http` calls. allow = [] +# ComposableCoW registry the poll path eth_calls; must match the +# chain-log subscription addresses below. +[config] +registry = "0xfdaFc9d1902f4e0b84f65F49f244b32b31013b74" + # --- subscriptions ------------------------------------------------------ # ComposableCoW.ConditionalOrderCreated emissions on Sepolia. topic-0 = diff --git a/modules/twap-monitor/src/keeper.rs b/modules/twap-monitor/src/keeper.rs index c56b7979..001f4152 100644 --- a/modules/twap-monitor/src/keeper.rs +++ b/modules/twap-monitor/src/keeper.rs @@ -5,13 +5,17 @@ //! [`CowClient`] over [`VenueTransport`]. Gate discipline, the //! `submitted:` journal, and retry dispatch live in //! `composable_cow::run`. +//! +//! `[config]` keys: `registry` (required, ComposableCoW address the +//! poll path eth_calls). + +use std::sync::{PoisonError, RwLock}; use alloy_primitives::{Address, B256, Bytes, keccak256}; use alloy_sol_types::{SolCall, SolEvent, SolValue}; use composable_cow::{LegacyRevertAdapter, Verdict, run}; use cow_venue::CowClient; use cowprotocol::{ - COMPOSABLE_COW, ComposableCoW::{ConditionalOrderCreated, ConditionalOrderRemoved}, ConditionalOrderParams, GPv2OrderData, }; @@ -28,6 +32,63 @@ pub struct BlockInfo { pub timestamp: u64, } +/// Parsed `[config]`: the ComposableCoW registry the poll path calls. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct KeeperConfig { + pub registry: Address, +} + +impl KeeperConfig { + /// Parse the wire config table. Unknown keys are ignored; a + /// missing or malformed registry is a hard error. + pub fn parse(config: &[(String, String)]) -> Result { + let invalid = |key: &str, value: &str| { + Fault::InvalidInput(format!("config {key} is invalid: {value}")) + }; + let mut registry = None; + for (key, value) in config { + if key == "registry" { + registry = Some(value.parse::
().map_err(|_| invalid(key, value))?); + } + } + let registry = registry + .ok_or_else(|| Fault::InvalidInput("config requires a registry address".to_owned()))?; + Ok(Self { registry }) + } +} + +/// Configured keeper state; `init` replaces it whole. +static CONFIG: RwLock> = RwLock::new(None); + +pub fn store_config(config: KeeperConfig) { + *CONFIG.write().unwrap_or_else(PoisonError::into_inner) = Some(config); +} + +/// The stored config, or a typed refusal when `init` has not run. +fn config() -> Result { + CONFIG + .read() + .unwrap_or_else(PoisonError::into_inner) + .ok_or_else(|| Fault::Unavailable("keeper not initialised".to_owned())) +} + +#[cfg(test)] +pub(crate) fn stored_config() -> Option { + *CONFIG.read().unwrap_or_else(PoisonError::into_inner) +} + +#[cfg(test)] +pub(crate) fn clear_config() { + *CONFIG.write().unwrap_or_else(PoisonError::into_inner) = None; +} + +/// Serialises tests touching the `CONFIG` static. +#[cfg(test)] +pub(crate) fn config_test_guard() -> std::sync::MutexGuard<'static, ()> { + static LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + LOCK.lock().unwrap_or_else(PoisonError::into_inner) +} + mod abi { use alloy_sol_types::sol; @@ -76,12 +137,13 @@ where H: ChainHost + LocalStoreHost, T: VenueTransport, { + let registry = config()?.registry; let tick = Tick { chain_id: block.chain_id, block: block.number, epoch_s: block.timestamp / 1000, }; - run(host, venue, &TwapSource, &tick) + run(host, venue, &TwapSource { registry }, &tick) } /// Chain position of a mined log, ordered as the chain orders logs. @@ -208,9 +270,12 @@ fn remove_watch( } /// TWAP conditional source: decode the stored row and evaluate -/// `getTradeableOrderWithSignature` on chain. An undecodable row polls -/// again next block rather than tearing down the run. -struct TwapSource; +/// `getTradeableOrderWithSignature` on the configured registry. An +/// undecodable row polls again next block rather than tearing down the +/// run. +struct TwapSource { + registry: Address, +} impl Poller for TwapSource { type Outcome = Verdict; @@ -231,7 +296,7 @@ impl Poller for TwapSource { ); return Verdict::TryNextBlock { reason: [0; 4] }; }; - let outcome = poll_one(host, tick.chain_id, &owner, ¶ms); + let outcome = poll_one(host, &self.registry, tick.chain_id, &owner, ¶ms); tracing::info!("poll {} -> {}", watch.key(), outcome_label(&outcome)); outcome } @@ -243,6 +308,7 @@ impl Poller for TwapSource { fn poll_one( host: &H, + registry: &Address, chain_id: u64, owner: &Address, params: &ConditionalOrderParams, @@ -257,7 +323,7 @@ fn poll_one( offchainInput: Bytes::new(), proof: Vec::new(), }; - let params_json = eth_call_params(&COMPOSABLE_COW, &call.abi_encode()); + let params_json = eth_call_params(registry, &call.abi_encode()); match host.request(chain_id, "eth_call", ¶ms_json) { Ok(result_json) => parse_eth_call_result(&result_json) .and_then(|bytes| decode_return(&bytes)) @@ -365,6 +431,9 @@ mod tests { const SEPOLIA: u64 = 11_155_111; + /// The registry pinned in module.toml. + const REGISTRY: Address = address!("fdaFc9d1902f4e0b84f65F49f244b32b31013b74"); + /// Scripted [`VenueTransport`]: one submit outcome per queued entry. /// Quote, status, and cancel are off the poll path. #[derive(Default)] @@ -420,11 +489,24 @@ mod tests { } } - /// Dispatch one block through `on_block` over the scripted transport. - fn dispatch(host: &MockHost, venue: &MockVenue, block: BlockInfo) -> Result<(), Fault> { + /// Dispatch one block through `on_block` with `registry` stored, + /// under the config guard. + fn dispatch_at( + host: &MockHost, + venue: &MockVenue, + registry: Address, + block: BlockInfo, + ) -> Result<(), Fault> { + let _guard = config_test_guard(); + store_config(KeeperConfig { registry }); on_block(host, &CowClient::with_transport(venue), block) } + /// [`dispatch_at`] pinned to the manifest registry. + fn dispatch(host: &MockHost, venue: &MockVenue, block: BlockInfo) -> Result<(), Fault> { + dispatch_at(host, venue, REGISTRY, block) + } + /// `validTo` `seconds` from the wall clock, saturating. fn valid_to_in(seconds: u32) -> u32 { let now = std::time::SystemTime::now() @@ -496,7 +578,7 @@ mod tests { b256!("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa").to_vec(), ]; let log: Log = nexum_sdk::events::ChainLogParts { - address: COMPOSABLE_COW.as_slice(), + address: REGISTRY.as_slice(), topics: &topics, ..Default::default() } @@ -507,7 +589,7 @@ mod tests { #[test] fn rejects_empty_topics() { let log: Log = nexum_sdk::events::ChainLogParts { - address: COMPOSABLE_COW.as_slice(), + address: REGISTRY.as_slice(), ..Default::default() } .into(); @@ -556,7 +638,7 @@ mod tests { owner_topic.extend_from_slice(owner.as_slice()); let topics = vec![topic0.to_vec(), owner_topic]; nexum_sdk::events::ChainLogParts { - address: COMPOSABLE_COW.as_slice(), + address: REGISTRY.as_slice(), topics: &topics, data, block_number: Some(position.block), @@ -587,7 +669,11 @@ mod tests { } /// Build the `params_json` `poll_one` passes to `host.request`. - fn programmed_eth_call_params(owner: Address, params: &ConditionalOrderParams) -> String { + fn programmed_eth_call_params_at( + registry: &Address, + owner: Address, + params: &ConditionalOrderParams, + ) -> String { let call = abi::getTradeableOrderWithSignatureCall { owner, params: abi::Params { @@ -598,7 +684,12 @@ mod tests { offchainInput: Bytes::new(), proof: Vec::new(), }; - eth_call_params(&COMPOSABLE_COW, &call.abi_encode()) + eth_call_params(registry, &call.abi_encode()) + } + + /// [`programmed_eth_call_params_at`] pinned to the manifest registry. + fn programmed_eth_call_params(owner: Address, params: &ConditionalOrderParams) -> String { + programmed_eth_call_params_at(®ISTRY, owner, params) } /// JSON-encode a hex blob as a JSON-RPC `result` field. @@ -1225,4 +1316,111 @@ mod tests { "module.toml chain-log topics and the sol! decoder topic-0s have diverged", ); } + + /// The `[config]` registry and every chain-log subscription + /// address pin must be one address. + #[test] + fn manifest_registry_matches_the_subscription_address_pins() { + let manifest: toml::Value = + toml::from_str(include_str!("../module.toml")).expect("module.toml parses"); + let registry: Address = manifest["config"]["registry"] + .as_str() + .expect("module.toml pins a [config] registry") + .parse() + .expect("registry parses as an address"); + let pins: Vec
= manifest["subscription"] + .as_array() + .expect("module.toml declares subscriptions") + .iter() + .filter(|sub| sub.get("kind").and_then(toml::Value::as_str) == Some("chain-log")) + .map(|sub| { + sub.get("address") + .and_then(toml::Value::as_str) + .expect("every chain-log subscription pins an address") + .parse() + .expect("subscription address parses") + }) + .collect(); + assert!(!pins.is_empty(), "chain-log subscriptions exist"); + for pin in pins { + assert_eq!( + pin, registry, + "module.toml [config] registry and a subscription address have diverged", + ); + } + assert_eq!( + registry, REGISTRY, + "module.toml [config] registry and the test fixture const have diverged", + ); + } + + #[test] + fn config_parses_the_registry_and_ignores_unknown_keys() { + let pairs = [ + ("name".to_owned(), "twap".to_owned()), + ("registry".to_owned(), format!("{REGISTRY:#x}")), + ]; + let parsed = KeeperConfig::parse(&pairs).expect("registry parses"); + assert_eq!(parsed.registry, REGISTRY); + } + + #[test] + fn config_refuses_a_missing_registry() { + assert!(matches!( + KeeperConfig::parse(&[]), + Err(Fault::InvalidInput(message)) + if message == "config requires a registry address", + )); + } + + #[test] + fn config_refuses_a_malformed_registry() { + let pairs = [("registry".to_owned(), "0xnope".to_owned())]; + assert!(matches!( + KeeperConfig::parse(&pairs), + Err(Fault::InvalidInput(message)) + if message == "config registry is invalid: 0xnope", + )); + } + + /// Clears the stored config under the config guard. + #[test] + fn on_block_without_stored_config_is_a_typed_refusal() { + let _guard = config_test_guard(); + clear_config(); + let host = MockHost::new(); + let venue = MockVenue::default(); + let err = on_block(&host, &CowClient::with_transport(&venue), sample_block(1)) + .expect_err("uninitialised keeper refuses the dispatch"); + assert!(matches!(err, Fault::Unavailable(_))); + assert_eq!(host.chain.call_count(), 0); + assert_eq!(venue.submit_count(), 0); + } + + /// A registry distinct from the manifest pin reaches the eth_call + /// `to`. + #[test] + fn poll_targets_the_configured_registry() { + let host = MockHost::new(); + let venue = MockVenue::default(); + let owner = address!("0011223344556677889900AABBCCDDEEFF001122"); + let params = sample_params(); + seed_watch(&host, owner, ¶ms); + let other = Address::repeat_byte(0xab); + + let ready_order = submittable_order(); + let signature: Bytes = hex!("c0ffeec0ffeec0ffee").to_vec().into(); + let wire = (ready_order, signature).abi_encode_params(); + let programmed = programmed_eth_call_params_at(&other, owner, ¶ms); + host.chain + .respond_to("eth_call", programmed.clone(), Ok(quoted_hex(&wire))); + venue.enqueue_submit(Ok(SubmitOutcome::Accepted(hex!("feedface").to_vec()))); + + dispatch_at(&host, &venue, other, sample_block(1_000)).unwrap(); + + let call = host.chain.last_call().expect("one eth_call"); + assert_eq!(call.params, programmed, "the call's `to` is the config"); + assert_eq!(host.chain.call_count(), 1); + assert_eq!(venue.submit_count(), 1, "the poll succeeded on that `to`"); + } } diff --git a/modules/twap-monitor/src/lib.rs b/modules/twap-monitor/src/lib.rs index 3f24ebbe..b9fba962 100644 --- a/modules/twap-monitor/src/lib.rs +++ b/modules/twap-monitor/src/lib.rs @@ -21,8 +21,12 @@ struct TwapMonitor; #[videre_sdk::keeper] impl TwapMonitor { - fn init(_config: Vec<(String, String)>) -> Result<(), Fault> { - install_tracing(); + fn init(config: Vec<(String, String)>) -> Result<(), Fault> { + // The host log sink is wasm-only; native unit tests skip it. + if cfg!(not(test)) { + install_tracing(); + } + keeper::store_config(keeper::KeeperConfig::parse(&config)?); tracing::info!("twap-monitor init"); Ok(()) } @@ -54,3 +58,39 @@ impl TwapMonitor { Ok(()) } } + +#[cfg(test)] +mod tests { + use alloy_primitives::address; + + use super::*; + + fn config_pairs(registry: &str) -> Vec<(String, String)> { + vec![("registry".to_owned(), registry.to_owned())] + } + + /// Holds the config guard: `init` writes the process-wide store. + #[test] + fn init_stores_the_configured_registry() { + let _guard = keeper::config_test_guard(); + let registry = address!("abababababababababababababababababababab"); + TwapMonitor::init(config_pairs(&format!("{registry:#x}"))).expect("init succeeds"); + assert_eq!( + keeper::stored_config(), + Some(keeper::KeeperConfig { registry }), + ); + } + + #[test] + fn init_without_a_registry_is_a_hard_error() { + let err = TwapMonitor::init(vec![]).expect_err("missing registry refuses init"); + assert!(matches!(err, Fault::InvalidInput(_))); + } + + #[test] + fn init_with_a_malformed_registry_is_a_hard_error() { + let err = + TwapMonitor::init(config_pairs("0xnope")).expect_err("malformed registry refuses init"); + assert!(matches!(err, Fault::InvalidInput(_))); + } +}