diff --git a/Cargo.lock b/Cargo.lock index db6e15e7..23e6f02d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5101,15 +5101,21 @@ dependencies = [ "alloy-sol-types", "cowprotocol", "nexum-sdk", + "nexum-sdk-test", "proptest", + "serde_json", + "shepherd-sdk-test", "strum", "thiserror 2.0.18", + "tracing", ] [[package]] name = "shepherd-sdk-test" version = "0.1.0" dependencies = [ + "alloy-primitives", + "cowprotocol", "nexum-sdk", "nexum-sdk-test", "serde_json", @@ -5779,8 +5785,6 @@ dependencies = [ "serde_json", "shepherd-sdk", "shepherd-sdk-test", - "strum", - "thiserror 2.0.18", "tracing", "wit-bindgen 0.59.0", ] diff --git a/crates/nexum-runtime/src/engine_config.rs b/crates/nexum-runtime/src/engine_config.rs index b93a0096..35b5d7de 100644 --- a/crates/nexum-runtime/src/engine_config.rs +++ b/crates/nexum-runtime/src/engine_config.rs @@ -209,6 +209,12 @@ const DEFAULT_HTTP_TOTAL_DEADLINE: Duration = Duration::from_secs(60); /// guest heap that has to buffer it. const DEFAULT_HTTP_RESPONSE_BODY_MAX: u64 = 16 * 1024 * 1024; +/// Default cap on one chain JSON-RPC response body (1 MiB). Large enough +/// for typical read responses (receipts, log batches, contract state), +/// while preventing a misbehaving or adversarial node from filling the +/// guest heap via a single large reply. +const DEFAULT_CHAIN_RESPONSE_MAX_BYTES: usize = 1024 * 1024; + /// Ceiling for the `[limits.http]` millisecond knobs (24 h). const HTTP_LIMIT_MS_MAX: u64 = 86_400_000; @@ -251,6 +257,9 @@ fn clamp_http_ms(ms: u64) -> Duration { /// total_deadline_ms = 60_000 /// response_body_max_bytes = 16_777_216 /// +/// [limits.chain] +/// response_body_max_bytes = 1_048_576 +/// /// [limits.logs] /// bytes_per_run = 262_144 /// runs_retained = 16 @@ -268,6 +277,9 @@ pub struct ModuleLimits { /// Outbound wasi:http limits. #[serde(default)] pub http: HttpLimitsSection, + /// Chain JSON-RPC response size limits. + #[serde(default)] + pub chain: ChainLimitsSection, /// Per-run log retention limits. #[serde(default)] pub logs: LogLimitsSection, @@ -287,6 +299,17 @@ impl ModuleLimits { self.memory_bytes.unwrap_or(DEFAULT_MEMORY_LIMIT) } + /// Resolved chain response size cap (override or default). A + /// degenerate `0` saturates to 1 byte, matching the `logs` / + /// `poison` sections' zero handling, so resolution never yields a + /// cap that rejects even an empty body. + pub fn chain_response_max_bytes(&self) -> usize { + self.chain + .response_body_max_bytes + .map(|b| (b.max(1)) as usize) + .unwrap_or(DEFAULT_CHAIN_RESPONSE_MAX_BYTES) + } + /// Resolved outbound HTTP limits (overrides or defaults). pub fn http(&self) -> OutboundHttpLimits { OutboundHttpLimits { @@ -376,6 +399,20 @@ pub struct HttpLimitsSection { pub response_body_max_bytes: Option, } +/// `[limits.chain]` chain JSON-RPC response size limit. Optional; +/// omitted values resolve to the built-in 1 MiB default. +/// +/// ```toml +/// [limits.chain] +/// response_body_max_bytes = 1_048_576 +/// ``` +#[derive(Debug, Default, Deserialize)] +pub struct ChainLimitsSection { + /// Cap on one chain JSON-RPC response body, in bytes. Named for + /// symmetry with `[limits.http].response_body_max_bytes`. + pub response_body_max_bytes: Option, +} + /// Resolved outbound HTTP limits the wasi:http gate enforces per /// request. Built by [`ModuleLimits::http`]. #[derive(Debug, Clone, Copy)] @@ -687,6 +724,42 @@ response_body_max_bytes = 1_024 assert_eq!(http.between_bytes_timeout_max, Duration::from_secs(30)); } + #[test] + fn chain_limits_default_when_absent() { + assert_eq!( + ModuleLimits::default().chain_response_max_bytes(), + 1024 * 1024, + ); + } + + #[test] + fn chain_limits_parse_with_override() { + let cfg: EngineConfig = toml::from_str( + r#" +[limits.chain] +response_body_max_bytes = 2_048 +"#, + ) + .expect("limits.chain parses"); + assert_eq!(cfg.limits.chain_response_max_bytes(), 2_048); + } + + #[test] + fn chain_limits_saturate_degenerate_zero() { + let cfg: EngineConfig = toml::from_str( + r#" +[limits.chain] +response_body_max_bytes = 0 +"#, + ) + .expect("limits.chain parses"); + assert_eq!( + cfg.limits.chain_response_max_bytes(), + 1, + "zero saturates to 1 so resolution never rejects an empty body", + ); + } + #[test] fn http_limits_saturate_degenerate_millisecond_values() { // Zero would fail every request instantly; u64::MAX would diff --git a/crates/nexum-runtime/src/host/impls/chain.rs b/crates/nexum-runtime/src/host/impls/chain.rs index 3fb67286..f9afad2a 100644 --- a/crates/nexum-runtime/src/host/impls/chain.rs +++ b/crates/nexum-runtime/src/host/impls/chain.rs @@ -24,6 +24,40 @@ fn resolve_method(method: &str) -> Result { }) } +/// Return an error if `body` exceeds `cap` bytes. The check is applied +/// host-side before the response is copied into the guest, so an +/// oversized node response cannot saturate the guest heap. +fn check_response_cap( + body: &str, + cap: usize, + chain_id: u64, + method: &str, +) -> Result<(), ChainError> { + if body.len() > cap { + tracing::warn!( + chain_id, + method, + body_bytes = body.len(), + cap_bytes = cap, + "chain response exceeds size cap - rejecting before guest copy" + ); + metrics::counter!( + "shepherd_chain_response_capped_total", + "chain_id" => chain_id.to_string(), + "method" => method.to_owned(), + ) + .increment(1); + return Err(ChainError::Fault( + crate::bindings::nexum::host::types::Fault::InvalidInput(format!( + "chain response ({} bytes) exceeds the configured cap ({} bytes)", + body.len(), + cap, + )), + )); + } + Ok(()) +} + impl nexum::host::chain::Host for HostState { async fn request( &mut self, @@ -57,7 +91,11 @@ impl nexum::host::chain::Host for HostState { .chain .request(chain, method, params) .await - .map_err(ChainError::from); + .map_err(ChainError::from) + .and_then(|body| { + check_response_cap(&body, self.chain_response_max_bytes, chain_id, name)?; + Ok(body) + }); tracing::trace!(elapsed_ms = ?start.elapsed(), "chain::request done"); let outcome = if result.is_ok() { "ok" } else { "err" }; metrics::counter!( @@ -90,10 +128,44 @@ impl nexum::host::chain::Host for HostState { // per-chain timeout, so the worst-case blocking time for a batch // is N x request_timeout_secs. tracing::debug!(chain_id, count = requests.len(), "chain::request-batch"); + let cap = self.chain_response_max_bytes; let mut out = Vec::with_capacity(requests.len()); + // The per-entry cap (inside `request`) bounds each body; this + // running total bounds the aggregate `Vec` lowered into + // guest memory in one go, so a wide batch of individually-legal + // bodies cannot saturate the guest heap either - the exact failure + // the guidance in #154 (block-range chunking via request-batch) + // would otherwise re-introduce. + let mut total_bytes: usize = 0; for req in requests { + let method = req.method.clone(); match nexum::host::chain::Host::request(self, chain_id, req.method, req.params).await { - Ok(s) => out.push(nexum::host::chain::RpcResult::Ok(s)), + Ok(s) => { + total_bytes = total_bytes.saturating_add(s.len()); + if total_bytes > cap { + tracing::warn!( + chain_id, + method = %method, + total_bytes, + cap_bytes = cap, + "chain batch aggregate exceeds size cap - rejecting entry before guest copy" + ); + metrics::counter!( + "shepherd_chain_response_capped_total", + "chain_id" => chain_id.to_string(), + "method" => method, + ) + .increment(1); + out.push(nexum::host::chain::RpcResult::Err(ChainError::Fault( + crate::bindings::nexum::host::types::Fault::InvalidInput(format!( + "batch aggregate ({total_bytes} bytes) exceeds the configured \ + cap ({cap} bytes)", + )), + ))); + } else { + out.push(nexum::host::chain::RpcResult::Ok(s)); + } + } Err(e) => out.push(nexum::host::chain::RpcResult::Err(e)), } } @@ -283,4 +355,26 @@ mod tests { )); assert!(resolved[2].is_ok()); } + + // ── response size cap tests (#154) ── + + #[test] + fn response_at_cap_is_accepted() { + let body = "x".repeat(10); + assert!( + check_response_cap(&body, 10, 1, "eth_call").is_ok(), + "body exactly at cap should pass" + ); + } + + #[test] + fn response_over_cap_returns_invalid_input() { + let body = "x".repeat(11); + let err = + check_response_cap(&body, 10, 1, "eth_call").expect_err("over-cap body should fail"); + assert!( + matches!(err, ChainError::Fault(Fault::InvalidInput(_))), + "expected InvalidInput fault, got {err:?}" + ); + } } diff --git a/crates/nexum-runtime/src/host/state.rs b/crates/nexum-runtime/src/host/state.rs index 32723fb8..848012aa 100644 --- a/crates/nexum-runtime/src/host/state.rs +++ b/crates/nexum-runtime/src/host/state.rs @@ -38,6 +38,9 @@ pub struct HostState { pub ext: T::Ext, /// `chain` backend - per-chain alloy `DynProvider` pool. pub chain: T::Chain, + /// Host-enforced cap on a single chain JSON-RPC response body. + /// Responses larger than this are rejected before they reach the guest. + pub chain_response_max_bytes: usize, /// `local-store` backend - per-module handle with pre-computed /// keccak256 namespace prefix. pub store: Handle, diff --git a/crates/nexum-runtime/src/supervisor.rs b/crates/nexum-runtime/src/supervisor.rs index 0b428d6b..085be188 100644 --- a/crates/nexum-runtime/src/supervisor.rs +++ b/crates/nexum-runtime/src/supervisor.rs @@ -176,6 +176,9 @@ struct LoadedModule { /// Cached for restart: outbound HTTP limits baked into the /// `HostState` we rebuild on each re-instantiation. http_limits: OutboundHttpLimits, + /// Cached for restart: chain response size cap baked into the + /// `HostState` we rebuild on each re-instantiation. + chain_response_max_bytes: usize, /// Set to `false` when `on_event` traps. Dead modules are /// excluded from dispatch until `next_attempt` is in the past. /// Modules whose `init` failed have `alive = false` @@ -299,6 +302,7 @@ impl Supervisor { http_limits: OutboundHttpLimits, memory_limit: usize, fuel: u64, + chain_response_max_bytes: usize, clocks: Option<&WasiClockOverride>, ) -> Result> { let namespace: &str = &run.module; @@ -351,6 +355,7 @@ impl Supervisor { log_router: router, ext: components.ext.clone(), chain: components.chain.clone(), + chain_response_max_bytes, store: module_store, }, ); @@ -436,6 +441,7 @@ impl Supervisor { limits_cfg.http(), limits_cfg.memory(), limits_cfg.fuel(), + limits_cfg.chain_response_max_bytes(), clocks, )?; let bindings = EventModule::instantiate_async(&mut store, &component, linker) @@ -508,6 +514,7 @@ impl Supervisor { init_config: config, http_allowlist: loaded_manifest.http_allowlist.clone(), http_limits: limits_cfg.http(), + chain_response_max_bytes: limits_cfg.chain_response_max_bytes(), failure_timestamps: std::collections::VecDeque::new(), poisoned: false, }) @@ -635,6 +642,7 @@ impl Supervisor { module.http_limits, module.memory_limit, module.fuel_per_event, + module.chain_response_max_bytes, clocks.as_ref(), )?; let bindings = EventModule::instantiate_async(&mut store, &module.component, &linker) diff --git a/crates/nexum-runtime/src/test_utils/harness.rs b/crates/nexum-runtime/src/test_utils/harness.rs index e06e163a..29ee4203 100644 --- a/crates/nexum-runtime/src/test_utils/harness.rs +++ b/crates/nexum-runtime/src/test_utils/harness.rs @@ -579,6 +579,94 @@ direction = "above" rt.wait().await.expect("clean shutdown"); } + /// `[limits.chain].response_body_max_bytes` is enforced on the real + /// `chain::request` path, end to end: the configured cap reaches + /// `HostState`, an over-cap node response is rejected before the guest + /// copy, and the module observes the typed `invalid-input` fault + /// instead of the body. Guards the wiring the unit tests on + /// `check_response_cap` cannot see (issue #154). + #[tokio::test] + async fn harness_enforces_chain_response_cap_on_the_request_path() { + use crate::engine_config::ChainLimitsSection; + use crate::host::component::ChainMethod; + + let Some(wasm) = module_wasm_or_skip("price_alert.wasm") else { + return; + }; + + // A syntactically valid oracle answer, ~330 bytes - far over the + // 16-byte cap below, so the module must never see it. + fn word(v: u128) -> String { + format!("{v:064x}") + } + let result = format!( + "\"0x{}{}{}{}{}\"", + word(1), + word(300_000_000_000), + word(0), + word(0), + word(1), + ); + + let builder = TestRuntime::builder(wasm) + .manifest_inline( + r#" +[module] +name = "price-alert" + +[capabilities] +required = ["logging", "chain"] + +[[subscription]] +kind = "block" +chain_id = 1 + +[config] +oracle_address = "0x694AA1769357215DE4FAC081bf1f309aDC325306" +decimals = "8" +threshold = "2500.00" +direction = "above" +"#, + ) + .limits(ModuleLimits { + chain: ChainLimitsSection { + response_body_max_bytes: Some(16), + }, + ..Default::default() + }); + builder.chain().on_method(ChainMethod::EthCall, result); + + let mut rt = builder + .launch() + .await + .expect("launch price-alert with a 16-byte chain response cap"); + + rt.push_block(header_numbered(19_000_000)); + let record = rt + .wait_for_log("price-alert", "exceeds the configured cap") + .await + .expect("the module logs the guest-visible cap fault"); + assert!( + record.message.contains("eth_call failed"), + "the cap surfaces as a failed eth_call, got: {}", + record.message, + ); + + // The module never saw the oracle answer, so it must not trigger. + let runs = rt.logs().list_runs("price-alert"); + let triggered = runs.into_iter().any(|meta| { + rt.logs() + .read(&meta.run, 0) + .records + .iter() + .any(|r| r.message.contains("TRIGGERED")) + }); + assert!(!triggered, "an over-cap response must never reach classify"); + + rt.shutdown(); + rt.wait().await.expect("clean shutdown"); + } + /// A dropped block stream is not the end of dispatch: the event loop's /// reconnect task reopens the subscription after backoff and the /// re-armed mock resumes delivery, matching a real provider that comes diff --git a/crates/nexum-sdk-test/src/lib.rs b/crates/nexum-sdk-test/src/lib.rs index 621da7bd..b9e135bb 100644 --- a/crates/nexum-sdk-test/src/lib.rs +++ b/crates/nexum-sdk-test/src/lib.rs @@ -60,9 +60,10 @@ #![cfg_attr(not(test), warn(unused_crate_dependencies))] #![warn(missing_docs)] -use std::cell::RefCell; +use std::cell::{Cell, RefCell}; use std::collections::{BTreeMap, HashMap}; use std::fmt::{self, Write as _}; +use std::rc::Rc; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; @@ -191,41 +192,99 @@ impl ChainHost for MockChain { // ---------------------------------------------------------------- local-store -/// In-memory [`LocalStoreHost`] backed by a `HashMap`. Each operation -/// runs in O(1) except `list_keys`, which scans (small N expected for -/// tests). +/// In-memory [`LocalStoreHost`] mirroring the runtime store's shape: +/// namespaced views over one shared row map, plus store-wide entry +/// and byte limits. /// -/// Supports optional error injection via [`MockLocalStore::fail_on`] -/// and entry-count limits via [`MockLocalStore::set_max_entries`]. +/// A fresh store is the root view. [`namespaced`](Self::namespaced) +/// derives a sibling view over the same backing rows - identical key +/// strings in different namespaces never collide, matching the host's +/// per-module key prefixing. Limits sit on the shared backing store, +/// so one namespace's writes can exhaust another's headroom exactly +/// as two modules share one database file. Fault injection via +/// [`fail_on`](Self::fail_on) stays per-view. #[derive(Default)] pub struct MockLocalStore { - rows: RefCell>>, - /// When set, `set` returns `StorageFull` if the store reaches this many entries. - max_entries: RefCell>, + shared: Rc, + namespace: String, /// Key patterns that trigger injected faults on any operation. error_patterns: RefCell>, } +/// Backing rows and limits shared by every namespaced view. +#[derive(Default)] +struct SharedRows { + /// Rows keyed by `(namespace, key)`. + rows: RefCell>>, + /// Total stored bytes (key + value) across all namespaces. + bytes: Cell, + /// When set, `set` on a new key fails once the store holds this + /// many rows. + max_entries: Cell>, + /// When set, `set` fails once stored bytes would exceed this. + max_bytes: Cell>, +} + impl MockLocalStore { - /// Number of rows currently held. + /// A view over the same backing rows under `namespace`. Views with + /// the same namespace alias the same data (two handles onto one + /// module store); different namespaces are fully isolated even for + /// identical key strings. + /// + /// # Panics + /// + /// On an empty namespace - the runtime rejects those too. + pub fn namespaced(&self, namespace: impl Into) -> MockLocalStore { + let namespace = namespace.into(); + assert!( + !namespace.is_empty(), + "MockLocalStore: namespace must not be empty", + ); + MockLocalStore { + shared: Rc::clone(&self.shared), + namespace, + error_patterns: RefCell::new(Vec::new()), + } + } + + /// Number of rows in this view's namespace. pub fn len(&self) -> usize { - self.rows.borrow().len() + self.shared + .rows + .borrow() + .keys() + .filter(|(ns, _)| *ns == self.namespace) + .count() } - /// Whether the store is empty. + /// Whether this view's namespace holds no rows. pub fn is_empty(&self) -> bool { - self.rows.borrow().is_empty() + self.len() == 0 } - /// Direct read for assertions - bypasses the trait. + /// Direct read of this view's namespace for assertions - bypasses + /// the trait. pub fn snapshot(&self) -> HashMap> { - self.rows.borrow().clone() + self.shared + .rows + .borrow() + .iter() + .filter(|((ns, _), _)| *ns == self.namespace) + .map(|((_, key), value)| (key.clone(), value.clone())) + .collect() } - /// Set a maximum number of entries. Once reached, `set` on a new - /// key returns a `StorageFull` error. `None` disables the limit. + /// Cap the row count across every namespace. Once reached, `set` + /// on a new key fails; overwriting an existing key still succeeds. pub fn set_max_entries(&self, limit: usize) { - *self.max_entries.borrow_mut() = Some(limit); + self.shared.max_entries.set(Some(limit)); + } + + /// Cap total stored bytes (key + value, across every namespace). + /// A `set` that would push the total past the cap fails; deletes + /// and same-key overwrites release the bytes they displace. + pub fn set_max_bytes(&self, limit: usize) { + self.shared.max_bytes.set(Some(limit)); } /// Inject a fault for any operation where the key starts with @@ -250,36 +309,64 @@ impl MockLocalStore { impl LocalStoreHost for MockLocalStore { fn get(&self, key: &str) -> Result>, Fault> { self.check_injected_error(key)?; - Ok(self.rows.borrow().get(key).cloned()) + Ok(self + .shared + .rows + .borrow() + .get(&(self.namespace.clone(), key.to_string())) + .cloned()) } fn set(&self, key: &str, value: &[u8]) -> Result<(), Fault> { self.check_injected_error(key)?; - if let Some(limit) = *self.max_entries.borrow() { - let rows = self.rows.borrow(); - if rows.len() >= limit && !rows.contains_key(key) { - return Err(Fault::Internal(format!( - "MockLocalStore: max entries ({limit}) reached" - ))); - } + let mut rows = self.shared.rows.borrow_mut(); + let compound = (self.namespace.clone(), key.to_string()); + let existing = rows.get(&compound).map(Vec::len); + if existing.is_none() + && let Some(limit) = self.shared.max_entries.get() + && rows.len() >= limit + { + return Err(Fault::Internal(format!( + "MockLocalStore: max entries ({limit}) reached" + ))); } - self.rows - .borrow_mut() - .insert(key.to_string(), value.to_vec()); + // Same-key overwrites release the displaced bytes before the + // new row is charged. + let displaced = existing.map_or(0, |len| key.len() + len); + let total = self.shared.bytes.get() - displaced + key.len() + value.len(); + if let Some(budget) = self.shared.max_bytes.get() + && total > budget + { + return Err(Fault::Internal(format!( + "MockLocalStore: max bytes ({budget}) reached" + ))); + } + rows.insert(compound, value.to_vec()); + self.shared.bytes.set(total); Ok(()) } fn delete(&self, key: &str) -> Result<(), Fault> { self.check_injected_error(key)?; - self.rows.borrow_mut().remove(key); + if let Some(value) = self + .shared + .rows + .borrow_mut() + .remove(&(self.namespace.clone(), key.to_string())) + { + self.shared + .bytes + .set(self.shared.bytes.get() - key.len() - value.len()); + } Ok(()) } fn list_keys(&self, prefix: &str) -> Result, Fault> { self.check_injected_error(prefix)?; let mut keys: Vec = self + .shared .rows .borrow() .keys() - .filter(|k| k.starts_with(prefix)) - .cloned() + .filter(|(ns, key)| *ns == self.namespace && key.starts_with(prefix)) + .map(|(_, key)| key.clone()) .collect(); keys.sort(); Ok(keys) @@ -671,6 +758,77 @@ mod tests { assert_eq!(store.len(), 2); } + #[test] + fn local_store_namespaces_isolate_identical_keys() { + let store = MockLocalStore::default(); + let other = store.namespaced("other-module"); + store.set("watch:a", b"mine").unwrap(); + other.set("watch:a", b"theirs").unwrap(); + + assert_eq!(store.get("watch:a").unwrap().as_deref(), Some(&b"mine"[..])); + assert_eq!( + other.get("watch:a").unwrap().as_deref(), + Some(&b"theirs"[..]), + ); + + // Scans, counts, and snapshots stay view-scoped. + assert_eq!(store.len(), 1); + assert_eq!(other.len(), 1); + assert_eq!(store.list_keys("").unwrap(), vec!["watch:a"]); + assert_eq!(store.snapshot().get("watch:a").unwrap(), b"mine"); + + // Deletes never reach across the namespace boundary. + other.delete("watch:a").unwrap(); + assert!(other.is_empty()); + assert_eq!(store.get("watch:a").unwrap().as_deref(), Some(&b"mine"[..])); + } + + #[test] + fn local_store_same_namespace_views_alias_the_same_rows() { + let store = MockLocalStore::default(); + let one = store.namespaced("mod"); + let two = store.namespaced("mod"); + one.set("k", b"v").unwrap(); + assert_eq!(two.get("k").unwrap().as_deref(), Some(&b"v"[..])); + } + + #[test] + #[should_panic(expected = "namespace must not be empty")] + fn local_store_empty_namespace_panics() { + let _ = MockLocalStore::default().namespaced(""); + } + + #[test] + fn local_store_entry_limit_spans_namespaces() { + let store = MockLocalStore::default(); + store.set_max_entries(2); + let other = store.namespaced("other-module"); + store.set("a", b"1").unwrap(); + other.set("b", b"2").unwrap(); + // The store is one shared file: a sibling namespace's rows + // consume the same headroom. + let err = store.set("c", b"3").unwrap_err(); + assert!(matches!(err, Fault::Internal(ref m) if m.contains("max entries"))); + } + + #[test] + fn local_store_byte_budget_enforced_and_released() { + let store = MockLocalStore::default(); + store.set_max_bytes(8); + store.set("abcd", b"1234").unwrap(); // 4 + 4 = 8, exactly at budget + let err = store.set("x", b"y").unwrap_err(); + assert!(matches!(err, Fault::Internal(ref m) if m.contains("max bytes"))); + + // A same-key overwrite releases the displaced value first. + store.set("abcd", b"12").unwrap(); + store.set("x", b"y").unwrap(); + + // Deleting releases the whole row's bytes. + store.delete("abcd").unwrap(); + store.set("ab", b"12").unwrap(); + assert_eq!(store.len(), 2); + } + #[test] fn mock_host_dispatches_through_supertrait() { let host = MockHost::new(); diff --git a/crates/nexum-sdk/src/keeper.rs b/crates/nexum-sdk/src/keeper.rs index 06d1cd9e..ff78e082 100644 --- a/crates/nexum-sdk/src/keeper.rs +++ b/crates/nexum-sdk/src/keeper.rs @@ -13,6 +13,14 @@ //! - [`Journal`] - the receipt-keyed idempotency journal of //! `submitted:` / `observed:` presence markers. //! +//! Two pieces drive the stores from the poll loop: +//! +//! - [`ConditionalSource`] - the world-neutral poll seam: one watch in, +//! one outcome out, at a given [`Tick`]. Implementations own the +//! transport and the outcome shape. +//! - [`Retrier`] - runs a [`RetryAction`]'s effect through the +//! stores after a failed run attempt. +//! //! [`WatchRef`] ties the first two together: gate keys are derived //! from the exact hex substrings of the stored watch key, and //! [`WatchSet::remove`] drops a watch together with all of its gate @@ -69,6 +77,7 @@ //! ``` use alloy_primitives::{Address, B256}; +use strum::IntoStaticStr; use crate::host::{Fault, LocalStoreHost}; @@ -292,3 +301,96 @@ impl<'h, H: LocalStoreHost> Journal<'h, H> { .is_some()) } } + +/// One poll dispatch's world view: chain, block height, and the block +/// clock in Unix seconds. Gate checks and backoff arithmetic read the +/// same instant a source is polled at, so a watch can never gate +/// itself against a clock it was not judged by. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct Tick { + /// Chain the dispatch targets. + pub chain_id: u64, + /// Block height at the tick. + pub block: u64, + /// Block timestamp, Unix seconds. + pub epoch_s: u64, +} + +/// A source of conditional commitments: poll one watch, produce one +/// outcome. Generic over the host so implementations stay mock- +/// testable; deliberately no venue-transport abstraction - the source +/// owns its own wire (an `eth_call`, an HTTP probe, a stub). +/// +/// A transient failure should surface as a retry-flavoured outcome, +/// not tear down the caller's sweep: `poll` is infallible by contract. +pub trait ConditionalSource { + /// What one poll produces. + type Outcome; + + /// Poll the source for `watch` at `tick`. `params` is the stored + /// watch value (the encoded commitment parameters), passed + /// verbatim so the source owns the decode. + fn poll(&self, host: &H, watch: WatchRef<'_>, params: &[u8], tick: &Tick) -> Self::Outcome; + + /// Short strategy name compositions prefix shared log lines with + /// (for example `"twap"`). Diagnostic only - no behaviour keys + /// off it. + fn label(&self) -> &'static str { + "conditional" + } +} + +/// What the retry ledger should do to a watch after a failed +/// run attempt. +/// +/// `IntoStaticStr` exposes each variant as a snake_case `&'static +/// str` for log and metric labels. `#[non_exhaustive]` so the +/// contract can grow a variant; downstream dispatch should treat an +/// unknown variant as "leave the watch in place" (the conservative +/// choice). +#[derive(Clone, Copy, Debug, Eq, PartialEq, IntoStaticStr)] +#[strum(serialize_all = "snake_case")] +#[non_exhaustive] +pub enum RetryAction { + /// Leave the watch untouched; the next tick re-attempts. + TryNextBlock, + /// Gate the watch until `now + seconds` on the epoch clock. + Backoff { + /// Seconds to wait before retrying. + seconds: u64, + }, + /// Remove the watch and its gates; no retry can succeed. + Drop, +} + +/// Retry ledger: runs a [`RetryAction`]'s effect through the keeper +/// stores. `Backoff` saturates at `u64::MAX` on the epoch clock; +/// `Drop` delegates to [`WatchSet::remove`], so gates go first and no +/// failure path can orphan one. +pub struct Retrier<'h, H> { + host: &'h H, +} + +impl<'h, H: LocalStoreHost> Retrier<'h, H> { + /// Ledger view over the given host. + pub fn new(host: &'h H) -> Self { + Self { host } + } + + /// Apply `action` to the watch, with `now_epoch_s` as the backoff + /// origin. + pub fn apply( + &self, + watch: WatchRef<'_>, + action: RetryAction, + now_epoch_s: u64, + ) -> Result<(), Fault> { + match action { + RetryAction::TryNextBlock => Ok(()), + RetryAction::Backoff { seconds } => { + Gates::new(self.host).set_next_epoch(watch, now_epoch_s.saturating_add(seconds)) + } + RetryAction::Drop => WatchSet::new(self.host).remove(watch), + } + } +} diff --git a/crates/nexum-sdk/src/lib.rs b/crates/nexum-sdk/src/lib.rs index ac6531fe..b11cd21e 100644 --- a/crates/nexum-sdk/src/lib.rs +++ b/crates/nexum-sdk/src/lib.rs @@ -31,7 +31,8 @@ //! - [`keeper`] - strategy-keeper stores over [`LocalStoreHost`]: //! the watch-set registry ([`WatchSet`]), block/epoch gate keys //! ([`Gates`]) and the receipt-keyed idempotency journal -//! ([`Journal`]). +//! ([`Journal`]); plus the [`ConditionalSource`] poll seam and the +//! [`Retrier`] dispatching a [`RetryAction`] through the stores. //! //! - [`chain`] - `eth_call` JSON plumbing ([`eth_call_params`], //! [`parse_eth_call_result`]) and the Chainlink AggregatorV3 reader @@ -81,6 +82,9 @@ //! [`WatchSet`]: keeper::WatchSet //! [`Gates`]: keeper::Gates //! [`Journal`]: keeper::Journal +//! [`ConditionalSource`]: keeper::ConditionalSource +//! [`Retrier`]: keeper::Retrier +//! [`RetryAction`]: keeper::RetryAction //! [`eth_call_params`]: chain::eth_call_params //! [`parse_eth_call_result`]: chain::parse_eth_call_result //! [`read_latest_answer`]: chain::chainlink::read_latest_answer diff --git a/crates/nexum-sdk/tests/keeper.rs b/crates/nexum-sdk/tests/keeper.rs index 5ac54c87..ff8f9746 100644 --- a/crates/nexum-sdk/tests/keeper.rs +++ b/crates/nexum-sdk/tests/keeper.rs @@ -7,7 +7,8 @@ use alloy_primitives::{Address, B256, address, b256}; use nexum_sdk::host::{Fault, LocalStoreHost as _}; use nexum_sdk::keeper::{ - Gates, Journal, NEXT_BLOCK_PREFIX, NEXT_EPOCH_PREFIX, WATCH_PREFIX, WatchRef, WatchSet, + ConditionalSource, Gates, Journal, NEXT_BLOCK_PREFIX, NEXT_EPOCH_PREFIX, Retrier, RetryAction, + Tick, WATCH_PREFIX, WatchRef, WatchSet, }; use shepherd_sdk_test::MockHost; @@ -351,3 +352,138 @@ fn submitted_and_observed_keyspaces_are_disjoint() { assert!(snapshot.contains_key("submitted:0xuid")); assert!(!snapshot.contains_key("observed:0xuid")); } + +// ---- retry ledger ---- + +fn seeded_watch(host: &MockHost) -> String { + WatchSet::new(host) + .put(&sample_owner(), &sample_hash(), b"params") + .unwrap() +} + +#[test] +fn ledger_try_next_block_leaves_the_store_untouched() { + let host = MockHost::new(); + let key = seeded_watch(&host); + let before = host.store.snapshot(); + + Retrier::new(&host) + .apply( + WatchRef::parse(&key).unwrap(), + RetryAction::TryNextBlock, + 1_000, + ) + .unwrap(); + + assert_eq!(host.store.snapshot(), before); +} + +#[test] +fn ledger_backoff_gates_the_watch_on_the_epoch_clock() { + let host = MockHost::new(); + let key = seeded_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + let ledger = Retrier::new(&host); + + ledger + .apply(watch, RetryAction::Backoff { seconds: 30 }, 1_000) + .unwrap(); + + let gates = Gates::new(&host); + assert!(!gates.is_ready(watch, u64::MAX, 1_029).unwrap()); + assert!(gates.is_ready(watch, u64::MAX, 1_030).unwrap()); + assert_eq!( + host.store.snapshot().get(&watch.next_epoch_key()).unwrap(), + &1_030_u64.to_le_bytes().to_vec(), + ); + assert!( + host.store.snapshot().contains_key(&key), + "backoff must keep the watch", + ); +} + +#[test] +fn ledger_backoff_saturates_on_the_epoch_clock() { + let host = MockHost::new(); + let key = seeded_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + + Retrier::new(&host) + .apply(watch, RetryAction::Backoff { seconds: u64::MAX }, 1_000) + .unwrap(); + + assert_eq!( + host.store.snapshot().get(&watch.next_epoch_key()).unwrap(), + &u64::MAX.to_le_bytes().to_vec(), + ); +} + +#[test] +fn ledger_drop_removes_the_watch_and_its_gates() { + let host = MockHost::new(); + let key = seeded_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + Gates::new(&host).set_next_block(watch, 500).unwrap(); + + Retrier::new(&host) + .apply(watch, RetryAction::Drop, 1_000) + .unwrap(); + + assert!(host.store.is_empty(), "watch and gates must go"); +} + +#[test] +fn retry_action_labels_are_stable_snake_case() { + let cases: [(RetryAction, &str); 3] = [ + (RetryAction::TryNextBlock, "try_next_block"), + (RetryAction::Backoff { seconds: 1 }, "backoff"), + (RetryAction::Drop, "drop"), + ]; + for (action, label) in cases { + assert_eq!(<&'static str>::from(action), label); + } +} + +// ---- conditional source ---- + +/// A source is generic over the host and owns its outcome shape; the +/// keeper passes the stored params verbatim and the tick it judged +/// the gates by. +#[test] +fn conditional_source_sees_params_and_tick_verbatim() { + struct EchoSource; + impl ConditionalSource for EchoSource { + type Outcome = (usize, u64, u64, u64, String); + fn poll( + &self, + _host: &H, + watch: WatchRef<'_>, + params: &[u8], + tick: &Tick, + ) -> Self::Outcome { + ( + params.len(), + tick.chain_id, + tick.block, + tick.epoch_s, + watch.key(), + ) + } + } + + let host = MockHost::new(); + let key = seeded_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + let tick = Tick { + chain_id: 1, + block: 42, + epoch_s: 1_700_000_000, + }; + + let (len, chain_id, block, epoch_s, echoed) = EchoSource.poll(&host, watch, b"params", &tick); + assert_eq!(len, b"params".len()); + assert_eq!(chain_id, 1); + assert_eq!(block, 42); + assert_eq!(epoch_s, 1_700_000_000); + assert_eq!(echoed, key); +} diff --git a/crates/shepherd-sdk-test/Cargo.toml b/crates/shepherd-sdk-test/Cargo.toml index a09c388f..83a3525c 100644 --- a/crates/shepherd-sdk-test/Cargo.toml +++ b/crates/shepherd-sdk-test/Cargo.toml @@ -15,3 +15,9 @@ nexum-sdk = { path = "../nexum-sdk" } nexum-sdk-test = { path = "../nexum-sdk-test" } shepherd-sdk = { path = "../shepherd-sdk" } serde_json = { workspace = true, features = ["std"] } + +[dev-dependencies] +# Order construction for the MockVenue acceptance tests that drive +# the keeper run end to end. +alloy-primitives.workspace = true +cowprotocol = { version = "0.2.0", default-features = false } diff --git a/crates/shepherd-sdk-test/src/lib.rs b/crates/shepherd-sdk-test/src/lib.rs index e1289cd0..2e08b74f 100644 --- a/crates/shepherd-sdk-test/src/lib.rs +++ b/crates/shepherd-sdk-test/src/lib.rs @@ -23,6 +23,23 @@ //! assert_eq!(host.cow_api.call_count(), 1); //! ``` //! +//! Per-call venue scripting - outcome queues, status sequences, fault +//! injection - goes through [`MockVenue`] on the same seam: +//! +//! ```rust +//! use nexum_sdk::host::Fault; +//! use shepherd_sdk::cow::{CowApiError, CowApiHost as _}; +//! use shepherd_sdk_test::MockHost; +//! +//! let host = MockHost::with_venue(); +//! host.cow_api +//! .enqueue_submit(Err(CowApiError::Fault(Fault::Timeout))); +//! host.cow_api.enqueue_submit(Ok("0xuid".into())); +//! +//! assert!(host.submit_order(1, b"{}").is_err()); +//! assert_eq!(host.submit_order(1, b"{}").unwrap(), "0xuid"); +//! ``` +//! //! Modules that never touch the orderbook use `nexum-sdk-test`'s //! `MockHost` directly instead. @@ -30,6 +47,7 @@ #![warn(missing_docs)] use std::cell::RefCell; +use std::collections::{HashMap, VecDeque}; use nexum_sdk::Level; use nexum_sdk::host::{ChainError, ChainHost, Fault, LocalStoreHost, LoggingHost}; @@ -37,16 +55,18 @@ use nexum_sdk_test::{MockChain, MockLocalStore, MockLogging}; use shepherd_sdk::cow::{CowApiError, CowApiHost}; /// Composed in-memory host for CoW modules: the generic per-trait -/// mocks plus [`MockCowApi`]. Each field exposes the per-trait mock so -/// tests can program responses and assert on calls. +/// mocks plus a venue mock on the `shepherd:cow/cow-api` seam - +/// [`MockCowApi`] by default, [`MockVenue`] via +/// [`with_venue`](MockHost::with_venue). Each field exposes the +/// per-trait mock so tests can program responses and assert on calls. #[derive(Default)] -pub struct MockHost { +pub struct MockHost { /// `nexum:host/chain` mock. pub chain: MockChain, /// `nexum:host/local-store` mock. pub store: MockLocalStore, /// `shepherd:cow/cow-api` mock. - pub cow_api: MockCowApi, + pub cow_api: V, /// `nexum:host/logging` mock. pub logging: MockLogging, } @@ -58,13 +78,20 @@ impl MockHost { } } -impl ChainHost for MockHost { +impl MockHost { + /// Fresh empty host with [`MockVenue`] on the cow-api seam. + pub fn with_venue() -> Self { + Self::default() + } +} + +impl ChainHost for MockHost { fn request(&self, chain_id: u64, method: &str, params: &str) -> Result { self.chain.request(chain_id, method, params) } } -impl LocalStoreHost for MockHost { +impl LocalStoreHost for MockHost { fn get(&self, key: &str) -> Result>, Fault> { self.store.get(key) } @@ -79,7 +106,7 @@ impl LocalStoreHost for MockHost { } } -impl CowApiHost for MockHost { +impl CowApiHost for MockHost { fn submit_order(&self, chain_id: u64, body: &[u8]) -> Result { self.cow_api.submit_order(chain_id, body) } @@ -94,7 +121,7 @@ impl CowApiHost for MockHost { } } -impl LoggingHost for MockHost { +impl LoggingHost for MockHost { fn log(&self, level: Level, message: &str) { self.logging.log(level, message); } @@ -240,6 +267,181 @@ impl CowApiHost for MockCowApi { } } +// ---------------------------------------------------------------- venue + +/// Scripted in-memory venue on the [`CowApiHost`] seam: programmable +/// per-call behaviour, unlike [`MockCowApi`]'s single replayed +/// response. Compose it with the generic mocks via +/// [`MockHost::with_venue`]. +/// +/// The two queue disciplines differ deliberately. Submissions are +/// discrete effects, so the submit queue strictly drains - one outcome +/// per call, then the configured fallback (default: an `Unsupported` +/// fault), so a test that scripts N outcomes catches an unexpected +/// N+1th submit. Responses are observations, so a `(method, path)` +/// sequence advances per call and its final entry replays forever - a +/// terminal order status persists no matter how often it is re-polled. +/// An injected fault overrides both (without consuming the queues) +/// until cleared, modelling a venue outage. +#[derive(Default)] +pub struct MockVenue { + submit_queue: RefCell>, + submit_fallback: RefCell>, + response_sequences: RefCell>>, + response_fallback: RefCell>, + fault: RefCell>, + calls: RefCell>, + request_calls: RefCell>, +} + +/// One scripted venue reply: the body / UID on success, a typed +/// [`CowApiError`] otherwise. +type VenueOutcome = Result; + +impl MockVenue { + /// Append one `submit_order` outcome to the queue; each call + /// consumes one, in order. + pub fn enqueue_submit(&self, outcome: Result) { + self.submit_queue.borrow_mut().push_back(outcome); + } + + /// Steady-state `submit_order` response once the queue is drained. + /// Unset, a drained queue yields an `Unsupported` fault. + pub fn set_submit_fallback(&self, outcome: Result) { + *self.submit_fallback.borrow_mut() = Some(outcome); + } + + /// Append one outcome to the `(method, path)` response sequence. + /// Each matching `cow_api_request` call advances the sequence; the + /// final entry sticks. + pub fn enqueue_response( + &self, + method: impl Into, + path: impl Into, + outcome: Result, + ) { + self.response_sequences + .borrow_mut() + .entry((method.into(), path.into())) + .or_default() + .push_back(outcome); + } + + /// Append one status-probe outcome for the order, keyed on the + /// orderbook's `GET /api/v1/orders/{uid}` route. + pub fn enqueue_order_status(&self, uid: &str, outcome: Result) { + self.enqueue_response("GET", format!("/api/v1/orders/{uid}"), outcome); + } + + /// Catch-all `cow_api_request` response for calls with no + /// programmed sequence. Unset, those yield an `Unsupported` fault. + pub fn set_response_fallback(&self, outcome: Result) { + *self.response_fallback.borrow_mut() = Some(outcome); + } + + /// Fail every venue call with `err` until + /// [`clear_fault`](Self::clear_fault) - a scripted outage. Queued + /// outcomes are not consumed while the fault is active. + pub fn inject_fault(&self, err: CowApiError) { + *self.fault.borrow_mut() = Some(err); + } + + /// Lift an injected fault; queued outcomes resume where they left + /// off. + pub fn clear_fault(&self) { + *self.fault.borrow_mut() = None; + } + + /// All submissions, in arrival order. + pub fn calls(&self) -> Vec { + self.calls.borrow().clone() + } + + /// Last submission, if any. + pub fn last_call(&self) -> Option { + self.calls.borrow().last().cloned() + } + + /// Convenience: parse the most recent submission body as JSON. + pub fn last_body_as_json(&self) -> Option { + self.last_call() + .and_then(|c| serde_json::from_slice(&c.body).ok()) + } + + /// Count of submissions (failed and injected-fault calls included). + pub fn call_count(&self) -> usize { + self.calls.borrow().len() + } + + /// All `cow_api_request` invocations, in arrival order. + pub fn request_calls(&self) -> Vec { + self.request_calls.borrow().clone() + } + + /// Scripted submit outcomes not yet consumed - assert `0` to prove + /// a scenario played out in full. + pub fn pending_submits(&self) -> usize { + self.submit_queue.borrow().len() + } +} + +impl CowApiHost for MockVenue { + fn submit_order(&self, chain_id: u64, body: &[u8]) -> Result { + self.calls.borrow_mut().push(SubmitCall { + chain_id, + body: body.to_vec(), + }); + if let Some(err) = self.fault.borrow().as_ref() { + return Err(err.clone()); + } + if let Some(outcome) = self.submit_queue.borrow_mut().pop_front() { + return outcome; + } + self.submit_fallback.borrow().clone().unwrap_or_else(|| { + Err(CowApiError::Fault(Fault::Unsupported( + "MockVenue: submit queue exhausted and no fallback configured".to_string(), + ))) + }) + } + + fn cow_api_request( + &self, + chain_id: u64, + method: &str, + path: &str, + body: Option<&str>, + ) -> Result { + self.request_calls.borrow_mut().push(RequestCall { + chain_id, + method: method.to_string(), + path: path.to_string(), + body: body.map(str::to_string), + }); + if let Some(err) = self.fault.borrow().as_ref() { + return Err(err.clone()); + } + if let Some(sequence) = self + .response_sequences + .borrow_mut() + .get_mut(&(method.to_string(), path.to_string())) + { + // Advance until one entry remains, then replay it: the + // sequence's final state persists. + if sequence.len() > 1 { + return sequence.pop_front().expect("length checked above"); + } + if let Some(last) = sequence.front() { + return last.clone(); + } + } + self.response_fallback.borrow().clone().unwrap_or_else(|| { + Err(CowApiError::Fault(Fault::Unsupported( + "MockVenue: no response programmed for this request".to_string(), + ))) + }) + } +} + #[cfg(test)] mod tests { use super::*; @@ -266,6 +468,119 @@ mod tests { ); } + // ---- MockVenue ---- + + #[test] + fn venue_submit_queue_drains_in_order_then_falls_back() { + let venue = MockVenue::default(); + venue.enqueue_submit(Err(CowApiError::Fault(Fault::Timeout))); + venue.enqueue_submit(Ok("0xuid".into())); + + assert!(matches!( + venue.submit_order(1, b"{}"), + Err(CowApiError::Fault(Fault::Timeout)), + )); + assert_eq!(venue.submit_order(1, b"{}").unwrap(), "0xuid"); + assert_eq!(venue.pending_submits(), 0); + + // A drained queue is unsupported by default: an unscripted + // extra submit fails loudly. + assert!(matches!( + venue.submit_order(1, b"{}"), + Err(CowApiError::Fault(Fault::Unsupported(_))), + )); + + venue.set_submit_fallback(Ok("0xsteady".into())); + assert_eq!(venue.submit_order(1, b"{}").unwrap(), "0xsteady"); + assert_eq!(venue.call_count(), 4, "every call is recorded"); + } + + #[test] + fn venue_records_submissions_like_the_single_shot_mock() { + let venue = MockVenue::default(); + venue.enqueue_submit(Ok("0xuid".into())); + venue.submit_order(7, b"{\"x\":1}").unwrap(); + + let last = venue.last_call().unwrap(); + assert_eq!(last.chain_id, 7); + assert_eq!(last.body, b"{\"x\":1}"); + assert_eq!(venue.last_body_as_json().unwrap()["x"], 1); + } + + #[test] + fn venue_fault_injection_overrides_queues_until_cleared() { + let venue = MockVenue::default(); + venue.enqueue_submit(Ok("0xuid".into())); + venue.enqueue_response("GET", "/api/v1/orders/0x1", Ok("{}".into())); + venue.inject_fault(CowApiError::Fault(Fault::Unavailable("down".into()))); + + assert!(matches!( + venue.submit_order(1, b"{}"), + Err(CowApiError::Fault(Fault::Unavailable(_))), + )); + assert!( + venue + .cow_api_request(1, "GET", "/api/v1/orders/0x1", None) + .is_err() + ); + + // The outage consumed nothing: outcomes resume on recovery. + venue.clear_fault(); + assert_eq!(venue.submit_order(1, b"{}").unwrap(), "0xuid"); + assert_eq!( + venue + .cow_api_request(1, "GET", "/api/v1/orders/0x1", None) + .unwrap(), + "{}", + ); + assert_eq!(venue.call_count(), 2); + assert_eq!(venue.request_calls().len(), 2); + } + + #[test] + fn venue_response_sequence_advances_and_final_entry_sticks() { + let venue = MockVenue::default(); + for body in ["\"open\"", "\"open\"", "\"fulfilled\""] { + venue.enqueue_order_status("0xuid", Ok(body.into())); + } + let probe = || { + venue + .cow_api_request(1, "GET", "/api/v1/orders/0xuid", None) + .unwrap() + }; + assert_eq!(probe(), "\"open\""); + assert_eq!(probe(), "\"open\""); + assert_eq!(probe(), "\"fulfilled\""); + // The terminal entry replays for any later re-poll. + assert_eq!(probe(), "\"fulfilled\""); + } + + #[test] + fn venue_unscripted_request_uses_the_fallback_then_defaults() { + let venue = MockVenue::default(); + assert!(matches!( + venue.cow_api_request(1, "GET", "/api/v1/anything", None), + Err(CowApiError::Fault(Fault::Unsupported(_))), + )); + venue.set_response_fallback(Ok("catch-all".into())); + assert_eq!( + venue + .cow_api_request(1, "GET", "/api/v1/anything", None) + .unwrap(), + "catch-all", + ); + } + + #[test] + fn mock_host_with_venue_dispatches_through_cow_host_bound() { + let host = MockHost::with_venue(); + host.cow_api.enqueue_submit(Ok("0xuid".into())); + + let _: &dyn shepherd_sdk::cow::CowHost = &host; + assert_eq!(host.submit_order(1, b"{}").unwrap(), "0xuid"); + assert_eq!(host.cow_api.call_count(), 1); + } + #[test] fn mock_host_dispatches_through_cow_host_bound() { let host = MockHost::new(); diff --git a/crates/shepherd-sdk-test/tests/mock_venue.rs b/crates/shepherd-sdk-test/tests/mock_venue.rs new file mode 100644 index 00000000..411f012e --- /dev/null +++ b/crates/shepherd-sdk-test/tests/mock_venue.rs @@ -0,0 +1,348 @@ +//! MockVenue acceptance tests: the scripted venue driving the keeper +//! run (multi-tick retry, backoff, and outage scenarios the +//! single-replayed-response mock cannot express) and module-shaped +//! strategy code polling the venue directly. + +use alloy_primitives::{Address, B256, U256, address, hex, keccak256}; +use cowprotocol::{BuyTokenDestination, GPv2OrderData, OrderKind, SellTokenSource}; +use nexum_sdk::host::{Fault, LocalStoreHost as _, RateLimit}; +use nexum_sdk::keeper::{ConditionalSource, Journal, Tick, WatchRef, WatchSet}; +use shepherd_sdk::cow::{CowApiError, CowHost, OrderRejection, PollOutcome, order_uid_hex, run}; +use shepherd_sdk_test::{MockHost, MockVenue}; + +const SEPOLIA: u64 = 11_155_111; + +type VenueHost = MockHost; + +/// Closure-backed source so each test scripts its own outcome. +struct FnSource(F); + +impl ConditionalSource for FnSource +where + F: Fn(&H, WatchRef<'_>, &[u8], &Tick) -> PollOutcome, +{ + type Outcome = PollOutcome; + + fn poll(&self, host: &H, watch: WatchRef<'_>, params: &[u8], tick: &Tick) -> PollOutcome { + (self.0)(host, watch, params, tick) + } +} + +/// Pin the closure to the higher-ranked source signature at the +/// construction site so inference never guesses a too-narrow lifetime. +fn src(f: F) -> FnSource +where + F: Fn(&VenueHost, WatchRef<'_>, &[u8], &Tick) -> PollOutcome, +{ + FnSource(f) +} + +fn sample_owner() -> Address { + address!("00112233445566778899aabbccddeeff00112233") +} + +fn sample_tick() -> Tick { + Tick { + chain_id: SEPOLIA, + block: 1_000, + epoch_s: 1_700_000_000, + } +} + +/// `validTo` a given number of seconds from now. The `OrderCreation` +/// constructor's client-side max-horizon policy reads the wall clock +/// (not the block clock), so test orders must expire relative to it. +fn valid_to_in(seconds: u64) -> u32 { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock is after the epoch") + .as_secs(); + u32::try_from(now + seconds).expect("test validTo fits u32") +} + +fn submittable_order() -> GPv2OrderData { + GPv2OrderData { + sellToken: address!("6810e776880C02933D47DB1b9fc05908e5386b96"), + buyToken: address!("DAE5F1590db13E3B40423B5b5c5fbf175515910b"), + receiver: Address::ZERO, + sellAmount: U256::from(1_000_000_u64), + buyAmount: U256::from(999_u64), + validTo: valid_to_in(3_600), + appData: cowprotocol::EMPTY_APP_DATA_HASH, + feeAmount: U256::ZERO, + kind: OrderKind::SELL, + partiallyFillable: false, + sellTokenBalance: SellTokenSource::ERC20, + buyTokenBalance: BuyTokenDestination::ERC20, + } +} + +fn ready_outcome(order: &GPv2OrderData) -> PollOutcome { + PollOutcome::Ready { + order: Box::new(order.clone()), + signature: hex!("c0ffeec0ffeec0ffee").to_vec().into(), + } +} + +fn ready_source( + order: &GPv2OrderData, +) -> FnSource, &[u8], &Tick) -> PollOutcome> { + let order = order.clone(); + src(move |_, _, _, _| ready_outcome(&order)) +} + +fn seed_watch(host: &VenueHost) -> String { + WatchSet::new(host) + .put( + &sample_owner(), + &keccak256(b"conditional order params"), + b"params", + ) + .unwrap() +} + +fn client_uid(order: &GPv2OrderData) -> String { + order_uid_hex(SEPOLIA, order, sample_owner()).expect("supported chain, known markers") +} + +fn rejection(error_type: &str) -> CowApiError { + CowApiError::Rejected(OrderRejection { + status: 400, + error_type: error_type.into(), + description: "test".into(), + data: None, + }) +} + +// ---- keeper use ---- + +/// A transient rejection on the first tick keeps the watch alive; the +/// next tick's scripted success is journalled. Per-call scripting is +/// the point: one venue plays a different outcome on each tick. +#[test] +fn keeper_retries_a_transient_rejection_then_submits() { + let host = MockHost::with_venue(); + let key = seed_watch(&host); + let order = submittable_order(); + host.cow_api + .enqueue_submit(Err(rejection("InsufficientFee"))); + host.cow_api.enqueue_submit(Ok(client_uid(&order))); + + let source = ready_source(&order); + run(&host, &source, &sample_tick()).unwrap(); + assert_eq!(host.cow_api.call_count(), 1); + assert!(host.store.snapshot().contains_key(&key), "watch survives"); + assert!( + !Journal::submitted(&host) + .contains(&client_uid(&order)) + .unwrap() + ); + + run(&host, &source, &sample_tick()).unwrap(); + assert_eq!(host.cow_api.call_count(), 2); + assert!( + Journal::submitted(&host) + .contains(&client_uid(&order)) + .unwrap() + ); + assert_eq!( + host.cow_api.pending_submits(), + 0, + "scenario played out in full" + ); +} + +/// A rate-limit with server guidance gates the watch on the epoch +/// clock; the venue is only reached again once the gate clears, and +/// the queued success then lands. +#[test] +fn keeper_backs_off_on_rate_limit_and_submits_after_the_gate() { + let host = MockHost::with_venue(); + seed_watch(&host); + let order = submittable_order(); + host.cow_api + .enqueue_submit(Err(CowApiError::Fault(Fault::RateLimited(RateLimit { + retry_after_ms: Some(2_500), + })))); + host.cow_api.enqueue_submit(Ok(client_uid(&order))); + + let t0 = sample_tick(); + let source = ready_source(&order); + run(&host, &source, &t0).unwrap(); + assert_eq!(host.cow_api.call_count(), 1); + + // 2500ms rounds up to a 3s epoch gate: a tick inside it never + // reaches the venue. + let gated = Tick { + epoch_s: t0.epoch_s + 2, + ..t0 + }; + run(&host, &source, &gated).unwrap(); + assert_eq!(host.cow_api.call_count(), 1, "gated tick must not submit"); + + let clear = Tick { + epoch_s: t0.epoch_s + 3, + ..t0 + }; + run(&host, &source, &clear).unwrap(); + assert_eq!(host.cow_api.call_count(), 2); + assert!( + Journal::submitted(&host) + .contains(&client_uid(&order)) + .unwrap() + ); +} + +/// A venue outage is transient: the watch stays, nothing is gated, and +/// the first tick after recovery submits the queued outcome. +#[test] +fn keeper_survives_a_venue_outage_and_submits_on_recovery() { + let host = MockHost::with_venue(); + let key = seed_watch(&host); + let watch_key = WatchRef::parse(&key).unwrap(); + let order = submittable_order(); + host.cow_api + .inject_fault(CowApiError::Fault(Fault::Unavailable("venue down".into()))); + + let source = ready_source(&order); + run(&host, &source, &sample_tick()).unwrap(); + let snapshot = host.store.snapshot(); + assert!(snapshot.contains_key(&key)); + assert!(!snapshot.contains_key(&watch_key.next_block_key())); + assert!(!snapshot.contains_key(&watch_key.next_epoch_key())); + assert_eq!(host.cow_api.call_count(), 1); + + host.cow_api.clear_fault(); + host.cow_api.enqueue_submit(Ok(client_uid(&order))); + run(&host, &source, &sample_tick()).unwrap(); + assert_eq!(host.cow_api.call_count(), 2); + assert!( + Journal::submitted(&host) + .contains(&client_uid(&order)) + .unwrap() + ); +} + +/// A scripted permanent rejection drops the watch through the ledger. +#[test] +fn keeper_drops_the_watch_on_a_scripted_permanent_rejection() { + let host = MockHost::with_venue(); + seed_watch(&host); + let order = submittable_order(); + host.cow_api + .enqueue_submit(Err(rejection("InvalidSignature"))); + + run(&host, &ready_source(&order), &sample_tick()).unwrap(); + + assert!(host.store.is_empty(), "watch and gates must go"); + assert_eq!(host.cow_api.call_count(), 1); +} + +/// Keeper rows written through the composed host stay invisible to a +/// sibling store namespace, and a decoy watch planted there never +/// reaches the sweep - the store-fidelity seam under the venue tests. +#[test] +fn keeper_sweep_ignores_sibling_namespace_watches() { + let host = MockHost::with_venue(); + seed_watch(&host); + let sibling = host.store.namespaced("other-module"); + assert!(sibling.is_empty(), "keeper rows must not leak across"); + sibling + .set( + "watch:0x00112233445566778899aabbccddeeff00112233:0xdead", + b"decoy", + ) + .unwrap(); + + let order = submittable_order(); + host.cow_api.enqueue_submit(Ok(client_uid(&order))); + let polls = std::cell::Cell::new(0_u32); + run( + &host, + &src(|_, _, _, _| { + polls.set(polls.get() + 1); + ready_outcome(&order) + }), + &sample_tick(), + ) + .unwrap(); + + assert_eq!(polls.get(), 1, "only this module's watch is swept"); + assert_eq!(host.cow_api.call_count(), 1); +} + +// ---- module use ---- + +/// Module-shaped fill tracker: probe the orderbook status route and +/// journal an `observed:` receipt once the order reports fulfilled. +/// Generic over [`CowHost`] exactly like production strategy code. +fn record_fill(host: &H, chain_id: u64, uid: &str) -> Result { + let path = format!("/api/v1/orders/{uid}"); + let Ok(body) = host.cow_api_request(chain_id, "GET", &path, None) else { + return Ok(false); + }; + let fulfilled = serde_json::from_str::(&body) + .ok() + .and_then(|v| v.get("status").and_then(|s| s.as_str().map(str::to_owned))) + .is_some_and(|status| status == "fulfilled"); + if fulfilled { + Journal::observed(host).record(uid)?; + } + Ok(fulfilled) +} + +/// The status sequence advances one entry per module poll and its +/// terminal entry persists across any number of re-polls. +#[test] +fn module_tracks_a_fill_through_a_status_sequence() { + let host = MockHost::with_venue(); + for body in [ + r#"{"status":"open"}"#, + r#"{"status":"open"}"#, + r#"{"status":"fulfilled"}"#, + ] { + host.cow_api.enqueue_order_status("0xuid", Ok(body.into())); + } + + assert!(!record_fill(&host, SEPOLIA, "0xuid").unwrap()); + assert!(!record_fill(&host, SEPOLIA, "0xuid").unwrap()); + assert!(record_fill(&host, SEPOLIA, "0xuid").unwrap()); + // Terminal status sticks: an over-eager re-poll sees it again. + assert!(record_fill(&host, SEPOLIA, "0xuid").unwrap()); + + assert!(Journal::observed(&host).contains("0xuid").unwrap()); + assert_eq!(host.cow_api.request_calls().len(), 4); + assert_eq!(host.cow_api.request_calls()[0].path, "/api/v1/orders/0xuid"); +} + +/// An outage mid-sequence surfaces to the module as a failed probe and +/// consumes nothing: the sequence resumes where it left off. +#[test] +fn module_probe_rides_out_an_injected_outage() { + let host = MockHost::with_venue(); + host.cow_api + .enqueue_order_status("0xuid", Ok(r#"{"status":"open"}"#.into())); + host.cow_api + .enqueue_order_status("0xuid", Ok(r#"{"status":"fulfilled"}"#.into())); + + assert!(!record_fill(&host, SEPOLIA, "0xuid").unwrap()); + + host.cow_api + .inject_fault(CowApiError::Fault(Fault::Timeout)); + assert!(!record_fill(&host, SEPOLIA, "0xuid").unwrap()); + + host.cow_api.clear_fault(); + assert!(record_fill(&host, SEPOLIA, "0xuid").unwrap()); + assert!(Journal::observed(&host).contains("0xuid").unwrap()); +} + +/// A B256 hash keeps the watch-key helper honest against the venue +/// host's default type parameter. +#[test] +fn watch_key_helper_unifies_with_the_venue_host() { + let hash: B256 = keccak256(b"conditional order params"); + assert_eq!( + WatchSet::::key(&sample_owner(), &hash), + WatchSet::::key(&sample_owner(), &hash), + ); +} diff --git a/crates/shepherd-sdk/Cargo.toml b/crates/shepherd-sdk/Cargo.toml index fe8e1af3..2e4547f2 100644 --- a/crates/shepherd-sdk/Cargo.toml +++ b/crates/shepherd-sdk/Cargo.toml @@ -16,8 +16,18 @@ nexum-sdk = { path = "../nexum-sdk" } cowprotocol = { version = "0.2.0", default-features = false } alloy-primitives.workspace = true alloy-sol-types.workspace = true +serde_json.workspace = true strum.workspace = true thiserror.workspace = true +tracing.workspace = true [dev-dependencies] +# `capture_tracing` observes the run's diagnostics in the +# acceptance tests. +nexum-sdk-test = { path = "../nexum-sdk-test" } proptest.workspace = true +# Dev-only cycle (this crate <- shepherd-sdk-test): cargo permits it, +# and the run acceptance-tests against the composed MockHost +# as an integration test - the mock crate links shepherd-sdk +# externally, so the unit-test copy of the traits would not unify. +shepherd-sdk-test = { path = "../shepherd-sdk-test" } diff --git a/crates/shepherd-sdk/src/cow/composable.rs b/crates/shepherd-sdk/src/cow/composable.rs index 212e1240..47aa22ee 100644 --- a/crates/shepherd-sdk/src/cow/composable.rs +++ b/crates/shepherd-sdk/src/cow/composable.rs @@ -12,6 +12,7 @@ use alloy_primitives::{Bytes, U256}; use alloy_sol_types::{SolError, sol}; use cowprotocol::GPv2OrderData; +use nexum_sdk::host::ChainError; sol! { /// Five custom errors `IConditionalOrder.verify` reverts with. @@ -75,9 +76,9 @@ pub enum PollOutcome { /// /// Returns `None` when the selector is not one of the five /// [`IConditionalOrder`] errors - including a bare `Error(string)` -/// require-revert. Callers should treat that as `TryNextBlock` (the -/// safe default) so a transient RPC blip does not drop a still-valid -/// watch. +/// require-revert. [`classify_poll_error`] is the lifecycle policy on +/// top: it treats any such foreign selector as a permanent +/// contract-level rejection. #[must_use] pub fn decode_revert(data: &[u8]) -> Option { if data.len() < 4 { @@ -105,6 +106,29 @@ pub fn decode_revert(data: &[u8]) -> Option { } } +/// Classify a failed poll `eth_call` into a [`PollOutcome`] - the one +/// policy for what a poll failure means to the watch lifecycle. +/// +/// A revert payload big enough to carry a selector that +/// [`decode_revert`] does not recognise maps to `DontTryAgain`: it is +/// a contract-level rejection outside the `IConditionalOrder` +/// vocabulary (a handler-specific error, typically permanent), and +/// retrying it on every block loops forever. Only payload-free +/// failures - transport faults and reverts whose `data` is absent or +/// shorter than a selector - stay `TryNextBlock`. +#[must_use] +pub fn classify_poll_error(err: &ChainError) -> PollOutcome { + match err { + ChainError::Rpc(rpc) => match rpc.data.as_deref() { + Some(data) if data.len() >= 4 => { + decode_revert(data).unwrap_or(PollOutcome::DontTryAgain) + } + _ => PollOutcome::TryNextBlock, + }, + ChainError::Fault(_) => PollOutcome::TryNextBlock, + } +} + fn u256_to_u64_saturating(v: U256) -> u64 { u64::try_from(v).unwrap_or(u64::MAX) } @@ -187,4 +211,68 @@ mod tests { assert_eq!(u256_to_u64_saturating(U256::MAX), u64::MAX); assert_eq!(u256_to_u64_saturating(U256::from(42_u64)), 42); } + + // ---- classify_poll_error ---- + + use nexum_sdk::host::{Fault, RpcError}; + + fn rpc(data: Option>) -> ChainError { + ChainError::Rpc(RpcError { + code: -32000, + message: "execution reverted".into(), + data: data.map(Into::into), + }) + } + + #[test] + fn classify_dispatches_a_recognised_selector() { + let revert = IConditionalOrder::PollTryAtBlock { + blockNumber: U256::from(777_u64), + reason: "wait".to_string(), + } + .abi_encode(); + assert!(matches!( + classify_poll_error(&rpc(Some(revert))), + PollOutcome::TryOnBlock(777) + )); + } + + /// A handler-specific selector outside the `IConditionalOrder` + /// vocabulary is a permanent contract-level rejection: it must + /// drop, not re-poll every block forever. + #[test] + fn classify_unrecognised_selector_drops() { + let mut data = vec![0x7a, 0x93, 0x32, 0x34]; + data.extend_from_slice(&[0u8; 32]); + assert!(matches!( + classify_poll_error(&rpc(Some(data))), + PollOutcome::DontTryAgain + )); + // A bare 4-byte selector with no body classifies the same way. + assert!(matches!( + classify_poll_error(&rpc(Some(vec![0x2c, 0x7c, 0xa6, 0xd7]))), + PollOutcome::DontTryAgain + )); + } + + #[test] + fn classify_payload_free_failures_stay_try_next_block() { + assert!(matches!( + classify_poll_error(&rpc(None)), + PollOutcome::TryNextBlock + )); + assert!(matches!( + classify_poll_error(&rpc(Some(Vec::new()))), + PollOutcome::TryNextBlock + )); + // Sub-selector payloads cannot name a contract error. + assert!(matches!( + classify_poll_error(&rpc(Some(vec![0x01, 0x02]))), + PollOutcome::TryNextBlock + )); + assert!(matches!( + classify_poll_error(&ChainError::Fault(Fault::Timeout)), + PollOutcome::TryNextBlock + )); + } } diff --git a/crates/shepherd-sdk/src/cow/error.rs b/crates/shepherd-sdk/src/cow/error.rs index 23360273..4129cd71 100644 --- a/crates/shepherd-sdk/src/cow/error.rs +++ b/crates/shepherd-sdk/src/cow/error.rs @@ -7,12 +7,16 @@ //! envelope. The guest dispatches on the variant directly, so no //! second JSON decode of a failure body happens strategy-side. //! -//! [`classify_api_error`] maps a decoded [`OrderRejection`] into a -//! [`RetryAction`] the lifecycle layer dispatches on. +//! [`classify_api_error`] maps a decoded [`OrderRejection`] into the +//! keeper [`RetryAction`] the retry ledger dispatches on; +//! [`classify_submit_error`] widens the table to the whole +//! [`CowApiError`] surface. use nexum_sdk::host::{Fault, HostFault}; use strum::IntoStaticStr; +pub use nexum_sdk::keeper::RetryAction; + /// A non-2xx orderbook reply with no typed rejection envelope. `body` /// is the raw response text, foreign orderbook JSON kept verbatim: a /// caller matches on `status` and reads `body` only for diagnostics. @@ -79,49 +83,19 @@ impl HostFault for CowApiError { } } -/// What the lifecycle layer should do after a failed submission. -/// -/// Mirrors the retry contract: `TryNextBlock` / -/// `BackoffSeconds(s)` / `Drop`. The `Backoff` arm has no producer -/// today because the retry classifier is bool-only; the -/// variant is kept so dispatch can grow into it once a server -/// `Retry-After` hint shows up. -/// -/// `IntoStaticStr` exposes each variant as a snake_case `&'static -/// str` so the dispatch layer can record -/// `shepherd_cow_api_retry_total{action=...}` and surface the action -/// in `tracing::info!(retry_action = ...)` without an ad-hoc match -/// ladder. -#[derive(Debug, Eq, PartialEq, IntoStaticStr)] -#[strum(serialize_all = "snake_case")] -#[non_exhaustive] -pub enum RetryAction { - /// Leave the watch / placement in place; the next event will - /// re-attempt. - TryNextBlock, - /// Persist `next_attempt = now + seconds`. Reserved - no producer - /// today (kept so the dispatch contract is stable). - #[allow(dead_code)] - Backoff { - /// Seconds to wait before retrying. - seconds: u64, - }, - /// Remove the watch / mark as terminally rejected. The orderbook - /// will not accept this body on a retry. - Drop, -} - -/// Classify a decoded orderbook [`OrderRejection`] into a -/// [`RetryAction`]. +/// Classify a decoded orderbook [`OrderRejection`] into the keeper +/// [`RetryAction`] - the CoW `errorType` classification table. /// /// - Retriable `error_type`s (`InsufficientFee`, `TooManyLimitOrders`, /// `PriceExceedsMarketPrice`) -> `TryNextBlock`. +/// - Already-submitted rejections ([`is_already_submitted`]) -> +/// `TryNextBlock`: the order is live, so the watch must survive; the +/// run also records the `submitted:` receipt so the next +/// tick short-circuits instead of re-posting. /// - Every other (including unrecognised) kind -> `Drop`. /// -/// Non-`Rejected` failures (transport faults, raw HTTP errors) carry -/// no `error_type` and are not classified here; the caller treats them -/// as transient (leave the watch in place) so a flaky orderbook does -/// not poison a still-valid order. +/// Non-`Rejected` failures carry no `error_type`; classify those with +/// [`classify_submit_error`]. /// /// # Example /// @@ -147,13 +121,48 @@ pub enum RetryAction { /// assert_eq!(classify_api_error(&permanent), RetryAction::Drop); /// ``` pub fn classify_api_error(rejection: &OrderRejection) -> RetryAction { - if is_retriable(&rejection.error_type) { + if is_already_submitted(rejection) || is_retriable(&rejection.error_type) { RetryAction::TryNextBlock } else { RetryAction::Drop } } +/// Whether the rejection says the orderbook already holds this exact +/// order: `DuplicatedOrder` (the orderbook's spelling) plus the +/// `DuplicateOrder` variant older deployments emit. Already-submitted +/// is success wearing an error status - dropping the watch on it would +/// kill every future tranche of a TWAP - so the caller records the +/// `submitted:` receipt and keeps the watch. +pub fn is_already_submitted(rejection: &OrderRejection) -> bool { + matches!( + rejection.error_type.as_str(), + "DuplicatedOrder" | "DuplicateOrder" + ) +} + +/// Classify a whole [`CowApiError`] from a submission into the keeper +/// [`RetryAction`]. +/// +/// A typed rejection dispatches through [`classify_api_error`]; a +/// rate-limit fault with server guidance becomes `Backoff` (hint +/// rounded up to whole seconds, minimum one). Everything else +/// (transport faults, raw HTTP errors, unguided rate limits) is +/// transient -> `TryNextBlock`, so a flaky orderbook never poisons a +/// still-valid order. +pub fn classify_submit_error(err: &CowApiError) -> RetryAction { + match err { + CowApiError::Rejected(rejection) => classify_api_error(rejection), + CowApiError::Fault(Fault::RateLimited(limit)) => match limit.retry_after_ms { + Some(ms) => RetryAction::Backoff { + seconds: ms.div_ceil(1000).max(1), + }, + None => RetryAction::TryNextBlock, + }, + _ => RetryAction::TryNextBlock, + } +} + /// Orderbook `errorType` values the protocol treats as transient: a /// fresh submission on a later block may succeed. Everything else /// (including unrecognised types) is permanent. Mirrors the upstream @@ -199,7 +208,6 @@ mod tests { for kind in [ "InvalidSignature", "WrongOwner", - "DuplicateOrder", "UnsupportedToken", "InvalidAppData", "InvalidErc1271Signature", @@ -220,6 +228,69 @@ mod tests { ); } + /// Both spellings pin: the orderbook emits `DuplicatedOrder`, the + /// older `DuplicateOrder` form must classify identically. Neither + /// may drop the watch - that would kill every future tranche. + #[test] + fn duplicated_order_is_already_submitted_and_never_drops() { + for kind in ["DuplicatedOrder", "DuplicateOrder"] { + assert!(is_already_submitted(&rejection(kind)), "{kind}"); + assert_eq!( + classify_api_error(&rejection(kind)), + RetryAction::TryNextBlock, + "{kind}", + ); + } + assert!(!is_already_submitted(&rejection("InsufficientFee"))); + assert!(!is_already_submitted(&rejection("InvalidSignature"))); + } + + #[test] + fn submit_error_rejection_routes_through_the_table() { + assert_eq!( + classify_submit_error(&CowApiError::Rejected(rejection("InvalidSignature"))), + RetryAction::Drop, + ); + assert_eq!( + classify_submit_error(&CowApiError::Rejected(rejection("InsufficientFee"))), + RetryAction::TryNextBlock, + ); + } + + #[test] + fn submit_error_rate_limit_hint_becomes_backoff_in_whole_seconds() { + let limited = |ms| CowApiError::Fault(Fault::RateLimited(RateLimit { retry_after_ms: ms })); + assert_eq!( + classify_submit_error(&limited(Some(2_500))), + RetryAction::Backoff { seconds: 3 }, + ); + // Sub-second hints round up to a full second, never to zero. + assert_eq!( + classify_submit_error(&limited(Some(1))), + RetryAction::Backoff { seconds: 1 }, + ); + // No guidance -> plain next-block retry. + assert_eq!( + classify_submit_error(&limited(None)), + RetryAction::TryNextBlock + ); + } + + #[test] + fn submit_error_transient_shapes_stay_try_next_block() { + assert_eq!( + classify_submit_error(&CowApiError::Fault(Fault::Timeout)), + RetryAction::TryNextBlock, + ); + assert_eq!( + classify_submit_error(&CowApiError::Http(HttpFailure { + status: 502, + body: None, + })), + RetryAction::TryNextBlock, + ); + } + #[test] fn fault_case_recovers_embedded_fault_and_label() { let err = CowApiError::Fault(Fault::Timeout); diff --git a/crates/shepherd-sdk/src/cow/mod.rs b/crates/shepherd-sdk/src/cow/mod.rs index 36cbdcca..893a80e6 100644 --- a/crates/shepherd-sdk/src/cow/mod.rs +++ b/crates/shepherd-sdk/src/cow/mod.rs @@ -3,20 +3,27 @@ //! Type conversions and ABI decoding helpers that translate between //! the on-chain shape (`GPv2OrderData`, `IConditionalOrder` reverts, //! orderbook JSON) and the typed Rust surface (`OrderData`, -//! `PollOutcome`, `RetryAction`). +//! `PollOutcome`, `RetryAction`), plus [`run()`] - the +//! poll/submit composition over the keeper stores. //! -//! Each submodule stays purely host-neutral: helpers take primitive -//! arguments (`&[u8]`, `Option<&str>`, slices) so they can be unit- -//! tested without wit-bindgen scaffolding and re-used unchanged by -//! TWAP, EthFlow, and future strategy modules. +//! The codec submodules stay purely host-neutral: helpers take +//! primitive arguments (`&[u8]`, `Option<&str>`, slices) so they can +//! be unit-tested without wit-bindgen scaffolding and re-used +//! unchanged by TWAP, EthFlow, and future strategy modules. The +//! run is generic over the host traits alone. pub mod composable; pub mod error; pub mod order; +pub mod run; -pub use composable::{IConditionalOrder, PollOutcome, decode_revert}; -pub use error::{CowApiError, HttpFailure, OrderRejection, RetryAction, classify_api_error}; -pub use order::gpv2_to_order_data; +pub use composable::{IConditionalOrder, PollOutcome, classify_poll_error, decode_revert}; +pub use error::{ + CowApiError, HttpFailure, OrderRejection, RetryAction, classify_api_error, + classify_submit_error, is_already_submitted, +}; +pub use order::{gpv2_to_order_data, order_uid_hex}; +pub use run::run; use nexum_sdk::host::Host; diff --git a/crates/shepherd-sdk/src/cow/order.rs b/crates/shepherd-sdk/src/cow/order.rs index d983d649..5bcef8fb 100644 --- a/crates/shepherd-sdk/src/cow/order.rs +++ b/crates/shepherd-sdk/src/cow/order.rs @@ -7,7 +7,9 @@ //! into Rust enums. [`gpv2_to_order_data`] is the bridge. use alloy_primitives::Address; -use cowprotocol::{BuyTokenDestination, GPv2OrderData, OrderData, OrderKind, SellTokenSource}; +use cowprotocol::{ + BuyTokenDestination, Chain, GPv2OrderData, OrderData, OrderKind, SellTokenSource, +}; /// Convert a freshly-polled / freshly-placed [`GPv2OrderData`] into the /// typed [`OrderData`] shape `OrderCreation::from_signed_order_data` @@ -71,6 +73,23 @@ pub fn gpv2_to_order_data(gpv2: &GPv2OrderData) -> Option { }) } +/// Orderbook UID hex (`0x` + 112 hex chars) for the given on-chain +/// (order, owner, chain) tuple - the same value the orderbook derives +/// server-side from the signed payload, so a client can key +/// idempotency state before any network work. +/// +/// `None` when the chain id has no settlement domain or the order +/// carries an unknown enum marker; both also stop the submit path +/// downstream, so callers fall through and let it surface the +/// diagnostic. +#[must_use] +pub fn order_uid_hex(chain_id: u64, order: &GPv2OrderData, owner: Address) -> Option { + let chain = Chain::try_from(chain_id).ok()?; + let domain = chain.settlement_domain(); + let order_data = gpv2_to_order_data(order)?; + Some(format!("{}", order_data.uid(&domain, owner))) +} + #[cfg(test)] mod tests { use super::*; @@ -137,4 +156,34 @@ mod tests { g.buyTokenBalance = B256::repeat_byte(0x55); assert!(gpv2_to_order_data(&g).is_none()); } + + // ---- order_uid_hex ---- + + const SEPOLIA: u64 = 11_155_111; + + #[test] + fn uid_hex_is_deterministic_and_canonical_shape() { + let g = submittable_gpv2(); + let owner = address!("00112233445566778899aabbccddeeff00112233"); + let uid = order_uid_hex(SEPOLIA, &g, owner).expect("supported chain, known markers"); + // 56 bytes: 32 digest + 20 owner + 4 validTo. + assert_eq!(uid.len(), 2 + 112); + assert!(uid.starts_with("0x")); + assert!( + uid.to_lowercase() + .contains("00112233445566778899aabbccddeeff00112233",) + ); + assert_eq!(order_uid_hex(SEPOLIA, &g, owner).unwrap(), uid); + } + + #[test] + fn uid_hex_none_on_unsupported_chain_or_unknown_marker() { + let g = submittable_gpv2(); + let owner = address!("00112233445566778899aabbccddeeff00112233"); + assert!(order_uid_hex(u64::MAX, &g, owner).is_none()); + + let mut bad = submittable_gpv2(); + bad.kind = B256::repeat_byte(0x42); + assert!(order_uid_hex(SEPOLIA, &bad, owner).is_none()); + } } diff --git a/crates/shepherd-sdk/src/cow/run.rs b/crates/shepherd-sdk/src/cow/run.rs new file mode 100644 index 00000000..cd688170 --- /dev/null +++ b/crates/shepherd-sdk/src/cow/run.rs @@ -0,0 +1,176 @@ +//! Watch run: the poll-loop composition conditional- +//! commitment modules share. +//! +//! [`run`] walks the keeper watch set, polls each gate-ready +//! watch through a [`ConditionalSource`], and runs the +//! [`PollOutcome`]'s effect: lifecycle outcomes update the gate and +//! watch stores, `Ready` drives one submission through the +//! [`CowApiHost`](super::CowApiHost) seam with the `submitted:` +//! journal as the idempotency guard and the keeper [`Retrier`] +//! as the failure dispatch. +//! +//! Store faults abort the sweep (the next tick replays it); +//! submission failures never do - they classify into a +//! [`RetryAction`], the ledger applies the effect, and the sweep +//! moves on. Diagnostics go through the guest `tracing` facade - +//! the same channel strategy code logs on - so module tests observe +//! the composed behaviour with one capture. + +use alloy_primitives::{Address, Bytes}; +use cowprotocol::{GPv2OrderData, OrderCreation, Signature}; +use nexum_sdk::host::Fault; +use nexum_sdk::keeper::{ + ConditionalSource, Gates, Journal, Retrier, RetryAction, Tick, WatchRef, WatchSet, +}; + +use super::{ + CowApiError, CowHost, PollOutcome, classify_submit_error, gpv2_to_order_data, + is_already_submitted, order_uid_hex, +}; + +/// Poll every gate-ready watch once at `tick` and run each outcome's +/// effect. One source poll per ready watch; a `Ready` outcome makes at +/// most one `submit_order` call. +pub fn run(host: &H, source: &S, tick: &Tick) -> Result<(), Fault> +where + H: CowHost, + S: ConditionalSource, +{ + let watches = WatchSet::new(host); + let gates = Gates::new(host); + for key in watches.list()? { + let Some(watch) = WatchRef::parse(&key) else { + continue; + }; + if !gates.is_ready(watch, tick.block, tick.epoch_s)? { + continue; + } + let Some(params) = watches.get(watch)? else { + continue; + }; + match source.poll(host, watch, ¶ms, tick) { + PollOutcome::Ready { order, signature } => { + submit_ready(host, watch, &order, signature, tick, source.label())?; + } + PollOutcome::TryNextBlock => {} + PollOutcome::TryOnBlock(block) => gates.set_next_block(watch, block)?, + PollOutcome::TryAtEpoch(epoch_s) => gates.set_next_epoch(watch, epoch_s)?, + PollOutcome::DontTryAgain => watches.remove(watch)?, + } + } + Ok(()) +} + +/// Submit one freshly-polled `Ready` order, guarding on the +/// `submitted:` journal and dispatching any failure through the retry +/// ledger. +/// +/// The UID is deterministic from on-chain inputs, so the idempotency +/// check runs before any network work; the same value keys the journal +/// marker after, so the read and write paths agree. +fn submit_ready( + host: &H, + watch: WatchRef<'_>, + order: &GPv2OrderData, + signature: Bytes, + tick: &Tick, + label: &str, +) -> Result<(), Fault> { + let Ok(owner) = watch.owner_hex().parse::
() else { + tracing::warn!( + "watch {} carries an unparseable owner; skipping submit", + watch.key(), + ); + return Ok(()); + }; + + let journal = Journal::submitted(host); + let client_uid = order_uid_hex(tick.chain_id, order, owner); + if let Some(uid) = client_uid.as_deref() + && journal.contains(uid)? + { + tracing::info!("{label} {uid} already submitted; skipping re-submit"); + return Ok(()); + } + + let creation = match build_order_creation(order, signature, owner) { + Ok(creation) => creation, + Err(message) => { + tracing::warn!("{label} submit skipped for {owner:#x}: {message}"); + return Ok(()); + } + }; + let body = match serde_json::to_vec(&creation) { + Ok(body) => body, + Err(e) => { + tracing::error!("OrderCreation JSON encode failed: {e}"); + return Ok(()); + } + }; + + match host.submit_order(tick.chain_id, &body) { + Ok(server_uid) => { + // Prefer the client-computed UID so the guard above reads + // what this writes; a divergence would be a protocol bug + // worth a warning, never a silently split keyspace. + let marker = client_uid.as_deref().unwrap_or(server_uid.as_str()); + journal.record(marker)?; + if let Some(client) = client_uid.as_deref() + && client != server_uid + { + tracing::warn!( + "{label} UID divergence: client={client} server={server_uid} \ + (marker keyed on the client UID)" + ); + } + tracing::info!("submitted {marker}"); + } + Err(CowApiError::Rejected(rejection)) if is_already_submitted(&rejection) => { + // Success wearing an error status: the orderbook already + // holds this order. Record the receipt and keep the watch + // so the next tick short-circuits instead of re-posting. + if let Some(uid) = client_uid.as_deref() { + journal.record(uid)?; + } + tracing::info!( + "orderbook already holds this order ({}); receipt recorded", + rejection.error_type, + ); + } + Err(err) => { + let action = classify_submit_error(&err); + Retrier::new(host).apply(watch, action, tick.epoch_s)?; + match action { + RetryAction::TryNextBlock => tracing::warn!("submit retry-next-block: {err}"), + RetryAction::Backoff { seconds } => { + tracing::warn!("submit backoff {seconds}s: {err}"); + } + RetryAction::Drop => tracing::warn!("submit dropped watch: {err}"), + // `RetryAction` is non-exhaustive; the ledger already + // ran the effect, so the log needs only the name. + other => { + let action_label: &'static str = other.into(); + tracing::warn!("submit retry action {action_label}: {err}"); + } + } + } + } + Ok(()) +} + +/// Assemble the `OrderCreation` body the orderbook expects from a +/// polled conditional order. The signed `appData` digest goes out +/// verbatim in the hash-only wire shape (watch-tower parity), and the +/// signature is EIP-1271 - the conditional-order contract is the +/// verifier. +fn build_order_creation( + order: &GPv2OrderData, + signature: Bytes, + from: Address, +) -> Result { + let order_data = gpv2_to_order_data(order) + .ok_or_else(|| "GPv2OrderData carried an unknown enum marker".to_string())?; + let signature = Signature::Eip1271(signature.to_vec()); + OrderCreation::new_app_data_hash_only(&order_data, signature, from, None) + .map_err(|e| e.to_string()) +} diff --git a/crates/shepherd-sdk/src/lib.rs b/crates/shepherd-sdk/src/lib.rs index 86e00dea..70dacc1c 100644 --- a/crates/shepherd-sdk/src/lib.rs +++ b/crates/shepherd-sdk/src/lib.rs @@ -17,8 +17,10 @@ //! (and the [`CowHost`] bound over the core [`Host`]), //! `GPv2OrderData` -> `OrderData` bridging ([`gpv2_to_order_data`]), //! `IConditionalOrder` revert decoding ([`PollOutcome`] + -//! [`decode_revert`]), and the [`RetryAction`] classifier driving -//! submit-failure dispatch. +//! [`decode_revert`]), the classifiers mapping submit failures into +//! the keeper [`RetryAction`], and [`run`] - the poll -> +//! outcome -> gate/journal/submit composition over the keeper +//! stores. //! //! - [`bind_cow_host_via_wit_bindgen!`](bind_cow_host_via_wit_bindgen) - //! the CoW layering of `nexum_sdk::bind_host_via_wit_bindgen!`: @@ -50,6 +52,7 @@ //! [`PollOutcome`]: cow::PollOutcome //! [`decode_revert`]: cow::decode_revert //! [`RetryAction`]: cow::RetryAction +//! [`run`]: cow::run() #![cfg_attr(not(test), warn(unused_crate_dependencies))] #![warn(missing_docs)] diff --git a/crates/shepherd-sdk/tests/run.rs b/crates/shepherd-sdk/tests/run.rs new file mode 100644 index 00000000..d5088626 --- /dev/null +++ b/crates/shepherd-sdk/tests/run.rs @@ -0,0 +1,424 @@ +//! Run acceptance tests against the composed +//! `shepherd_sdk_test::MockHost`. These live as an integration test +//! (not `#[cfg(test)]`) because the mock crate links `shepherd-sdk` +//! externally, and the external and unit-test copies of the traits +//! are distinct types. + +use std::cell::Cell; + +use alloy_primitives::{Address, B256, U256, address, hex, keccak256}; +use cowprotocol::{BuyTokenDestination, GPv2OrderData, OrderKind, SellTokenSource}; +use nexum_sdk::host::{Fault, LocalStoreHost as _, RateLimit}; +use nexum_sdk::keeper::{ConditionalSource, Gates, Journal, Tick, WatchRef, WatchSet}; +use nexum_sdk_test::capture_tracing; +use shepherd_sdk::cow::{CowApiError, OrderRejection, PollOutcome, order_uid_hex, run}; +use shepherd_sdk_test::MockHost; + +const SEPOLIA: u64 = 11_155_111; + +/// Closure-backed source so each test scripts its own outcome and +/// observes its own poll calls. +struct FnSource(F); + +impl ConditionalSource for FnSource +where + F: Fn(&H, WatchRef<'_>, &[u8], &Tick) -> PollOutcome, +{ + type Outcome = PollOutcome; + + fn poll(&self, host: &H, watch: WatchRef<'_>, params: &[u8], tick: &Tick) -> PollOutcome { + (self.0)(host, watch, params, tick) + } +} + +/// Pin the closure to the higher-ranked source signature at the +/// construction site so inference never guesses a too-narrow lifetime. +fn src(f: F) -> FnSource +where + F: Fn(&MockHost, WatchRef<'_>, &[u8], &Tick) -> PollOutcome, +{ + FnSource(f) +} + +fn sample_owner() -> Address { + address!("00112233445566778899aabbccddeeff00112233") +} + +fn sample_hash() -> B256 { + keccak256(b"conditional order params") +} + +fn sample_tick() -> Tick { + Tick { + chain_id: SEPOLIA, + block: 1_000, + epoch_s: 1_700_000_000, + } +} + +/// `validTo` a given number of seconds from now. The `OrderCreation` +/// constructor's client-side max-horizon policy reads the wall clock +/// (not the block clock), so test orders must expire relative to it. +fn valid_to_in(seconds: u64) -> u32 { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .expect("system clock is after the epoch") + .as_secs(); + u32::try_from(now + seconds).expect("test validTo fits u32") +} + +fn submittable_order() -> GPv2OrderData { + GPv2OrderData { + sellToken: address!("6810e776880C02933D47DB1b9fc05908e5386b96"), + buyToken: address!("DAE5F1590db13E3B40423B5b5c5fbf175515910b"), + receiver: Address::ZERO, + sellAmount: U256::from(1_000_000_u64), + buyAmount: U256::from(999_u64), + validTo: valid_to_in(3_600), + appData: cowprotocol::EMPTY_APP_DATA_HASH, + feeAmount: U256::ZERO, + kind: OrderKind::SELL, + partiallyFillable: false, + sellTokenBalance: SellTokenSource::ERC20, + buyTokenBalance: BuyTokenDestination::ERC20, + } +} + +fn ready_outcome(order: &GPv2OrderData) -> PollOutcome { + PollOutcome::Ready { + order: Box::new(order.clone()), + signature: hex!("c0ffeec0ffeec0ffee").to_vec().into(), + } +} + +fn seed_watch(host: &MockHost) -> String { + WatchSet::new(host) + .put(&sample_owner(), &sample_hash(), b"params") + .unwrap() +} + +fn client_uid(order: &GPv2OrderData) -> String { + order_uid_hex(SEPOLIA, order, sample_owner()).expect("supported chain, known markers") +} + +// ---- lifecycle outcomes ---- + +#[test] +fn try_next_block_leaves_the_store_untouched() { + let host = MockHost::new(); + seed_watch(&host); + let before = host.store.snapshot(); + + run( + &host, + &src(|_, _, _, _| PollOutcome::TryNextBlock), + &sample_tick(), + ) + .unwrap(); + + assert_eq!(host.store.snapshot(), before); + assert_eq!(host.cow_api.call_count(), 0); +} + +#[test] +fn try_on_block_sets_the_block_gate() { + let host = MockHost::new(); + let key = seed_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + + run( + &host, + &src(|_, _, _, _| PollOutcome::TryOnBlock(2_000)), + &sample_tick(), + ) + .unwrap(); + + assert_eq!( + host.store.snapshot().get(&watch.next_block_key()).unwrap(), + &2_000_u64.to_le_bytes().to_vec(), + ); +} + +#[test] +fn try_at_epoch_sets_the_epoch_gate() { + let host = MockHost::new(); + let key = seed_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + + run( + &host, + &src(|_, _, _, _| PollOutcome::TryAtEpoch(1_800_000_000)), + &sample_tick(), + ) + .unwrap(); + + assert_eq!( + host.store.snapshot().get(&watch.next_epoch_key()).unwrap(), + &1_800_000_000_u64.to_le_bytes().to_vec(), + ); +} + +#[test] +fn dont_try_again_removes_the_watch_and_its_gates() { + let host = MockHost::new(); + let key = seed_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + Gates::new(&host).set_next_block(watch, 1).unwrap(); + + run( + &host, + &src(|_, _, _, _| PollOutcome::DontTryAgain), + &sample_tick(), + ) + .unwrap(); + + assert!(host.store.is_empty(), "watch and gates must go"); +} + +// ---- gating and skipping ---- + +#[test] +fn gated_watch_is_not_polled() { + let host = MockHost::new(); + let key = seed_watch(&host); + Gates::new(&host) + .set_next_block(WatchRef::parse(&key).unwrap(), 5_000) + .unwrap(); + let polls = Cell::new(0_u32); + + run( + &host, + &src(|_, _, _, _| { + polls.set(polls.get() + 1); + PollOutcome::TryNextBlock + }), + &sample_tick(), + ) + .unwrap(); + + assert_eq!(polls.get(), 0, "a gated watch must not reach the source"); +} + +#[test] +fn malformed_watch_rows_are_skipped() { + let host = MockHost::new(); + host.store.set("watch:no-separator", b"junk").unwrap(); + let polls = Cell::new(0_u32); + + run( + &host, + &src(|_, _, _, _| { + polls.set(polls.get() + 1); + PollOutcome::TryNextBlock + }), + &sample_tick(), + ) + .unwrap(); + + assert_eq!(polls.get(), 0); +} + +// ---- ready -> submission ---- + +#[test] +fn ready_submits_once_and_journals_the_client_uid() { + let host = MockHost::new(); + seed_watch(&host); + let order = submittable_order(); + host.cow_api.respond(Ok(client_uid(&order))); + + let source = { + let order = order.clone(); + src(move |_, _, _, _| ready_outcome(&order)) + }; + run(&host, &source, &sample_tick()).unwrap(); + + assert_eq!(host.cow_api.call_count(), 1); + assert!( + Journal::submitted(&host) + .contains(&client_uid(&order)) + .unwrap(), + "submitted:{{client_uid}} receipt must be recorded", + ); + assert_eq!(host.cow_api.last_call().unwrap().chain_id, SEPOLIA); +} + +#[test] +fn ready_marker_keys_on_the_client_uid_when_the_server_diverges() { + let host = MockHost::new(); + seed_watch(&host); + let order = submittable_order(); + host.cow_api.respond(Ok("0xfeedface".to_string())); + + let source = { + let order = order.clone(); + src(move |_, _, _, _| ready_outcome(&order)) + }; + let (result, logs) = capture_tracing(|| run(&host, &source, &sample_tick())); + result.unwrap(); + + let snapshot = host.store.snapshot(); + assert!(snapshot.contains_key(&format!("submitted:{}", client_uid(&order)))); + assert!( + !snapshot.contains_key("submitted:0xfeedface"), + "marker must key on the client UID, not the divergent server UID", + ); + assert!(logs.any(|e| e.message.contains("UID divergence"))); +} + +#[test] +fn ready_skips_the_orderbook_when_the_receipt_is_journalled() { + let host = MockHost::new(); + seed_watch(&host); + let order = submittable_order(); + Journal::submitted(&host) + .record(&client_uid(&order)) + .unwrap(); + let polls = Cell::new(0_u32); + + run( + &host, + &src(|_, _, _, _| { + polls.set(polls.get() + 1); + ready_outcome(&order) + }), + &sample_tick(), + ) + .unwrap(); + + assert_eq!(polls.get(), 1, "the source is still consulted"); + assert_eq!( + host.cow_api.call_count(), + 0, + "the journal guard must short-circuit before any network work", + ); +} + +#[test] +fn ready_with_unknown_marker_skips_submit_and_keeps_the_watch() { + let host = MockHost::new(); + let key = seed_watch(&host); + let mut order = submittable_order(); + order.kind = B256::repeat_byte(0x42); + + run( + &host, + &src(move |_, _, _, _| ready_outcome(&order)), + &sample_tick(), + ) + .unwrap(); + + assert_eq!(host.cow_api.call_count(), 0); + assert!(host.store.snapshot().contains_key(&key)); +} + +// ---- submission failure dispatch ---- + +fn rejection(error_type: &str) -> CowApiError { + CowApiError::Rejected(OrderRejection { + status: 400, + error_type: error_type.into(), + description: "test".into(), + data: None, + }) +} + +#[test] +fn transient_rejection_keeps_the_watch_ungated() { + let host = MockHost::new(); + let key = seed_watch(&host); + let watch_key = WatchRef::parse(&key).unwrap(); + let order = submittable_order(); + host.cow_api.respond(Err(rejection("InsufficientFee"))); + + run( + &host, + &src(move |_, _, _, _| ready_outcome(&order)), + &sample_tick(), + ) + .unwrap(); + + let snapshot = host.store.snapshot(); + assert!(snapshot.contains_key(&key)); + assert!(!snapshot.contains_key(&watch_key.next_block_key())); + assert!(!snapshot.contains_key(&watch_key.next_epoch_key())); + assert!(!snapshot.keys().any(|k| k.starts_with("submitted:"))); +} + +#[test] +fn permanent_rejection_drops_the_watch_through_the_ledger() { + let host = MockHost::new(); + let key = seed_watch(&host); + Gates::new(&host) + .set_next_block(WatchRef::parse(&key).unwrap(), 1) + .unwrap(); + let order = submittable_order(); + host.cow_api.respond(Err(rejection("InvalidSignature"))); + + run( + &host, + &src(move |_, _, _, _| ready_outcome(&order)), + &sample_tick(), + ) + .unwrap(); + + assert!( + host.store.is_empty(), + "a permanent rejection must drop the watch and its gates", + ); +} + +/// The orderbook already holds the order: the receipt is recorded, the +/// watch survives, and the next tick short-circuits on the journal +/// instead of re-posting. +#[test] +fn duplicated_order_records_the_receipt_and_keeps_the_watch() { + let host = MockHost::new(); + let key = seed_watch(&host); + let order = submittable_order(); + host.cow_api.respond(Err(rejection("DuplicatedOrder"))); + + let source = { + let order = order.clone(); + src(move |_, _, _, _| ready_outcome(&order)) + }; + run(&host, &source, &sample_tick()).unwrap(); + + assert!(host.store.snapshot().contains_key(&key)); + assert!( + Journal::submitted(&host) + .contains(&client_uid(&order)) + .unwrap(), + "already-submitted must record the receipt", + ); + + // The next tick must not touch the orderbook again. + run(&host, &source, &sample_tick()).unwrap(); + assert_eq!(host.cow_api.call_count(), 1); +} + +/// A rate-limit fault with server guidance backs the watch off on the +/// epoch clock - `RetryAction::Backoff` reached through the ledger. +#[test] +fn rate_limited_submit_backs_off_through_the_epoch_gate() { + let host = MockHost::new(); + let key = seed_watch(&host); + let watch = WatchRef::parse(&key).unwrap(); + let order = submittable_order(); + host.cow_api + .respond(Err(CowApiError::Fault(Fault::RateLimited(RateLimit { + retry_after_ms: Some(2_500), + })))); + + let tick = sample_tick(); + run(&host, &src(move |_, _, _, _| ready_outcome(&order)), &tick).unwrap(); + + let snapshot = host.store.snapshot(); + assert!(snapshot.contains_key(&key), "backoff must keep the watch"); + assert_eq!( + snapshot.get(&watch.next_epoch_key()).unwrap(), + &(tick.epoch_s + 3).to_le_bytes().to_vec(), + "2500ms rounds up to a 3s backoff from the tick clock", + ); + assert!(!snapshot.keys().any(|k| k.starts_with("submitted:"))); +} diff --git a/engine.example.toml b/engine.example.toml index 39624528..a3f883bd 100644 --- a/engine.example.toml +++ b/engine.example.toml @@ -52,6 +52,15 @@ log_level = "info" # total_deadline_ms = 60_000 # response_body_max_bytes = 16_777_216 +# Chain JSON-RPC response size cap, applied host-side before a response +# is copied into guest memory. A single response - and the aggregate of +# one request-batch - beyond the cap fails with a typed invalid-input +# fault instead of inflating the module toward its memory_bytes limit. +# Modules that need more data per call should paginate (block-range +# chunking, request-batch). 0 saturates to 1. Default 1 MiB. +[limits.chain] +# response_body_max_bytes = 1_048_576 + # Per-run log retention. bytes_per_run is the byte budget for one run's # in-memory ring (oldest records evict first, the newest is never evicted # to nothing); runs_retained is how many past runs are kept per module. diff --git a/modules/twap-monitor/Cargo.toml b/modules/twap-monitor/Cargo.toml index 1910ae93..91a37a90 100644 --- a/modules/twap-monitor/Cargo.toml +++ b/modules/twap-monitor/Cargo.toml @@ -14,12 +14,10 @@ shepherd-sdk = { path = "../../crates/shepherd-sdk" } cowprotocol = { version = "0.2.0", default-features = false } alloy-primitives = { version = "1.6", default-features = false, features = ["std"] } alloy-sol-types = { version = "1.6", default-features = false, features = ["std"] } -serde_json = { version = "1", default-features = false, features = ["alloc"] } -strum = { version = "0.28", default-features = false, features = ["derive"] } -thiserror = "2" tracing = { version = "0.1", default-features = false } wit-bindgen = { version = "0.59", default-features = false, features = ["macros", "realloc"] } [dev-dependencies] +serde_json = { version = "1", default-features = false, features = ["alloc"] } shepherd-sdk-test = { path = "../../crates/shepherd-sdk-test" } nexum-sdk-test = { path = "../../crates/nexum-sdk-test" } diff --git a/modules/twap-monitor/src/strategy.rs b/modules/twap-monitor/src/strategy.rs index a58a58bf..fe0c72c1 100644 --- a/modules/twap-monitor/src/strategy.rs +++ b/modules/twap-monitor/src/strategy.rs @@ -7,20 +7,23 @@ //! imports and hands it to [`on_chain_logs`] / [`on_block`]; tests under //! `#[cfg(test)]` hand the same functions a //! `shepherd_sdk_test::MockHost`. +//! +//! The module owns decode and evaluate only: log decoding into the +//! keeper watch set, and the `getTradeableOrderWithSignature` poll +//! behind [`ConditionalSource`]. Gate discipline, the `submitted:` +//! journal, submission, and retry dispatch live in the shared +//! composition (`shepherd_sdk::cow::run`). -use alloy_primitives::{Address, B256, Bytes, keccak256}; +use alloy_primitives::{Address, Bytes, keccak256}; use alloy_sol_types::{SolCall, SolEvent, SolValue}; use cowprotocol::{ - COMPOSABLE_COW, Chain, ComposableCoW::ConditionalOrderCreated, ConditionalOrderParams, - GPv2OrderData, OrderCreation, Signature, + COMPOSABLE_COW, ComposableCoW::ConditionalOrderCreated, ConditionalOrderParams, GPv2OrderData, }; use nexum_sdk::chain::{eth_call_params, parse_eth_call_result}; use nexum_sdk::events::Log; use nexum_sdk::host::{ChainError, Fault}; -use shepherd_sdk::cow::{ - CowApiError, CowHost, PollOutcome, RetryAction, classify_api_error, decode_revert, - gpv2_to_order_data, -}; +use nexum_sdk::keeper::{ConditionalSource, Tick, WatchRef, WatchSet}; +use shepherd_sdk::cow::{CowHost, PollOutcome, classify_poll_error, run}; /// Block fields the poll path reads on every dispatch. pub struct BlockInfo { @@ -66,9 +69,16 @@ pub fn on_chain_logs(host: &H, logs: &[Log]) -> Result<(), Fault> { Ok(()) } -/// Poll entry: scan every persisted watch and dispatch ready tranches. +/// Poll entry: run every gate-ready watch through the keeper +/// composition. The block timestamp arrives in milliseconds; the tick +/// carries Unix seconds. pub fn on_block(host: &H, block: BlockInfo) -> Result<(), Fault> { - poll_all_watches(host, &block) + let tick = Tick { + chain_id: block.chain_id, + block: block.number, + epoch_s: block.timestamp / 1000, + }; + run(host, &TwapSource, &tick) } // ---- indexing path ---- @@ -78,64 +88,51 @@ fn decode_conditional_order_created(log: &Log) -> Option<(Address, ConditionalOr Some((decoded.data.owner, decoded.data.params)) } -/// `set` overwrites in place, so re-indexing the same log (re-org -/// replay, overlapping subscription windows) produces no observable -/// side effect. +/// The watch set overwrites in place, so re-indexing the same log +/// (re-org replay, overlapping subscription windows) produces no +/// observable side effect. fn persist_watch( host: &H, owner: Address, params: &ConditionalOrderParams, ) -> Result<(), Fault> { let encoded = params.abi_encode(); - let params_hash = keccak256(&encoded); - let key = watch_key(&owner, ¶ms_hash); - host.set(&key, &encoded)?; + let key = WatchSet::new(host).put(&owner, &keccak256(&encoded), &encoded)?; tracing::info!("indexed {key}"); Ok(()) } // ---- poll path ---- -fn poll_all_watches(host: &H, block: &BlockInfo) -> Result<(), Fault> { - let now_epoch_s = block.timestamp / 1000; - let keys = host.list_keys("watch:")?; - for key in keys { - let Some((owner_hex, hash_hex)) = parse_watch_key(&key) else { - continue; - }; - if !is_ready(host, owner_hex, hash_hex, block.number, now_epoch_s)? { - continue; - } - let Some(value) = host.get(&key)? else { - continue; - }; - let Ok(params) = ConditionalOrderParams::abi_decode(&value) else { - tracing::warn!("watch {key} carried unparseable params; skipping"); - continue; +/// TWAP conditional source: decode the stored `ConditionalOrderParams` +/// and evaluate `getTradeableOrderWithSignature` on chain. A row this +/// source cannot decode polls again next block rather than tearing +/// down the sweep. +struct TwapSource; + +impl ConditionalSource for TwapSource { + type Outcome = PollOutcome; + + fn poll(&self, host: &H, watch: WatchRef<'_>, params: &[u8], tick: &Tick) -> PollOutcome { + let Ok(params) = ConditionalOrderParams::abi_decode(params) else { + tracing::warn!("watch {} carried unparseable params; skipping", watch.key()); + return PollOutcome::TryNextBlock; }; - let Ok(owner) = owner_hex.parse::
() else { - continue; + let Ok(owner) = watch.owner_hex().parse::
() else { + tracing::warn!( + "watch {} carried an unparseable owner; skipping", + watch.key() + ); + return PollOutcome::TryNextBlock; }; - let outcome = poll_one(host, block.chain_id, &owner, ¶ms); - tracing::info!("poll {key} -> {}", outcome_label(&outcome)); - match outcome { - PollOutcome::Ready { order, signature } => { - submit_ready( - host, - block.chain_id, - owner, - &order, - signature, - &key, - now_epoch_s, - )?; - } - non_ready => { - apply_watch_update(host, outcome_to_update(&non_ready), &key)?; - } - } + let outcome = poll_one(host, tick.chain_id, &owner, ¶ms); + tracing::info!("poll {} -> {}", watch.key(), outcome_label(&outcome)); + outcome + } + + fn label(&self) -> &'static str { + "twap" } - Ok(()) } fn poll_one( @@ -159,28 +156,14 @@ fn poll_one( Ok(result_json) => parse_eth_call_result(&result_json) .and_then(|bytes| decode_return(&bytes)) .unwrap_or(PollOutcome::TryNextBlock), - // A structured JSON-RPC error (the normal shape for an - // `eth_call` revert): the chain backend has already hex-decoded - // the `error.data` payload, so `decode_revert` dispatches - // `PollTryAtBlock` / `PollTryAtEpoch` / `OrderNotValid` / - // `PollNever` straight off the bytes. A revert the decoder does - // not recognise falls through to the safe `TryNextBlock`. - Err(ChainError::Rpc(rpc)) => rpc - .data - .as_deref() - .and_then(|bytes| decode_revert(bytes)) - .unwrap_or_else(|| { - tracing::warn!( - "eth_call reverted ({}); defaulting to TryNextBlock", - rpc.message - ); - PollOutcome::TryNextBlock - }), - // A transport-level fault (timeout, RPC down, ...): retry on the - // next block. - Err(ChainError::Fault(fault)) => { - tracing::warn!("eth_call failed ({fault}); defaulting to TryNextBlock"); - PollOutcome::TryNextBlock + // `classify_poll_error` is the one policy for what a failed + // poll call means to the watch lifecycle; only a transport + // fault warrants its own diagnostic here. + Err(err) => { + if let ChainError::Fault(fault) = &err { + tracing::warn!("eth_call failed ({fault}); retrying next block"); + } + classify_poll_error(&err) } } } @@ -207,312 +190,36 @@ fn outcome_label(o: &PollOutcome) -> &'static str { } } -// ---- key conventions ---- +// ---- test-only seam mirrors ---- +// +// Thin views over the keeper / SDK canon so the dispatch tests can +// seed and inspect the store in the exact shapes production writes. -fn watch_key(owner: &Address, params_hash: &B256) -> String { - format!("watch:{owner:#x}:{params_hash:#x}") +#[cfg(test)] +fn watch_key(owner: &Address, params_hash: &alloy_primitives::B256) -> String { + WatchSet::::key(owner, params_hash) } +#[cfg(test)] fn parse_watch_key(key: &str) -> Option<(&str, &str)> { - let rest = key.strip_prefix("watch:")?; - let (owner, hash) = rest.split_once(':')?; - Some((owner, hash)) -} - -fn is_ready( - host: &H, - owner_hex: &str, - hash_hex: &str, - block_number: u64, - epoch_s: u64, -) -> Result { - if let Some(next) = read_u64(host, &format!("next_block:{owner_hex}:{hash_hex}"))? - && block_number < next - { - return Ok(false); - } - if let Some(next) = read_u64(host, &format!("next_epoch:{owner_hex}:{hash_hex}"))? - && epoch_s < next - { - return Ok(false); - } - Ok(true) -} - -fn read_u64(host: &H, key: &str) -> Result, Fault> { - let bytes = host.get(key)?; - Ok(bytes - .and_then(|b| <[u8; 8]>::try_from(b.as_slice()).ok()) - .map(u64::from_le_bytes)) -} - -// ---- submission path ---- - -/// `cowprotocol`-side rejection envelope for an `OrderCreation` we -/// failed to assemble. Surfaces in a Warn log; the watch is left in -/// place so the next poll can either re-construct or transition on -/// its own. -/// -/// `IntoStaticStr` exposes each variant as a snake_case `&'static -/// str` so the submission warning log can carry `error_kind = -/// unknown_marker` without a match-ladder in the call site. -#[derive(Debug, thiserror::Error, strum::IntoStaticStr)] -#[strum(serialize_all = "snake_case")] -#[non_exhaustive] -enum BuildError { - /// `GPv2OrderData` carried a marker (`kind`, balance enum) we don't - /// know how to map. - #[error("GPv2OrderData carried an unknown enum marker")] - UnknownMarker, - /// `cowprotocol` rejected the body - typically `from == - /// Address::ZERO` or a `validTo` beyond the client-side horizon. - #[error(transparent)] - Cowprotocol(#[from] cowprotocol::Error), -} - -/// Assemble the `OrderCreation` body the orderbook expects from a -/// freshly-polled TWAP tranche. -/// -/// The signed `order.appData` digest is submitted verbatim (the -/// hash-only `OrderCreationAppData::Hash` wire shape) - watch-tower -/// parity. The orderbook joins the document it already has registered -/// for that digest; when it has none, the submit rejects with -/// `INVALID_APP_DATA` and [`classify_api_error`] dispatches the retry. -fn build_order_creation( - order: &GPv2OrderData, - signature: Bytes, - from: Address, -) -> Result { - let order_data = gpv2_to_order_data(order).ok_or(BuildError::UnknownMarker)?; - let signature = Signature::Eip1271(signature.to_vec()); - let creation = OrderCreation::new_app_data_hash_only(&order_data, signature, from, None)?; - Ok(creation) + let watch = WatchRef::parse(key)?; + Some((watch.owner_hex(), watch.hash_hex())) } -fn submit_ready( - host: &H, - chain_id: u64, - owner: Address, - order: &GPv2OrderData, - signature: Bytes, - watch_key: &str, - now_epoch_s: u64, -) -> Result<(), Fault> { - // Short-circuit if the orderbook UID for this exact - // (order, owner, chain) tuple is already in our local-store as - // `submitted:`. The poll-tick can re-fire `Ready` for the same - // TWAP child in successive blocks - `getTradeableOrderWithSignature` - // does not know shepherd already POSTed it - and re-submitting - // wastes a submit_order call and emits a misleading - // `DuplicatedOrder` Warn. The UID computation is deterministic - // from on-chain inputs (and matches what the orderbook derives - // server-side from the signed payload), so we can check before - // doing any network work. We also reuse the computed value below - // as the `submitted:{uid}` marker key, so the read and write - // paths agree. - let client_uid_hex = compute_uid_hex(chain_id, order, owner); - if let Some(uid_hex) = client_uid_hex.as_deref() - && host.get(&format!("submitted:{uid_hex}"))?.is_some() - { - tracing::info!("twap {uid_hex} already submitted; skipping poll re-submit"); - return Ok(()); - } - - // CoW Swap UI (and other clients) sign TWAPs with a non-empty - // `appData` hash that points at a JSON document already registered - // with the orderbook. Submit the signed digest verbatim (hash-only - // shape) and let the orderbook join its own registry - watch-tower - // parity. An unregistered digest rejects as `INVALID_APP_DATA` and - // `classify_api_error` dispatches the backoff. - let creation = match build_order_creation(order, signature, owner) { - Ok(c) => c, - Err(e) => { - tracing::warn!("twap submit skipped for {owner:#x}: {e}"); - return Ok(()); - } - }; - let body = match serde_json::to_vec(&creation) { - Ok(b) => b, - Err(e) => { - tracing::error!("OrderCreation JSON encode failed: {e}"); - return Ok(()); - } - }; - match host.submit_order(chain_id, &body) { - Ok(server_uid) => { - // Prefer the client-computed UID for the marker key so the - // idempotency check at the top of `submit_ready` reads what - // we wrote. In production the server-returned - // UID is the same value (both sides derive it from the - // signed `OrderData` via the canonical - // `digest || owner || valid_to` layout); a divergence - // would be a protocol-level bug worth surfacing rather - // than silently splitting the keyspace. - let marker_uid = client_uid_hex.as_deref().unwrap_or(server_uid.as_str()); - let key = format!("submitted:{marker_uid}"); - // Empty marker - presence of the key is the receipt. - host.set(&key, b"")?; - if let Some(client_uid) = client_uid_hex.as_deref() - && client_uid != server_uid - { - tracing::warn!( - "twap UID divergence: client={client_uid} server={server_uid} \ - (marker stored under client UID for idempotency consistency)" - ); - } - tracing::info!("submitted {key}"); - } - Err(err) => { - apply_submit_retry(host, &err, watch_key, now_epoch_s)?; - } - } - Ok(()) -} - -/// Compute the orderbook UID hex (`0x` + 112 hex chars) for the given -/// on-chain (order, owner, chain) tuple, mirroring what `submit_order` -/// will deduce server-side. Used by [`submit_ready`] to short-circuit -/// poll-tick re-submissions of an already-submitted TWAP child. -/// -/// Returns `None` if the chain id is unsupported by `cowprotocol::Chain` -/// or the order carries an unknown enum marker - both cases also stop -/// the regular submit path downstream, so the caller can fall through -/// to the normal flow and let it surface the appropriate diagnostic. +#[cfg(test)] fn compute_uid_hex(chain_id: u64, order: &GPv2OrderData, owner: Address) -> Option { - let chain = Chain::try_from(chain_id).ok()?; - let domain = chain.settlement_domain(); - let order_data = gpv2_to_order_data(order)?; - Some(format!("{}", order_data.uid(&domain, owner))) -} - -// ---- OrderPostError -> retry action ---- - -fn apply_submit_retry( - host: &H, - err: &CowApiError, - watch_key: &str, - now_epoch_s: u64, -) -> Result<(), Fault> { - // Only a typed orderbook rejection classifies; transport faults and - // raw HTTP errors are transient, so the watch stays in place. - let action = match err { - CowApiError::Rejected(rejection) => classify_api_error(rejection), - _ => RetryAction::TryNextBlock, - }; - match action { - RetryAction::TryNextBlock => { - tracing::warn!("submit retry-next-block: {err}"); - } - RetryAction::Backoff { seconds } => { - let until = now_epoch_s.saturating_add(seconds); - if let Some((owner_hex, hash_hex)) = parse_watch_key(watch_key) { - host.set( - &format!("next_epoch:{owner_hex}:{hash_hex}"), - &until.to_le_bytes(), - )?; - } - tracing::warn!("submit backoff {seconds}s -> next_epoch={until}: {err}"); - } - RetryAction::Drop => { - host.delete(watch_key)?; - if let Some((owner_hex, hash_hex)) = parse_watch_key(watch_key) { - let _ = host.delete(&format!("next_block:{owner_hex}:{hash_hex}")); - let _ = host.delete(&format!("next_epoch:{owner_hex}:{hash_hex}")); - } - tracing::warn!("submit dropped watch: {err}"); - } - // `RetryAction` is `#[non_exhaustive]`; future variants - // default to "leave the watch in place" (the conservative - // dispatch choice). Once a new variant gets a real meaning - // its arm should be added explicitly. - _ => { - tracing::warn!("submit unknown retry-action: {err} - leaving watch in place"); - } - } - Ok(()) -} - -// ---- PollOutcome lifecycle dispatch ---- - -/// What `apply_watch_update` should do for a given outcome. Kept as a -/// data type (rather than running the effects directly) so the -/// decision is host-free testable. -#[derive(Debug, Eq, PartialEq)] -enum WatchUpdate { - /// Leave the store untouched. Next block re-polls the watch. - NoOp, - /// Write `next_block:` so subsequent polls skip until the given - /// block number is reached. - SetNextBlock(u64), - /// Write `next_epoch:` so subsequent polls skip until the given - /// Unix-seconds timestamp is reached. - SetNextEpoch(u64), - /// Delete the watch and any stale gate keys - TWAP completed, - /// cancelled, or otherwise irrecoverable. - DropWatch, -} - -/// Pure mapping from a non-Ready `PollOutcome` to the lifecycle effect -/// the contract specifies. `Ready` is handled by the submit -/// path and is rejected here so a caller cannot -/// accidentally erase the watch when an order was actually produced. -fn outcome_to_update(outcome: &PollOutcome) -> WatchUpdate { - match outcome { - PollOutcome::Ready { .. } => WatchUpdate::NoOp, - PollOutcome::TryNextBlock => WatchUpdate::NoOp, - PollOutcome::TryOnBlock(n) => WatchUpdate::SetNextBlock(*n), - PollOutcome::TryAtEpoch(t) => WatchUpdate::SetNextEpoch(*t), - PollOutcome::DontTryAgain => WatchUpdate::DropWatch, - } -} - -fn apply_watch_update( - host: &H, - update: WatchUpdate, - watch_key: &str, -) -> Result<(), Fault> { - match update { - WatchUpdate::NoOp => Ok(()), - WatchUpdate::SetNextBlock(n) => { - if let Some((owner_hex, hash_hex)) = parse_watch_key(watch_key) { - host.set( - &format!("next_block:{owner_hex}:{hash_hex}"), - &n.to_le_bytes(), - )?; - } - Ok(()) - } - WatchUpdate::SetNextEpoch(t) => { - if let Some((owner_hex, hash_hex)) = parse_watch_key(watch_key) { - host.set( - &format!("next_epoch:{owner_hex}:{hash_hex}"), - &t.to_le_bytes(), - )?; - } - Ok(()) - } - WatchUpdate::DropWatch => { - host.delete(watch_key)?; - if let Some((owner_hex, hash_hex)) = parse_watch_key(watch_key) { - let _ = host.delete(&format!("next_block:{owner_hex}:{hash_hex}")); - let _ = host.delete(&format!("next_epoch:{owner_hex}:{hash_hex}")); - } - tracing::info!("dropped watch {watch_key}"); - Ok(()) - } - } + shepherd_sdk::cow::order_uid_hex(chain_id, order, owner) } #[cfg(test)] mod tests { use super::*; - use alloy_primitives::{U256, address, b256, hex}; - use cowprotocol::OrderCreationAppData; + use alloy_primitives::{B256, U256, address, b256, hex}; use cowprotocol::{BuyTokenDestination, OrderKind, SellTokenSource}; use nexum_sdk::Level; use nexum_sdk::host::LocalStoreHost as _; use nexum_sdk_test::capture_tracing; - use shepherd_sdk::cow::OrderRejection; + use shepherd_sdk::cow::{CowApiError, OrderRejection}; use shepherd_sdk_test::MockHost; const SEPOLIA: u64 = 11_155_111; @@ -627,33 +334,6 @@ mod tests { } } - /// The signed `appData` digest goes into the body verbatim as the - /// hash-only shape - no document lookup, no digest re-derivation. - #[test] - fn build_order_creation_submits_app_data_hash_verbatim() { - let owner = address!("00112233445566778899aabbccddeeff00112233"); - let sig: Bytes = hex!("c0ffeec0ffeec0ffee").to_vec().into(); - let mut order = submittable_order(); - order.appData = B256::repeat_byte(0xee); - let creation = build_order_creation(&order, sig.clone(), owner).expect("build succeeds"); - assert_eq!(creation.from, owner); - assert_eq!(creation.signing_scheme, cowprotocol::SigningScheme::Eip1271); - assert_eq!(creation.signature.to_bytes(), sig.to_vec()); - assert_eq!( - creation.app_data, - OrderCreationAppData::Hash { - hash: order.appData - } - ); - } - - #[test] - fn build_order_creation_rejects_zero_from() { - let err = - build_order_creation(&submittable_order(), Bytes::new(), Address::ZERO).unwrap_err(); - assert!(matches!(err, BuildError::Cowprotocol(_))); - } - #[test] fn watch_key_round_trips_via_parse() { let owner = address!("00112233445566778899aabbccddeeff00112233"); @@ -664,48 +344,6 @@ mod tests { assert_eq!(h.parse::().unwrap(), hash); } - #[test] - fn outcome_try_next_block_is_no_op() { - assert_eq!( - outcome_to_update(&PollOutcome::TryNextBlock), - WatchUpdate::NoOp - ); - } - - #[test] - fn outcome_try_on_block_sets_next_block_gate() { - assert_eq!( - outcome_to_update(&PollOutcome::TryOnBlock(12_345)), - WatchUpdate::SetNextBlock(12_345), - ); - } - - #[test] - fn outcome_try_at_epoch_sets_next_epoch_gate() { - assert_eq!( - outcome_to_update(&PollOutcome::TryAtEpoch(1_700_000_000)), - WatchUpdate::SetNextEpoch(1_700_000_000), - ); - } - - #[test] - fn outcome_dont_try_again_drops_watch() { - assert_eq!( - outcome_to_update(&PollOutcome::DontTryAgain), - WatchUpdate::DropWatch - ); - } - - #[test] - fn outcome_ready_is_handled_by_submit_path_not_lifecycle() { - let order = Box::new(submittable_order()); - let outcome = PollOutcome::Ready { - order, - signature: Bytes::new(), - }; - assert_eq!(outcome_to_update(&outcome), WatchUpdate::NoOp); - } - // ---- MockHost dispatch tests ---- /// Build the alloy log the indexer expects from a well-formed