feat(cow): source the registry address from module [config]

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 =

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<Self, Fault> {
+        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::<Address>().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<Option<KeeperConfig>> = 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<KeeperConfig, Fault> {
+    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<KeeperConfig> {
+    *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<H: LocalStoreHost>(
 }
 
 /// 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<H: ChainHost> Poller<H> for TwapSource {
     type Outcome = Verdict;
@@ -231,7 +296,7 @@ impl<H: ChainHost> Poller<H> for TwapSource {
             );
             return Verdict::TryNextBlock { reason: [0; 4] };
         };
-        let outcome = poll_one(host, tick.chain_id, &owner, &params);
+        let outcome = poll_one(host, &self.registry, tick.chain_id, &owner, &params);
         tracing::info!("poll {} -> {}", watch.key(), outcome_label(&outcome));
         outcome
     }
@@ -243,6 +308,7 @@ impl<H: ChainHost> Poller<H> for TwapSource {
 
 fn poll_one<H: ChainHost>(
     host: &H,
+    registry: &Address,
     chain_id: u64,
     owner: &Address,
     params: &ConditionalOrderParams,
@@ -257,7 +323,7 @@ fn poll_one<H: ChainHost>(
         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", &params_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(&REGISTRY, 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<Address> = 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, &params);
+        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, &params);
+        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`");
+    }
 }