From 4e1ec36ec873cef6232a814351d106d08e569a0c Mon Sep 17 00:00:00 2001 From: mfw78 Date: Mon, 6 Jul 2026 16:24:58 +0000 Subject: [PATCH 1/9] feat(sdk): add the keeper conditional-source seam, retry ledger, and run Introduce the world-neutral half of the poll loop in nexum_sdk::keeper: a ConditionalSource trait (one watch in, one outcome out, generic over the host, no venue-transport pre-abstraction), a Tick carrying the dispatch instant, and the RetryAction (try-next-block / backoff / drop) whose effects the Retrier runs through the gate and watch-set stores. RetryAction moves out of shepherd-sdk; the CoW crate re-exports it so no module rewires. shepherd-sdk keeps classify_api_error as the CoW errorType table but returns the keeper action, and encodes two fixes in the one classification path: classify_poll_error maps an unrecognised revert selector with a structured payload to DontTryAgain instead of re-polling every block forever, and DuplicatedOrder (both spellings) classifies as already-submitted - the run records the submitted: receipt and keeps the watch rather than dropping every future tranche. classify_submit_error widens the table to the whole CowApiError surface and gives Backoff its producer: a rate-limit fault with server guidance gates the watch on the epoch clock. cow::run composes the loop: watch-set scan -> gate check -> source poll -> PollOutcome dispatch, driving submission through the existing CowApiHost seam with the journal as the idempotency guard and the retry ledger as the failure dispatch. Acceptance tests run against the composed shepherd-sdk-test MockHost as integration tests; no module is rewired yet. --- Cargo.lock | 2 + crates/nexum-sdk/src/keeper.rs | 95 +++++ crates/nexum-sdk/src/lib.rs | 6 +- crates/nexum-sdk/tests/keeper.rs | 138 ++++++- crates/shepherd-sdk/Cargo.toml | 6 + crates/shepherd-sdk/src/cow/composable.rs | 94 ++++- crates/shepherd-sdk/src/cow/error.rs | 155 +++++--- crates/shepherd-sdk/src/cow/mod.rs | 23 +- crates/shepherd-sdk/src/cow/order.rs | 51 ++- crates/shepherd-sdk/src/cow/run.rs | 182 ++++++++++ crates/shepherd-sdk/src/lib.rs | 7 +- crates/shepherd-sdk/tests/run.rs | 422 ++++++++++++++++++++++ 12 files changed, 1123 insertions(+), 58 deletions(-) create mode 100644 crates/shepherd-sdk/src/cow/run.rs create mode 100644 crates/shepherd-sdk/tests/run.rs diff --git a/Cargo.lock b/Cargo.lock index db6e15e7..fa9ae472 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5102,6 +5102,8 @@ dependencies = [ "cowprotocol", "nexum-sdk", "proptest", + "serde_json", + "shepherd-sdk-test", "strum", "thiserror 2.0.18", ] diff --git a/crates/nexum-sdk/src/keeper.rs b/crates/nexum-sdk/src/keeper.rs index 06d1cd9e..88edc6de 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,89 @@ 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; +} + +/// 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/Cargo.toml b/crates/shepherd-sdk/Cargo.toml index fe8e1af3..717d5c5d 100644 --- a/crates/shepherd-sdk/Cargo.toml +++ b/crates/shepherd-sdk/Cargo.toml @@ -16,8 +16,14 @@ 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 [dev-dependencies] 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..44a96464 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, + }) + } + + #[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..9f616d1f 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..a7412ca5 --- /dev/null +++ b/crates/shepherd-sdk/src/cow/run.rs @@ -0,0 +1,182 @@ +//! 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`](super::RetryAction), the ledger applies the +//! effect, and the sweep moves on. + +use alloy_primitives::{Address, Bytes}; +use cowprotocol::{GPv2OrderData, OrderCreation, Signature}; +use nexum_sdk::Level; +use nexum_sdk::host::Fault; +use nexum_sdk::keeper::{ConditionalSource, Gates, Journal, Retrier, 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)?; + } + 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, +) -> Result<(), Fault> { + let Ok(owner) = watch.owner_hex().parse::
() else { + host.log( + Level::WARN, + &format!( + "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)? + { + host.log( + Level::INFO, + &format!("{uid} already submitted; skipping re-submit"), + ); + return Ok(()); + } + + let creation = match build_order_creation(order, signature, owner) { + Ok(creation) => creation, + Err(message) => { + host.log( + Level::WARN, + &format!("submit skipped for {owner:#x}: {message}"), + ); + return Ok(()); + } + }; + let body = match serde_json::to_vec(&creation) { + Ok(body) => body, + Err(e) => { + host.log( + Level::ERROR, + &format!("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 + { + host.log( + Level::WARN, + &format!( + "UID divergence: client={client} server={server_uid} \ + (marker keyed on the client UID)" + ), + ); + } + host.log(Level::INFO, &format!("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)?; + } + host.log( + Level::INFO, + &format!( + "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)?; + let label: &'static str = action.into(); + host.log( + Level::WARN, + &format!("submit failed ({err}); retry action {label}"), + ); + } + } + 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..d3190991 --- /dev/null +++ b/crates/shepherd-sdk/tests/run.rs @@ -0,0 +1,422 @@ +//! 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 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)) + }; + run(&host, &source, &sample_tick()).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!(host.logging.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:"))); +} From 06c9b077a9927720d23cb50fd2c789cc57e2a10b Mon Sep 17 00:00:00 2001 From: mfw78 Date: Mon, 6 Jul 2026 16:33:17 +0000 Subject: [PATCH 2/9] docs(sdk): disambiguate the run intra-doc link --- crates/shepherd-sdk/src/cow/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/shepherd-sdk/src/cow/mod.rs b/crates/shepherd-sdk/src/cow/mod.rs index 9f616d1f..893a80e6 100644 --- a/crates/shepherd-sdk/src/cow/mod.rs +++ b/crates/shepherd-sdk/src/cow/mod.rs @@ -3,7 +3,7 @@ //! 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`), plus [`run`] - the +//! `PollOutcome`, `RetryAction`), plus [`run()`] - the //! poll/submit composition over the keeper stores. //! //! The codec submodules stay purely host-neutral: helpers take From 149c7d2f7a554d6237ef586642ad5c7abcece08d Mon Sep 17 00:00:00 2001 From: mfw78 Date: Wed, 8 Jul 2026 13:59:22 +0000 Subject: [PATCH 3/9] fix(cow): build the classify_poll_error test RpcError data as Bytes RpcError.data is now Bytes; the test helper takes raw Vec and wraps it (Vec -> Bytes is O(1)). --- crates/shepherd-sdk/src/cow/composable.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/shepherd-sdk/src/cow/composable.rs b/crates/shepherd-sdk/src/cow/composable.rs index 44a96464..47aa22ee 100644 --- a/crates/shepherd-sdk/src/cow/composable.rs +++ b/crates/shepherd-sdk/src/cow/composable.rs @@ -220,7 +220,7 @@ mod tests { ChainError::Rpc(RpcError { code: -32000, message: "execution reverted".into(), - data, + data: data.map(Into::into), }) } From 752163ab91788c2813430e504073caca73b9f505 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Mon, 6 Jul 2026 16:57:10 +0000 Subject: [PATCH 4/9] feat(sdk): log run diagnostics through the tracing facade The run logged through the LoggingHost seam, which the guest tracing capture cannot observe - module tests proving behaviour identity for keeper ports assert on tracing events. Route the submit-path diagnostics through the tracing macros with the wording the legacy twap poll loop established, and give ConditionalSource a defaulted label so shared log lines attribute the strategy that produced them. --- Cargo.lock | 2 + crates/nexum-sdk/src/keeper.rs | 7 +++ crates/shepherd-sdk/Cargo.toml | 4 ++ crates/shepherd-sdk/src/cow/run.rs | 76 ++++++++++++++---------------- crates/shepherd-sdk/tests/run.rs | 6 ++- 5 files changed, 52 insertions(+), 43 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index fa9ae472..81e5970e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5101,11 +5101,13 @@ dependencies = [ "alloy-sol-types", "cowprotocol", "nexum-sdk", + "nexum-sdk-test", "proptest", "serde_json", "shepherd-sdk-test", "strum", "thiserror 2.0.18", + "tracing", ] [[package]] diff --git a/crates/nexum-sdk/src/keeper.rs b/crates/nexum-sdk/src/keeper.rs index 88edc6de..ff78e082 100644 --- a/crates/nexum-sdk/src/keeper.rs +++ b/crates/nexum-sdk/src/keeper.rs @@ -331,6 +331,13 @@ pub trait ConditionalSource { /// 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 diff --git a/crates/shepherd-sdk/Cargo.toml b/crates/shepherd-sdk/Cargo.toml index 717d5c5d..2e4547f2 100644 --- a/crates/shepherd-sdk/Cargo.toml +++ b/crates/shepherd-sdk/Cargo.toml @@ -19,8 +19,12 @@ 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 diff --git a/crates/shepherd-sdk/src/cow/run.rs b/crates/shepherd-sdk/src/cow/run.rs index a7412ca5..cd688170 100644 --- a/crates/shepherd-sdk/src/cow/run.rs +++ b/crates/shepherd-sdk/src/cow/run.rs @@ -11,14 +11,17 @@ //! //! Store faults abort the sweep (the next tick replays it); //! submission failures never do - they classify into a -//! [`RetryAction`](super::RetryAction), the ledger applies the -//! effect, and the sweep moves on. +//! [`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::Level; use nexum_sdk::host::Fault; -use nexum_sdk::keeper::{ConditionalSource, Gates, Journal, Retrier, Tick, WatchRef, WatchSet}; +use nexum_sdk::keeper::{ + ConditionalSource, Gates, Journal, Retrier, RetryAction, Tick, WatchRef, WatchSet, +}; use super::{ CowApiError, CowHost, PollOutcome, classify_submit_error, gpv2_to_order_data, @@ -47,7 +50,7 @@ where }; match source.poll(host, watch, ¶ms, tick) { PollOutcome::Ready { order, signature } => { - submit_ready(host, watch, &order, signature, tick)?; + submit_ready(host, watch, &order, signature, tick, source.label())?; } PollOutcome::TryNextBlock => {} PollOutcome::TryOnBlock(block) => gates.set_next_block(watch, block)?, @@ -71,14 +74,12 @@ fn submit_ready( order: &GPv2OrderData, signature: Bytes, tick: &Tick, + label: &str, ) -> Result<(), Fault> { let Ok(owner) = watch.owner_hex().parse::
() else { - host.log( - Level::WARN, - &format!( - "watch {} carries an unparseable owner; skipping submit", - watch.key(), - ), + tracing::warn!( + "watch {} carries an unparseable owner; skipping submit", + watch.key(), ); return Ok(()); }; @@ -88,30 +89,21 @@ fn submit_ready( if let Some(uid) = client_uid.as_deref() && journal.contains(uid)? { - host.log( - Level::INFO, - &format!("{uid} already submitted; skipping re-submit"), - ); + 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) => { - host.log( - Level::WARN, - &format!("submit skipped for {owner:#x}: {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) => { - host.log( - Level::ERROR, - &format!("OrderCreation JSON encode failed: {e}"), - ); + tracing::error!("OrderCreation JSON encode failed: {e}"); return Ok(()); } }; @@ -126,15 +118,12 @@ fn submit_ready( if let Some(client) = client_uid.as_deref() && client != server_uid { - host.log( - Level::WARN, - &format!( - "UID divergence: client={client} server={server_uid} \ - (marker keyed on the client UID)" - ), + tracing::warn!( + "{label} UID divergence: client={client} server={server_uid} \ + (marker keyed on the client UID)" ); } - host.log(Level::INFO, &format!("submitted {marker}")); + tracing::info!("submitted {marker}"); } Err(CowApiError::Rejected(rejection)) if is_already_submitted(&rejection) => { // Success wearing an error status: the orderbook already @@ -143,22 +132,27 @@ fn submit_ready( if let Some(uid) = client_uid.as_deref() { journal.record(uid)?; } - host.log( - Level::INFO, - &format!( - "orderbook already holds this order ({}); receipt recorded", - rejection.error_type, - ), + 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)?; - let label: &'static str = action.into(); - host.log( - Level::WARN, - &format!("submit failed ({err}); retry action {label}"), - ); + 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(()) diff --git a/crates/shepherd-sdk/tests/run.rs b/crates/shepherd-sdk/tests/run.rs index d3190991..d5088626 100644 --- a/crates/shepherd-sdk/tests/run.rs +++ b/crates/shepherd-sdk/tests/run.rs @@ -10,6 +10,7 @@ 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; @@ -253,7 +254,8 @@ fn ready_marker_keys_on_the_client_uid_when_the_server_diverges() { let order = order.clone(); src(move |_, _, _, _| ready_outcome(&order)) }; - run(&host, &source, &sample_tick()).unwrap(); + 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)))); @@ -261,7 +263,7 @@ fn ready_marker_keys_on_the_client_uid_when_the_server_diverges() { !snapshot.contains_key("submitted:0xfeedface"), "marker must key on the client UID, not the divergent server UID", ); - assert!(host.logging.contains("UID divergence")); + assert!(logs.any(|e| e.message.contains("UID divergence"))); } #[test] From a56902abfea3e3988b9f903aaa445a17e4246432 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Mon, 6 Jul 2026 16:57:24 +0000 Subject: [PATCH 5/9] refactor(twap-monitor): port the poll loop onto the keeper composition Reduce strategy.rs to decode and evaluate: log decoding persists watches through the keeper watch set, and the getTradeableOrderWithSignature evaluation moves behind ConditionalSource. The hand-rolled gate keys, watch-update lifecycle, submitted: markers, and submit-retry dispatch are deleted in favour of cow::run over the keeper stores and retry ledger, still on the legacy CowApiHost seam. The MockHost dispatch tests are untouched - the behaviour-identity proof for the port. --- Cargo.lock | 2 - modules/twap-monitor/Cargo.toml | 4 +- modules/twap-monitor/src/strategy.rs | 504 ++++----------------------- 3 files changed, 72 insertions(+), 438 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 81e5970e..61076120 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5783,8 +5783,6 @@ dependencies = [ "serde_json", "shepherd-sdk", "shepherd-sdk-test", - "strum", - "thiserror 2.0.18", "tracing", "wit-bindgen 0.59.0", ] 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 From bf4d14e780cc49d85ff3700fe496178e8e0a43db Mon Sep 17 00:00:00 2001 From: mfw78 Date: Mon, 6 Jul 2026 17:28:08 +0000 Subject: [PATCH 6/9] feat(sdk-test): namespace the mock local store and share its limits Rework MockLocalStore to mirror the runtime store's shape: namespaced views over one shared row map, so identical key strings in sibling namespaces never collide, plus store-wide entry and byte limits that span namespaces the way one database file does. Fault injection stays per-view. --- crates/nexum-sdk-test/src/lib.rs | 222 ++++++++++++++++++++++++++----- 1 file changed, 190 insertions(+), 32 deletions(-) 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(); From 02e6276ff0e1cca799bd24eb40dcf3c48a8e6d9d Mon Sep 17 00:00:00 2001 From: mfw78 Date: Mon, 6 Jul 2026 17:28:08 +0000 Subject: [PATCH 7/9] feat(sdk-test): add the programmable MockVenue on the cow-api seam Script per-call venue behaviour where MockCowApi replays one response: a strictly-draining queue of submit outcomes with a configurable steady-state fallback, per-route response sequences whose final entry sticks (a terminal order status persists across re-polls), and venue fault injection that overrides both without consuming them. MockHost grows a defaulted venue type parameter so the composed host carries either mock on the same CowApiHost seam. Acceptance tests drive the keeper run across multi-tick retry, backoff, and outage scenarios, and module-shaped strategy code against status sequences. --- Cargo.lock | 2 + crates/shepherd-sdk-test/Cargo.toml | 6 + crates/shepherd-sdk-test/src/lib.rs | 331 +++++++++++++++++- crates/shepherd-sdk-test/tests/mock_venue.rs | 348 +++++++++++++++++++ 4 files changed, 679 insertions(+), 8 deletions(-) create mode 100644 crates/shepherd-sdk-test/tests/mock_venue.rs diff --git a/Cargo.lock b/Cargo.lock index 61076120..23e6f02d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5114,6 +5114,8 @@ dependencies = [ name = "shepherd-sdk-test" version = "0.1.0" dependencies = [ + "alloy-primitives", + "cowprotocol", "nexum-sdk", "nexum-sdk-test", "serde_json", 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), + ); +} From 8ec494b5e462acfdba03a4316b04c336e082866f Mon Sep 17 00:00:00 2001 From: Luiz Gustavo Abou Hatem de Liz Date: Thu, 9 Jul 2026 18:40:56 -0300 Subject: [PATCH 8/9] feat(chain): cap JSON-RPC response size before lowering into the guest (#154) Adds a configurable `[limits.chain] response_max_bytes` knob (default 1 MiB) that is enforced host-side in `chain::request()` before the response string is copied into the guest heap. Oversized responses are rejected with an `InvalidInput` fault, a `WARN` log, and a dedicated `shepherd_chain_response_capped_total` metric. Co-Authored-By: Claude Sonnet 4.6 Claude-Session: https://claude.ai/code/session_01TfSsJPYDQ1GQGbnSVFxATx --- crates/nexum-runtime/src/engine_config.rs | 32 ++++++++++ crates/nexum-runtime/src/host/impls/chain.rs | 62 +++++++++++++++++++- crates/nexum-runtime/src/host/state.rs | 3 + crates/nexum-runtime/src/supervisor.rs | 8 +++ 4 files changed, 104 insertions(+), 1 deletion(-) diff --git a/crates/nexum-runtime/src/engine_config.rs b/crates/nexum-runtime/src/engine_config.rs index b93a0096..66d0ed13 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_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,13 @@ impl ModuleLimits { self.memory_bytes.unwrap_or(DEFAULT_MEMORY_LIMIT) } + /// Resolved chain response size cap (override or default). + pub fn chain_response_max_bytes(&self) -> usize { + self.chain + .response_max_bytes + .unwrap_or(DEFAULT_CHAIN_RESPONSE_MAX_BYTES) + } + /// Resolved outbound HTTP limits (overrides or defaults). pub fn http(&self) -> OutboundHttpLimits { OutboundHttpLimits { @@ -376,6 +395,19 @@ 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_max_bytes = 1_048_576 +/// ``` +#[derive(Debug, Default, Deserialize)] +pub struct ChainLimitsSection { + /// Cap on one chain JSON-RPC response body, in bytes. + pub response_max_bytes: Option, +} + /// Resolved outbound HTTP limits the wasi:http gate enforces per /// request. Built by [`ModuleLimits::http`]. #[derive(Debug, Clone, Copy)] diff --git a/crates/nexum-runtime/src/host/impls/chain.rs b/crates/nexum-runtime/src/host/impls/chain.rs index 3fb67286..83cde88e 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!( @@ -283,4 +321,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) From 74e6866152e096307a10cc6c06e33a9bcaf20e18 Mon Sep 17 00:00:00 2001 From: Luiz Gustavo Abou Hatem de Liz Date: Fri, 10 Jul 2026 16:21:07 -0300 Subject: [PATCH 9/9] fix(#154): bound batch aggregates, harden config, test the real wiring Review follow-ups on the chain response cap: - request_batch: the per-entry cap bounds each body, but the aggregate Vec lowered into guest memory was unbounded - ~64 entries just under the cap is ~64 MiB, exactly the guest-heap saturation issue #154 exists to prevent, reachable via the block-range chunking the issue itself recommends. A running total now converts entries past the cap into typed invalid-input results. - config: renamed [limits.chain].response_max_bytes to response_body_max_bytes (u64) for exact symmetry with the [limits.http] sibling; a degenerate 0 saturates to 1, matching the logs/poison zero handling. - new harness test drives the cap through the real path - config -> ModuleLimits -> HostState -> chain::request - with the price-alert module: an over-cap oracle response surfaces to the guest as the typed fault and never reaches classify. The existing unit tests only covered the free function, so a deleted and_then would have passed the suite. - TOML parse tests for [limits.chain] (absent -> 1 MiB default, override, zero saturation), mirroring the HTTP section's tests. - engine.example.toml documents the new section; warn message dash matches house style. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TfSsJPYDQ1GQGbnSVFxATx --- crates/nexum-runtime/src/engine_config.rs | 53 +++++++++-- crates/nexum-runtime/src/host/impls/chain.rs | 38 +++++++- .../nexum-runtime/src/test_utils/harness.rs | 88 +++++++++++++++++++ engine.example.toml | 9 ++ 4 files changed, 180 insertions(+), 8 deletions(-) diff --git a/crates/nexum-runtime/src/engine_config.rs b/crates/nexum-runtime/src/engine_config.rs index 66d0ed13..35b5d7de 100644 --- a/crates/nexum-runtime/src/engine_config.rs +++ b/crates/nexum-runtime/src/engine_config.rs @@ -258,7 +258,7 @@ fn clamp_http_ms(ms: u64) -> Duration { /// response_body_max_bytes = 16_777_216 /// /// [limits.chain] -/// response_max_bytes = 1_048_576 +/// response_body_max_bytes = 1_048_576 /// /// [limits.logs] /// bytes_per_run = 262_144 @@ -299,10 +299,14 @@ impl ModuleLimits { self.memory_bytes.unwrap_or(DEFAULT_MEMORY_LIMIT) } - /// Resolved chain response size cap (override or default). + /// 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_max_bytes + .response_body_max_bytes + .map(|b| (b.max(1)) as usize) .unwrap_or(DEFAULT_CHAIN_RESPONSE_MAX_BYTES) } @@ -400,12 +404,13 @@ pub struct HttpLimitsSection { /// /// ```toml /// [limits.chain] -/// response_max_bytes = 1_048_576 +/// response_body_max_bytes = 1_048_576 /// ``` #[derive(Debug, Default, Deserialize)] pub struct ChainLimitsSection { - /// Cap on one chain JSON-RPC response body, in bytes. - pub response_max_bytes: Option, + /// 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 @@ -719,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 83cde88e..f9afad2a 100644 --- a/crates/nexum-runtime/src/host/impls/chain.rs +++ b/crates/nexum-runtime/src/host/impls/chain.rs @@ -39,7 +39,7 @@ fn check_response_cap( method, body_bytes = body.len(), cap_bytes = cap, - "chain response exceeds size cap — rejecting before guest copy" + "chain response exceeds size cap - rejecting before guest copy" ); metrics::counter!( "shepherd_chain_response_capped_total", @@ -128,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)), } } 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/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.