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 2ffdc075681722d0d7ee9bc610497505df6cba2a Mon Sep 17 00:00:00 2001 From: Luiz Gustavo Abou Hatem de Liz Date: Thu, 9 Jul 2026 19:25:31 -0300 Subject: [PATCH 8/9] =?UTF-8?q?docs:=20stale-reference=20sweep=20=E2=80=94?= =?UTF-8?q?=20engine=20rename,=20Dockerfile,=20ADR-0007=20errata=20(#123,?= =?UTF-8?q?=20#108,=20#62)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - deployment.md: replace the "planned for M5" Dockerfile sketch with a pointer to the real Dockerfile + docker.md; update nexum-runtime log directive examples - qa-signoff.md: nexum-engine → nexum-runtime in crate-dependency row - adr/0007: add errata noting the nexum-engine → nexum-runtime rename and the [patch.crates-io] retirement now that cowprotocol 0.1.0 is on crates.io Co-Authored-By: Claude Sonnet 4.6 Claude-Session: https://claude.ai/code/session_01TfSsJPYDQ1GQGbnSVFxATx --- .../0007-upstream-protocol-logic-to-cow-rs.md | 2 ++ docs/deployment.md | 29 +++++++------------ docs/qa-signoff.md | 2 +- 3 files changed, 14 insertions(+), 19 deletions(-) diff --git a/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md b/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md index 3a828f5e..1f28bb2a 100644 --- a/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md +++ b/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md @@ -44,3 +44,5 @@ Lower-priority follow-ons (`OrderUid::from_slice`, retry middleware on `OrderBoo - Guest modules consume `cowprotocol` types directly (gated on the wasm32 feature in item 3). The `shepherd-sdk` crate in M3 may add ergonomic wrappers on top, but those live on the guest side, not behind a WIT boundary. - A follow-on Bleu module - the Rust-side equivalent of `cowprotocol/refunder` (permissionless `invalidateOrder` triggering for expired EthFlow orders) - becomes natural to ship once an ethflow-watcher module lands. Out of scope for M2 but explicitly enabled by the same primitives. - TWAP polling logic (decode `ConditionalOrderCreated`, eth_call `getTradeableOrderWithSignature`, decode return, build `OrderCreation`) and EthFlow event decoding stay entirely in guest module code. The `cowprotocol` crate provides only the types and the orderbook client; the strategy is the module's. + +_Errata: `crates/nexum-engine` was renamed to `crates/nexum-runtime` + `crates/nexum-cli` in the 0.2 refactor. The `[patch.crates-io]` rev-bump mechanism was retired once `cowprotocol 0.1.0` published to crates.io; the workspace now pins it as a normal registry dependency with no git override._ diff --git a/docs/deployment.md b/docs/deployment.md index 2525e2f9..955e0d96 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -151,26 +151,19 @@ For systemd-style production runs, see `docs/production.md`. ## Docker -A reference Dockerfile + Compose file is planned for M5. -Until that lands, build manually: - -```dockerfile -# (sketch — full Dockerfile is planned for M5) -FROM rust:1.91 as build -COPY . /src -WORKDIR /src -RUN cargo build --release -p nexum-cli -RUN rustup target add wasm32-wasip2 \ - && cargo build --target wasm32-wasip2 --release \ - -p twap-monitor -p ethflow-watcher - -FROM gcr.io/distroless/cc-debian12 -COPY --from=build /src/target/release/nexum /usr/local/bin/ -COPY --from=build /src/target/wasm32-wasip2/release/*.wasm /modules/ -COPY engine.toml /etc/shepherd/engine.toml -ENTRYPOINT ["/usr/local/bin/nexum"] +A `Dockerfile` + `docker-compose.yml` ship at the repo root. See +[`docs/deployment/docker.md`](deployment/docker.md) for the full +container workflow. The quick start: + +```sh +cp .env.example .env +$EDITOR .env # paste wss:// RPC URLs +docker compose up -d ``` +The image is published to +`ghcr.io/nullislabs/shepherd:` on every merged commit to `main`. + Mount the `state_dir` as a volume so the redb file survives container restarts. diff --git a/docs/qa-signoff.md b/docs/qa-signoff.md index 808d03e6..8953a78b 100644 --- a/docs/qa-signoff.md +++ b/docs/qa-signoff.md @@ -12,7 +12,7 @@ | `cargo test --workspace` | ✅ | 145 host tests + 1 doctest passing. | | Em-dashes in `crates/`, `modules/`, `docs/` | ✅ | 0. One was in `price-alert/strategy.rs:4` (mine), fixed. | | Em-dashes in `wit/**.wit` | ⚠ | 3 in mfw78's M1 prose. Intentionally left alone; flag for him in upstream review. | -| `warn(unused_crate_dependencies)` on every crate root | ✅ | sdk, sdk-test, nexum-engine, twap, ethflow, price-alert, balance-tracker, stop-loss. | +| `warn(unused_crate_dependencies)` on every crate root | ✅ | sdk, sdk-test, nexum-runtime, twap, ethflow, price-alert, balance-tracker, stop-loss. | | WASM build (`wasm32-wasip2 --release`) | ✅ | All 5 modules build. Sizes: twap 314 KB, ethflow 282 KB, stop-loss 311 KB, price-alert 215 KB, balance-tracker 102 KB. | | String-wrapped errors outside WIT boundary | ✅ | All hits in `crates/nexum-runtime/src/host/impls/*` (FFI boundary - exception per rust-idiomatic skill). No leaks in SDK or modules. | From a9c0c5041dd02ddebe0dfb85618935dae2595b75 Mon Sep 17 00:00:00 2001 From: Luiz Gustavo Abou Hatem de Liz Date: Fri, 10 Jul 2026 16:28:28 -0300 Subject: [PATCH 9/9] docs: fix the sweep's own stale claims - cowprotocol override, production.md 3.1 Review catches on the stale-reference sweep itself: - ADR-0007 errata and sdk.md Versioning both claimed cowprotocol is a clean 0.1.0 registry dependency "with no git override". The branch's own Cargo.toml pins 0.2.0 WITH an active [patch.crates-io] override to nullislabs/cow-rs (pending the hash-only OrderCreationAppData constructor release). Both now describe that state and point at the patch-block comment as the source of truth. - production.md 3.1 still shipped the "interim" hand-rolled Dockerfile recipe issue #123 flagged; replaced with a pointer to the real root Dockerfile and docs/deployment/docker.md. - 02-modules-events-packaging.md: un-garble the parenthetical the earlier rename made self-contradictory ("was module.toml in earlier drafts" -> "was nexum.toml"). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01TfSsJPYDQ1GQGbnSVFxATx --- docs/02-modules-events-packaging.md | 2 +- .../0007-upstream-protocol-logic-to-cow-rs.md | 2 +- docs/production.md | 47 ++++--------------- docs/sdk.md | 11 +++-- 4 files changed, 17 insertions(+), 45 deletions(-) diff --git a/docs/02-modules-events-packaging.md b/docs/02-modules-events-packaging.md index 430d4114..dc8dfb87 100755 --- a/docs/02-modules-events-packaging.md +++ b/docs/02-modules-events-packaging.md @@ -6,7 +6,7 @@ A module is distributed as a **bundle** - a WASM component plus a manifest that ### Manifest (`module.toml`) -Every module ships with a manifest. The file is named `module.toml` in 0.2 (was `module.toml` in earlier drafts; per [ADR-0001](adr/0001-engine-toml-separate-from-nexum-toml.md) the operator/module split is now explicit). +Every module ships with a manifest. The file is named `module.toml` in 0.2 (was `nexum.toml` in earlier drafts; per [ADR-0001](adr/0001-engine-toml-separate-from-nexum-toml.md) the operator/module split is now explicit). ```toml [module] diff --git a/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md b/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md index 1f28bb2a..9ff06fa3 100644 --- a/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md +++ b/docs/adr/0007-upstream-protocol-logic-to-cow-rs.md @@ -45,4 +45,4 @@ Lower-priority follow-ons (`OrderUid::from_slice`, retry middleware on `OrderBoo - A follow-on Bleu module - the Rust-side equivalent of `cowprotocol/refunder` (permissionless `invalidateOrder` triggering for expired EthFlow orders) - becomes natural to ship once an ethflow-watcher module lands. Out of scope for M2 but explicitly enabled by the same primitives. - TWAP polling logic (decode `ConditionalOrderCreated`, eth_call `getTradeableOrderWithSignature`, decode return, build `OrderCreation`) and EthFlow event decoding stay entirely in guest module code. The `cowprotocol` crate provides only the types and the orderbook client; the strategy is the module's. -_Errata: `crates/nexum-engine` was renamed to `crates/nexum-runtime` + `crates/nexum-cli` in the 0.2 refactor. The `[patch.crates-io]` rev-bump mechanism was retired once `cowprotocol 0.1.0` published to crates.io; the workspace now pins it as a normal registry dependency with no git override._ +_Errata: `crates/nexum-engine` was renamed to `crates/nexum-runtime` + `crates/nexum-cli` in the 0.2 refactor. The `cowprotocol` crate now publishes to crates.io (the workspace pins `0.2.0`), so the PR-#5-head consumption model above is historical; a `[patch.crates-io]` git override to `nullislabs/cow-rs` remains active pending a release with the hash-only `OrderCreationAppData` constructor - see the comment above the patch block in the root `Cargo.toml` for the current state._ diff --git a/docs/production.md b/docs/production.md index c2b22499..f845f80a 100644 --- a/docs/production.md +++ b/docs/production.md @@ -140,45 +140,14 @@ journalctl -u shepherd -f --output=json | jq '.MESSAGE | fromjson?' ## 3. Container deploy: Docker Compose -> **Status note:** the official Dockerfile is tracked as a -> separate issue. Until it lands, build the image locally with -> the multi-stage recipe below; the Compose file is forward- -> compatible with the eventual published image. - -### 3.1 Dockerfile (interim) - -```dockerfile -# syntax=docker/dockerfile:1.6 -FROM rust:1.86-slim-bookworm AS build -WORKDIR /src -RUN apt-get update && apt-get install -y --no-install-recommends \ - pkg-config libssl-dev cmake clang \ - && rm -rf /var/lib/apt/lists/* -RUN rustup target add wasm32-wasip2 -COPY . . -RUN cargo build -p nexum-cli --release -# Build all 5 modules. Add yours here. -RUN cargo build -p twap-monitor --target wasm32-wasip2 --release \ - && cargo build -p ethflow-watcher --target wasm32-wasip2 --release \ - && cargo build -p price-alert --target wasm32-wasip2 --release \ - && cargo build -p balance-tracker --target wasm32-wasip2 --release \ - && cargo build -p stop-loss --target wasm32-wasip2 --release - -FROM debian:bookworm-slim AS runtime -RUN apt-get update && apt-get install -y --no-install-recommends \ - ca-certificates tini \ - && rm -rf /var/lib/apt/lists/* \ - && useradd -r -s /usr/sbin/nologin -d /var/lib/shepherd shepherd \ - && install -d -o shepherd -g shepherd /var/lib/shepherd -COPY --from=build /src/target/release/nexum /usr/local/bin/ -COPY --from=build /src/target/wasm32-wasip2/release/*.wasm /opt/shepherd/modules/ -COPY --from=build /src/modules /opt/shepherd/manifests -USER shepherd -WORKDIR /var/lib/shepherd -EXPOSE 9100 -ENTRYPOINT ["/usr/bin/tini", "--", "nexum"] -CMD ["--engine-config", "/etc/shepherd/engine.toml"] -``` +### 3.1 Dockerfile + +The official multi-stage `Dockerfile` ships at the repo root - it +builds the `nexum` binary, compiles all five module wasms, and runs as +a non-root user under `tini`. Build it with `docker build -t shepherd .` +or let the root `docker-compose.yml` build it for you. The full image +reference (published tags, pinning, .env wiring) is in +[`docs/deployment/docker.md`](deployment/docker.md). ### 3.2 docker-compose.yml diff --git a/docs/sdk.md b/docs/sdk.md index e752c5fa..9fb6b409 100644 --- a/docs/sdk.md +++ b/docs/sdk.md @@ -109,7 +109,10 @@ The SDK crates are currently `0.1.0` and live at `crates/nexum-sdk/` and `crates/shepherd-sdk/` in the shepherd monorepo. They are not yet published to crates.io; modules depend on them via workspace paths. -The `cowprotocol` crate is published to crates.io at `0.1.0`; the -workspace declares `cowprotocol = "0.1.0"` in `[workspace.dependencies]` -with no `[patch.crates-io]` or git overrides. Module Cargo.toml files -that inherit from the workspace pick it up automatically. +The `cowprotocol` crate is published to crates.io; the workspace +declares `cowprotocol = "0.2.0"` in `[workspace.dependencies]`, with a +temporary `[patch.crates-io]` git override to `nullislabs/cow-rs` +pending a release that ships the hash-only `OrderCreationAppData` +constructor (see the comment above the patch block in the root +`Cargo.toml`). Module Cargo.toml files that inherit from the workspace +pick it up automatically.