diff --git a/Cargo.lock b/Cargo.lock index 9dc66e23..f3c42cd6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5246,6 +5246,7 @@ dependencies = [ "strum", "thiserror 2.0.18", "tracing", + "videre-sdk", ] [[package]] diff --git a/crates/shepherd-sdk-test/tests/mock_venue.rs b/crates/shepherd-sdk-test/tests/mock_venue.rs index d8af0612..ec371ca2 100644 --- a/crates/shepherd-sdk-test/tests/mock_venue.rs +++ b/crates/shepherd-sdk-test/tests/mock_venue.rs @@ -9,8 +9,8 @@ use cowprotocol::{BuyTokenDestination, GPv2OrderData, OrderKind, SellTokenSource use nexum_sdk::host::{Fault, LocalStoreHost as _, RateLimit}; use nexum_sdk::keeper::{ConditionalSource, Journal, Tick, WatchRef, WatchSet, watch_key}; use shepherd_sdk::cow::{ - CowApiError, CowHost, CowIntent, CowIntentBody, OrderRejection, SignedOrder, - gpv2_to_order_data, order_data_to_body, order_uid_hex, run, + CowApiError, CowApiTransport, CowClient, CowHost, CowIntent, CowIntentBody, OrderRejection, + SignedOrder, gpv2_to_order_data, order_data_to_body, order_uid_hex, run, }; use shepherd_sdk_test::{MockHost, MockVenue}; @@ -18,6 +18,12 @@ const SEPOLIA: u64 = 11_155_111; type VenueHost = MockHost; +/// The typed client every test drives `run` with: the transitional +/// cow-api bridge over the scripted venue host. +fn venue(host: &VenueHost) -> CowClient> { + CowClient::with_transport(CowApiTransport::new(host, SEPOLIA)) +} + /// Closure-backed source so each test scripts its own outcome. struct FnSource(F); @@ -146,7 +152,7 @@ fn keeper_retries_a_transient_rejection_then_submits() { host.cow_api.enqueue_submit(Ok(client_uid(&order))); let source = ready_source(&order); - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); assert_eq!(host.cow_api.call_count(), 1); assert!(host.store.snapshot().contains_key(&key), "watch survives"); assert!( @@ -155,7 +161,7 @@ fn keeper_retries_a_transient_rejection_then_submits() { .unwrap() ); - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); assert_eq!(host.cow_api.call_count(), 2); assert!( Journal::submitted(&host) @@ -185,7 +191,7 @@ fn keeper_backs_off_on_rate_limit_and_submits_after_the_gate() { let t0 = sample_tick(); let source = ready_source(&order); - run(&host, &source, &t0).unwrap(); + run(&host, &venue(&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 @@ -194,14 +200,14 @@ fn keeper_backs_off_on_rate_limit_and_submits_after_the_gate() { epoch_s: t0.epoch_s + 2, ..t0 }; - run(&host, &source, &gated).unwrap(); + run(&host, &venue(&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(); + run(&host, &venue(&host), &source, &clear).unwrap(); assert_eq!(host.cow_api.call_count(), 2); assert!( Journal::submitted(&host) @@ -222,7 +228,7 @@ fn keeper_survives_a_venue_outage_and_submits_on_recovery() { .inject_fault(CowApiError::Fault(Fault::Unavailable("venue down".into()))); let source = ready_source(&order); - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); let snapshot = host.store.snapshot(); assert!(snapshot.contains_key(&key)); assert!(!snapshot.contains_key(&watch_key.next_block_key())); @@ -231,7 +237,7 @@ fn keeper_survives_a_venue_outage_and_submits_on_recovery() { host.cow_api.clear_fault(); host.cow_api.enqueue_submit(Ok(client_uid(&order))); - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); assert_eq!(host.cow_api.call_count(), 2); assert!( Journal::submitted(&host) @@ -249,7 +255,7 @@ fn keeper_drops_the_watch_on_a_scripted_permanent_rejection() { host.cow_api .enqueue_submit(Err(rejection("InvalidSignature"))); - run(&host, &ready_source(&order), &sample_tick()).unwrap(); + run(&host, &venue(&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); @@ -276,6 +282,7 @@ fn keeper_sweep_ignores_sibling_namespace_watches() { let polls = std::cell::Cell::new(0_u32); run( &host, + &venue(&host), &src(|_, _, _, _| { polls.set(polls.get() + 1); ready_outcome(&order) diff --git a/crates/shepherd-sdk/Cargo.toml b/crates/shepherd-sdk/Cargo.toml index b73fae63..a2c7f216 100644 --- a/crates/shepherd-sdk/Cargo.toml +++ b/crates/shepherd-sdk/Cargo.toml @@ -22,6 +22,9 @@ cow-venue = { path = "../cow-venue", features = ["client", "assembly"] } # The structured poll seam the keeper run dispatches on. composable-cow = { path = "../composable-cow" } nexum-sdk = { path = "../nexum-sdk" } +# The typed client seam the keeper run submits through; also the +# `VenueTransport` contract the legacy cow-api bridge implements. +videre-sdk = { path = "../videre-sdk" } cowprotocol = { version = "0.2.0", default-features = false } alloy-primitives.workspace = true serde_json.workspace = true diff --git a/crates/shepherd-sdk/src/cow/mod.rs b/crates/shepherd-sdk/src/cow/mod.rs index 86ecdf03..8909134c 100644 --- a/crates/shepherd-sdk/src/cow/mod.rs +++ b/crates/shepherd-sdk/src/cow/mod.rs @@ -1,9 +1,10 @@ //! CoW Protocol bridging. //! //! ABI decoding helpers, the orderbook error surface, and [`run()`] - -//! the poll/submit composition over the keeper stores. The chain-edge -//! order projections live in the `cow-venue` `assembly` slice (the -//! venue adapter owns them) and are re-exported here. +//! the poll/submit composition over the keeper stores, submitting +//! through the typed [`CowClient`] on the `videre:venue/client` seam. +//! The chain-edge order projections live in the `cow-venue` `assembly` +//! slice (the venue adapter owns them) and are re-exported here. //! //! The poll seam is the structured //! [`Verdict`](composable_cow::Verdict), carried by the @@ -14,11 +15,14 @@ //! 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 -//! keeper run is generic over the host traits alone. +//! keeper run is generic over the host traits and the venue transport +//! alone; [`CowApiTransport`] carries it over the legacy +//! `shepherd:cow/cow-api` import until the module worlds flip. pub mod error; pub mod events; pub mod run; +pub mod transport; /// Chain-edge order assembly, re-exported from the `cow-venue` /// `assembly` slice the venue adapter owns. @@ -28,19 +32,23 @@ pub use error::{ classify_submit_error, is_already_submitted, }; pub use run::run; +pub use transport::CowApiTransport; /// The venue-neutral intent body types and their borsh `IntentBody` -/// codec, re-exported from the `cow-venue` default slice. The shim keeps -/// this path stable while the module ports move off the legacy surface. +/// codec, re-exported from the `cow-venue` default slice, plus the +/// typed CoW venue client. The shim keeps this path stable while the +/// module ports move off the legacy surface. pub use cow_venue::{ - BuyToken, BuyTokenDestination, CowIntent, CowIntentBody, OrderBody, OrderBuilder, OrderKind, - OrderUid, SellToken, SellTokenSource, SignedOrder, intent_id, + BuyToken, BuyTokenDestination, CowClient, CowIntent, CowIntentBody, CowVenue, OrderBody, + OrderBuilder, OrderKind, OrderUid, SellToken, SellTokenSource, SignedOrder, intent_id, }; use nexum_sdk::host::Host; -/// `shepherd:cow/cow-api` - orderbook submission path. The CoW-domain -/// sibling of the core host traits in [`nexum_sdk::host`]. +/// `shepherd:cow/cow-api` - the legacy orderbook submission path, +/// retiring. The keeper [`run()`] submits through the typed +/// [`CowClient`]; [`CowApiTransport`] bridges it onto this seam until +/// the module worlds flip to `videre:venue/client`. pub trait CowApiHost { /// Submit an `OrderCreation` JSON body. The host returns the /// canonical order UID on success. A rejection surfaces as a typed diff --git a/crates/shepherd-sdk/src/cow/run.rs b/crates/shepherd-sdk/src/cow/run.rs index 6ad79803..0a19bc8e 100644 --- a/crates/shepherd-sdk/src/cow/run.rs +++ b/crates/shepherd-sdk/src/cow/run.rs @@ -4,40 +4,43 @@ //! [`run`] walks the keeper watch set, polls each gate-ready //! watch through a [`ConditionalSource`], and runs the //! [`Verdict`]'s effect: lifecycle outcomes update the gate and -//! watch stores, `Post` drives one submission through the -//! [`CowApiHost`](super::CowApiHost) seam with the `submitted:` -//! journal as the idempotency guard - keyed on the venue-and-body -//! [`intent_id`] - and the keeper [`Retrier`] as the failure -//! dispatch. +//! watch stores, `Post` drives one submission through the typed +//! [`CowClient`] onto the `videre:venue/client` seam with the +//! `submitted:` journal as the idempotency guard - keyed on the +//! venue-and-body [`intent_id`] - and the keeper [`Retrier`] +//! as the failure dispatch. //! //! Store faults abort the sweep (the next tick replays it); -//! submission failures never do - they classify into a -//! [`RetryAction`], the ledger applies the effect, and the sweep -//! moves on. Diagnostics go through the guest `tracing` facade - +//! submission failures never do - they fold into a +//! [`RetryAction`] through the videre +//! [`retry_action`] table, 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 alloy_primitives::{Address, Bytes, hex}; use composable_cow::Verdict; -use cow_venue::assembly::build_order_creation; use cowprotocol::GPv2OrderData; -use nexum_sdk::host::Fault; +use nexum_sdk::host::{Fault, LocalStoreHost}; use nexum_sdk::keeper::{ ConditionalSource, Gates, Journal, Retrier, RetryAction, Tick, WatchRef, WatchSet, }; +use videre_sdk::keeper::retry_action; +use videre_sdk::{ClientError, SubmitOutcome, VenueTransport, rt}; use super::{ - CowApiError, CowHost, CowIntent, CowIntentBody, SignedOrder, classify_submit_error, - gpv2_to_order_data, intent_id, is_already_submitted, order_data_to_body, + CowClient, CowIntent, CowIntentBody, SignedOrder, gpv2_to_order_data, intent_id, + order_data_to_body, }; /// Poll every gate-ready watch once at `tick` and run each outcome's /// effect. One source poll per ready watch; a `Post` outcome makes at -/// most one `submit_order` call. -pub fn run(host: &H, source: &S, tick: &Tick) -> Result<(), Fault> +/// most one venue submit through `venue`. +pub fn run(host: &H, venue: &CowClient, source: &S, tick: &Tick) -> Result<(), Fault> where - H: CowHost, + H: LocalStoreHost, S: ConditionalSource, + T: VenueTransport, { let watches = WatchSet::new(host); let gates = Gates::new(host); @@ -55,7 +58,7 @@ where Verdict::Post { order, signature, .. } => { - submit_ready(host, watch, &order, signature, tick, source.label())?; + submit_ready(host, venue, watch, &order, signature, tick, source.label())?; } Verdict::TryNextBlock { .. } => {} Verdict::WaitBlock { wait_until, .. } => gates.set_next_block(watch, wait_until)?, @@ -74,24 +77,27 @@ where Ok(()) } -/// Submit one freshly-polled `Ready` order, guarding on the -/// `submitted:` journal and dispatching any failure through the retry -/// ledger. +/// Submit one freshly-polled `Ready` order through the typed client, +/// guarding on the `submitted:` journal and dispatching any venue +/// refusal through the retry ledger. /// -/// The journal keys on the deterministic venue-and-body -/// [`intent_id`], derived before any network work from the same body -/// bytes a venue submit carries - never from the assembled -/// `OrderCreation` - so the guard survives assembly moving into the -/// venue adapter. The orderbook's UID is the receipt; it rides the -/// log only. -fn submit_ready( +/// The journal keys on the deterministic venue-and-body [`intent_id`], +/// derived before any network work from the same body bytes the venue +/// submit carries, so the guard is independent of where assembly +/// happens. The venue's receipt rides the log only. +fn submit_ready( host: &H, + venue: &CowClient, watch: WatchRef<'_>, order: &GPv2OrderData, signature: Bytes, tick: &Tick, label: &str, -) -> Result<(), Fault> { +) -> Result<(), Fault> +where + H: LocalStoreHost, + T: VenueTransport, +{ let Ok(owner) = watch.owner_hex().parse::
() else { tracing::warn!( "watch {} carries an unparseable owner; skipping submit", @@ -127,31 +133,16 @@ fn submit_ready( tracing::info!("{label} {intent_id} already submitted; skipping re-submit"); return Ok(()); } - let creation = match build_order_creation(&order_data, &signature, owner) { - Ok(creation) => creation, - Err(err) => { - // A constructor rejection (zero `from`, `validTo` beyond - // the client-side max horizon) is deterministic for this - // polled payload: keeping the watch would re-poll and - // re-warn on every block forever. Drop through the ledger - // - the same net effect as the pre-keeper flow, where - // the orderbook rejected the shipped body and the - // classifier dropped the watch. - tracing::warn!("{label} submit dropped watch for {owner:#x}: {err}"); - Retrier::new(host).apply(watch, RetryAction::Drop, tick.epoch_s)?; - return Ok(()); - } - }; - let body = match serde_json::to_vec(&creation) { - Ok(body) => body, - Err(e) => { - tracing::error!("OrderCreation JSON encode failed: {e}"); - return Ok(()); - } - }; - match host.submit_order(tick.chain_id, &body) { - Ok(receipt) => { + let Some(outcome) = rt::complete(venue.submit(&intent)) else { + // Guest transports never suspend; a pending future means a + // foreign transport misbehaved. The watch stays for the next + // tick. + tracing::error!("{label} submit future suspended; retrying next tick"); + return Ok(()); + }; + match outcome { + Ok(SubmitOutcome::Accepted(receipt)) => { // The submit already succeeded; a journal-store fault here // must not abort the sweep or unwind the accepted order. // Log and carry on - the already-submitted arm keeps the @@ -159,41 +150,39 @@ fn submit_ready( if let Err(fault) = journal.record(&intent_id) { tracing::error!("submitted {intent_id} but journal write failed: {fault}"); } - tracing::info!("submitted {intent_id} (receipt {receipt})"); - } - Err(CowApiError::Rejected(rejection)) if is_already_submitted(&rejection) => { - // Success wearing an error status: the orderbook already - // holds this order. Journal the intent-id and keep the - // watch so the next tick short-circuits instead of - // re-posting. As above, a journal fault post-submit only - // forfeits the short-circuit; it must not abort the sweep. - if let Err(fault) = journal.record(&intent_id) { - tracing::error!( - "orderbook already holds {intent_id} but journal write failed: {fault}" - ); - } tracing::info!( - "orderbook already holds this order ({}); intent-id journalled", - rejection.error_type, + "submitted {intent_id} (receipt {})", + hex::encode_prefixed(&receipt), ); } - Err(err) => { - let action = classify_submit_error(&err); + Ok(SubmitOutcome::RequiresSigning(_)) => { + // A sweep cannot sign; nothing is journalled, so the next + // tick surfaces the same ask afresh. + tracing::warn!("{label} submit for {owner:#x} requires signing; not journalled"); + } + Err(ClientError::Body(err)) => { + tracing::error!("intent body encode failed: {err}"); + } + Err(ClientError::Venue(fault)) => { + let action = retry_action(&fault); Retrier::new(host).apply(watch, action, tick.epoch_s)?; match action { - RetryAction::TryNextBlock => tracing::warn!("submit retry-next-block: {err}"), + RetryAction::TryNextBlock => tracing::warn!("submit retry-next-block: {fault}"), RetryAction::Backoff { seconds } => { - tracing::warn!("submit backoff {seconds}s: {err}"); + tracing::warn!("submit backoff {seconds}s: {fault}"); } - RetryAction::Drop => tracing::warn!("submit dropped watch: {err}"), + RetryAction::Drop => tracing::warn!("submit dropped watch: {fault}"), // `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}"); + tracing::warn!("submit retry action {action_label}: {fault}"); } } } + // `ClientError` is non-exhaustive; a future case leaves the + // watch for the next tick. + Err(err) => tracing::error!("submit failed: {err}"), } Ok(()) } diff --git a/crates/shepherd-sdk/src/cow/transport.rs b/crates/shepherd-sdk/src/cow/transport.rs new file mode 100644 index 00000000..ece99f70 --- /dev/null +++ b/crates/shepherd-sdk/src/cow/transport.rs @@ -0,0 +1,226 @@ +//! Transitional venue transport over the legacy `shepherd:cow/cow-api` +//! seam. +//! +//! [`CowApiTransport`] implements the videre [`VenueTransport`] +//! contract by assembling the orderbook `OrderCreation` from the +//! decoded [`CowIntentBody`] and driving +//! [`CowApiHost::submit_order`], so the keeper [`run`](super::run()) +//! submits through the typed [`CowClient`](super::CowClient) while +//! module worlds still import the legacy host extension. Deleted when +//! the worlds flip to `videre:venue/client`. + +use alloy_primitives::{Address, hex}; +use cow_venue::assembly; +use cow_venue::body::{CowIntent, CowIntentBody}; +use cowprotocol::Chain; +use nexum_sdk::host::Fault; +use nexum_sdk::keeper::RetryAction; +use videre_sdk::client::sealed::SealedTransport; +use videre_sdk::{ + IntentBody as _, IntentStatus, Quotation, SubmitOutcome, VenueFault, VenueId, VenueTransport, +}; + +use super::{CowApiError, CowApiHost, classify_api_error, is_already_submitted}; + +/// The `videre:venue/client` verbs carried over the legacy +/// `shepherd:cow/cow-api` import: submit only, pre-bound to one chain's +/// orderbook. Quote, status, and cancel have no legacy submission-path +/// counterpart and refuse as `unsupported`. +pub struct CowApiTransport<'h, H> { + host: &'h H, + chain_id: u64, +} + +impl<'h, H: CowApiHost> CowApiTransport<'h, H> { + /// Bind the legacy seam to one chain's orderbook. + #[must_use] + pub const fn new(host: &'h H, chain_id: u64) -> Self { + Self { host, chain_id } + } +} + +impl SealedTransport for CowApiTransport<'_, H> {} + +impl VenueTransport for CowApiTransport<'_, H> { + async fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { + Err(VenueFault::Unsupported) + } + + async fn submit(&self, _venue: &VenueId, body: Vec) -> Result { + let CowIntentBody::V1(intent) = + CowIntentBody::from_bytes(&body).map_err(|e| VenueFault::InvalidBody(e.to_string()))?; + // The legacy seam posts EIP-1271 only; the pre-sign flow needs + // the adapter. + let CowIntent::Signed(signed) = intent else { + return Err(VenueFault::Unsupported); + }; + let order = assembly::body_to_order_data(&signed.order); + let owner = Address::from(signed.owner); + let creation = assembly::build_order_creation(&order, &signed.signature, owner) + .map_err(|e| VenueFault::InvalidBody(e.to_string()))?; + let json = serde_json::to_vec(&creation) + .map_err(|e| VenueFault::Unavailable(format!("order encode failed: {e}")))?; + match self.host.submit_order(self.chain_id, &json) { + Ok(uid) => Ok(SubmitOutcome::Accepted(receipt_bytes(&uid))), + // Already-held is success wearing an error status; the + // receipt is the client-derived UID (empty on a chain the + // SDK cannot derive for). + Err(CowApiError::Rejected(r)) if is_already_submitted(&r) => { + let receipt = Chain::try_from(self.chain_id) + .map(|chain| { + assembly::order_uid(chain, &order, owner) + .as_slice() + .to_vec() + }) + .unwrap_or_default(); + Ok(SubmitOutcome::Accepted(receipt)) + } + Err(err) => Err(venue_fault(&err)), + } + } + + async fn status(&self, _venue: &VenueId, _receipt: &[u8]) -> Result { + Err(VenueFault::Unsupported) + } + + async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { + Err(VenueFault::Unsupported) + } +} + +/// The server UID at its wire spelling; a non-hex receipt rides through +/// as raw bytes rather than failing an accepted submit. +fn receipt_bytes(uid: &str) -> Vec { + hex::decode(uid).unwrap_or_else(|_| uid.as_bytes().to_vec()) +} + +/// Project a legacy submission failure onto the venue fault the typed +/// client reports, mirroring the adapter: throttles keep their hint, +/// host and server failures stay retryable, and only a structured +/// rejection, folded through the shipped classification table, carries +/// a permanent venue verdict. +fn venue_fault(err: &CowApiError) -> VenueFault { + match err { + CowApiError::Fault(Fault::RateLimited(limit)) => VenueFault::RateLimited { + retry_after_ms: limit.retry_after_ms, + }, + CowApiError::Fault(Fault::Timeout) => VenueFault::Timeout, + // Any other host fault is infrastructure, not a venue verdict: + // it stays retryable so an unprovisioned capability or unknown + // chain never drops a still-valid order. + CowApiError::Fault(fault) => VenueFault::Unavailable(fault.to_string()), + CowApiError::Http(http) if http.status == 429 => VenueFault::RateLimited { + retry_after_ms: None, + }, + CowApiError::Http(http) => { + VenueFault::Unavailable(format!("orderbook http {}", http.status)) + } + CowApiError::Rejected(rejection) => { + let detail = format!("{}: {}", rejection.error_type, rejection.description); + match classify_api_error(rejection) { + RetryAction::TryNextBlock => VenueFault::Unavailable(detail), + RetryAction::Backoff { seconds } => VenueFault::RateLimited { + retry_after_ms: Some(seconds.saturating_mul(1000)), + }, + _ => VenueFault::Denied(detail), + } + } + } +} + +#[cfg(test)] +mod tests { + use nexum_sdk::host::RateLimit; + use videre_sdk::keeper::retry_action; + + use super::super::{HttpFailure, OrderRejection}; + use super::*; + + fn rejected(error_type: &str) -> CowApiError { + CowApiError::Rejected(OrderRejection { + status: 400, + error_type: error_type.into(), + description: "d".into(), + data: None, + }) + } + + #[test] + fn legacy_failures_project_onto_the_venue_fault_by_shape() { + assert!(matches!( + venue_fault(&rejected("InsufficientFee")), + VenueFault::Unavailable(detail) if detail.contains("InsufficientFee") + )); + assert!(matches!( + venue_fault(&rejected("TooManyLimitOrders")), + VenueFault::RateLimited { + retry_after_ms: Some(30_000) + } + )); + assert!(matches!( + venue_fault(&rejected("InvalidSignature")), + VenueFault::Denied(detail) if detail.contains("InvalidSignature") + )); + assert!(matches!( + venue_fault(&CowApiError::Fault(Fault::RateLimited(RateLimit { + retry_after_ms: Some(2_500), + }))), + VenueFault::RateLimited { + retry_after_ms: Some(2_500) + } + )); + assert!(matches!( + venue_fault(&CowApiError::Fault(Fault::Timeout)), + VenueFault::Timeout + )); + assert!(matches!( + venue_fault(&CowApiError::Http(HttpFailure { + status: 429, + body: None, + })), + VenueFault::RateLimited { + retry_after_ms: None + } + )); + assert!(matches!( + venue_fault(&CowApiError::Http(HttpFailure { + status: 502, + body: None, + })), + VenueFault::Unavailable(_) + )); + } + + #[test] + fn host_faults_stay_retryable_and_never_drop_the_watch() { + for fault in [ + Fault::Unsupported("cow-api not provisioned".into()), + Fault::Denied("allowlist".into()), + Fault::Unavailable("rpc down".into()), + Fault::Internal("host bug".into()), + Fault::InvalidInput("mangled".into()), + ] { + let projected = venue_fault(&CowApiError::Fault(fault)); + assert!(matches!(projected, VenueFault::Unavailable(_))); + assert_eq!(retry_action(&projected), RetryAction::TryNextBlock); + } + assert_eq!( + retry_action(&venue_fault(&CowApiError::Fault(Fault::Timeout))), + RetryAction::TryNextBlock + ); + assert_eq!( + retry_action(&venue_fault(&CowApiError::Fault(Fault::RateLimited( + RateLimit { + retry_after_ms: Some(2_500), + } + )))), + RetryAction::Backoff { seconds: 3 } + ); + } + + #[test] + fn server_uid_decodes_to_wire_bytes_with_a_raw_fallback() { + assert_eq!(receipt_bytes("0xc0ffee"), vec![0xC0, 0xFF, 0xEE]); + assert_eq!(receipt_bytes("not-hex"), b"not-hex".to_vec()); + } +} diff --git a/crates/shepherd-sdk/src/lib.rs b/crates/shepherd-sdk/src/lib.rs index 40798f85..2201ad35 100644 --- a/crates/shepherd-sdk/src/lib.rs +++ b/crates/shepherd-sdk/src/lib.rs @@ -13,14 +13,16 @@ //! [`OrderData`], [`OrderUid`], [`OrderKind`], [`Signature`], //! [`Chain`], [`GPv2OrderData`], [`EMPTY_APP_DATA_JSON`]). //! -//! - [`cow`] - the [`CowApiHost`] trait for `shepherd:cow/cow-api` -//! (and the [`CowHost`] bound over the core [`Host`]), -//! `GPv2OrderData` -> `OrderData` bridging ([`gpv2_to_order_data`]), -//! the classifiers mapping submit failures into the keeper -//! [`RetryAction`], and [`run`] - the poll -> outcome -> -//! gate/journal/submit composition over the keeper stores, -//! dispatching the structured [`Verdict`] from the `composable-cow` -//! keeper crate. +//! - [`cow`] - [`run`], the poll -> outcome -> gate/journal/submit +//! composition over the keeper stores, dispatching the structured +//! [`Verdict`] from the `composable-cow` keeper crate and submitting +//! through the typed [`CowClient`] on the `videre:venue/client` +//! seam; `GPv2OrderData` -> `OrderData` bridging +//! ([`gpv2_to_order_data`]) and the classifiers mapping submit +//! failures into the keeper [`RetryAction`]. The legacy +//! [`CowApiHost`] trait for `shepherd:cow/cow-api` (and the +//! [`CowHost`] bound over the core [`Host`]) stays for the read +//! paths and the transitional [`CowApiTransport`] bridge. //! //! - [`bind_cow_host_via_wit_bindgen!`](bind_cow_host_via_wit_bindgen) - //! the CoW layering of `nexum_sdk::bind_host_via_wit_bindgen!`: @@ -47,6 +49,8 @@ //! [`EMPTY_APP_DATA_JSON`]: cowprotocol::EMPTY_APP_DATA_JSON //! [`CowApiHost`]: cow::CowApiHost //! [`CowHost`]: cow::CowHost +//! [`CowClient`]: cow::CowClient +//! [`CowApiTransport`]: cow::CowApiTransport //! [`Host`]: nexum_sdk::host::Host //! [`gpv2_to_order_data`]: cow::gpv2_to_order_data //! [`Verdict`]: composable_cow::Verdict diff --git a/crates/shepherd-sdk/tests/run.rs b/crates/shepherd-sdk/tests/run.rs index bff0f238..f4e32081 100644 --- a/crates/shepherd-sdk/tests/run.rs +++ b/crates/shepherd-sdk/tests/run.rs @@ -13,13 +13,19 @@ 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, CowIntent, CowIntentBody, OrderRejection, SignedOrder, gpv2_to_order_data, - order_data_to_body, run, + CowApiError, CowApiTransport, CowClient, CowIntent, CowIntentBody, CowVenue, OrderRejection, + SignedOrder, gpv2_to_order_data, order_data_to_body, run, }; use shepherd_sdk_test::MockHost; const SEPOLIA: u64 = 11_155_111; +/// The typed client every test drives `run` with: the transitional +/// cow-api bridge over the composed mock host. +fn venue(host: &MockHost) -> CowClient> { + CowClient::with_transport(CowApiTransport::new(host, SEPOLIA)) +} + /// Closure-backed source so each test scripts its own outcome and /// observes its own poll calls. struct FnSource(F); @@ -124,6 +130,7 @@ fn try_next_block_leaves_the_store_untouched() { run( &host, + &venue(&host), &src(|_, _, _, _| Verdict::TryNextBlock { reason: [0; 4] }), &sample_tick(), ) @@ -141,6 +148,7 @@ fn try_on_block_sets_the_block_gate() { run( &host, + &venue(&host), &src(|_, _, _, _| Verdict::WaitBlock { wait_until: 2_000, reason: [0; 4], @@ -163,6 +171,7 @@ fn try_at_epoch_sets_the_epoch_gate() { run( &host, + &venue(&host), &src(|_, _, _, _| Verdict::WaitTimestamp { wait_until: 1_800_000_000, reason: [0; 4], @@ -186,6 +195,7 @@ fn invalid_removes_the_watch_and_its_gates() { run( &host, + &venue(&host), &src(|_, _, _, _| Verdict::Invalid { reason: [0; 4] }), &sample_tick(), ) @@ -207,6 +217,7 @@ fn gated_watch_is_not_polled() { run( &host, + &venue(&host), &src(|_, _, _, _| { polls.set(polls.get() + 1); Verdict::TryNextBlock { reason: [0; 4] } @@ -226,6 +237,7 @@ fn malformed_watch_rows_are_skipped() { run( &host, + &venue(&host), &src(|_, _, _, _| { polls.set(polls.get() + 1); Verdict::TryNextBlock { reason: [0; 4] } @@ -250,7 +262,7 @@ fn ready_submits_once_and_journals_the_intent_id() { let order = order.clone(); src(move |_, _, _, _| ready_outcome(&order)) }; - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); assert_eq!(host.cow_api.call_count(), 1); assert!( @@ -273,7 +285,7 @@ fn ready_marker_keys_on_the_intent_id_never_the_server_receipt() { let order = order.clone(); src(move |_, _, _, _| ready_outcome(&order)) }; - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); let snapshot = host.store.snapshot(); assert!(snapshot.contains_key(&format!("submitted:{}", intent_id(&order)))); @@ -295,6 +307,7 @@ fn ready_skips_the_orderbook_when_the_intent_id_is_journalled() { run( &host, + &venue(&host), &src(|_, _, _, _| { polls.set(polls.get() + 1); ready_outcome(&order) @@ -320,6 +333,7 @@ fn ready_with_unknown_marker_skips_submit_and_keeps_the_watch() { run( &host, + &venue(&host), &src(move |_, _, _, _| ready_outcome(&order)), &sample_tick(), ) @@ -343,7 +357,7 @@ fn ready_beyond_the_valid_to_horizon_drops_the_watch() { order.validTo = valid_to_in(2 * 365 * 24 * 3_600); let source = src(move |_, _, _, _| ready_outcome(&order)); - let (result, logs) = capture_tracing(|| run(&host, &source, &sample_tick())); + let (result, logs) = capture_tracing(|| run(&host, &venue(&host), &source, &sample_tick())); result.unwrap(); assert_eq!(host.cow_api.call_count(), 0, "the body is never shipped"); @@ -377,6 +391,7 @@ fn transient_rejection_keeps_the_watch_ungated() { run( &host, + &venue(&host), &src(move |_, _, _, _| ready_outcome(&order)), &sample_tick(), ) @@ -401,6 +416,7 @@ fn permanent_rejection_drops_the_watch_through_the_ledger() { run( &host, + &venue(&host), &src(move |_, _, _, _| ready_outcome(&order)), &sample_tick(), ) @@ -426,7 +442,7 @@ fn duplicated_order_records_the_receipt_and_keeps_the_watch() { let order = order.clone(); src(move |_, _, _, _| ready_outcome(&order)) }; - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); assert!(host.store.snapshot().contains_key(&key)); assert!( @@ -437,7 +453,7 @@ fn duplicated_order_records_the_receipt_and_keeps_the_watch() { ); // The next tick must not touch the orderbook again. - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); assert_eq!(host.cow_api.call_count(), 1); } @@ -455,7 +471,7 @@ fn restart_with_a_journalled_intent_does_not_repost() { let order = order.clone(); src(move |_, _, _, _| ready_outcome(&order)) }; - run(&host, &source, &sample_tick()).unwrap(); + run(&host, &venue(&host), &source, &sample_tick()).unwrap(); assert_eq!(host.cow_api.call_count(), 1); // A restarted keeper: fresh instance, the local store carried over. @@ -465,7 +481,7 @@ fn restart_with_a_journalled_intent_does_not_repost() { } restarted.cow_api.respond(Ok("0xserveruid".to_string())); - run(&restarted, &source, &sample_tick()).unwrap(); + run(&restarted, &venue(&restarted), &source, &sample_tick()).unwrap(); assert_eq!( host.cow_api.call_count() + restarted.cow_api.call_count(), @@ -493,7 +509,13 @@ fn rate_limited_submit_backs_off_through_the_epoch_gate() { })))); let tick = sample_tick(); - run(&host, &src(move |_, _, _, _| ready_outcome(&order)), &tick).unwrap(); + run( + &host, + &venue(&host), + &src(move |_, _, _, _| ready_outcome(&order)), + &tick, + ) + .unwrap(); let snapshot = host.store.snapshot(); assert!(snapshot.contains_key(&key), "backoff must keep the watch"); @@ -504,3 +526,93 @@ fn rate_limited_submit_backs_off_through_the_epoch_gate() { ); assert!(!snapshot.keys().any(|k| k.starts_with("submitted:"))); } + +// ---- the generic seam ---- + +/// The seam proof: a `Post` verdict reaches the venue transport as the +/// encoded `CowIntentBody` under the CoW venue id, the journal keys on +/// the generic submission key, and the legacy cow-api seam is never +/// touched. +#[test] +fn ready_submits_the_encoded_intent_body_through_the_venue_seam() { + use std::cell::RefCell; + + use videre_sdk::keeper::submission_key; + use videre_sdk::{ + IntentBody as _, IntentStatus, Quotation, SubmitOutcome, Venue as _, VenueFault, VenueId, + VenueTransport, + }; + + /// Records the venue and wire bytes of every submit. + struct SpyTransport { + calls: RefCell)>>, + } + + impl videre_sdk::client::sealed::SealedTransport for &SpyTransport {} + + impl VenueTransport for &SpyTransport { + async fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { + unreachable!("quote not exercised") + } + + async fn submit( + &self, + venue: &VenueId, + body: Vec, + ) -> Result { + self.calls.borrow_mut().push((venue.to_string(), body)); + Ok(SubmitOutcome::Accepted(vec![0xAA])) + } + + async fn status( + &self, + _venue: &VenueId, + _receipt: &[u8], + ) -> Result { + unreachable!("status not exercised") + } + + async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { + unreachable!("cancel not exercised") + } + } + + let host = MockHost::new(); + seed_watch(&host); + let order = submittable_order(); + let spy = SpyTransport { + calls: RefCell::new(Vec::new()), + }; + let client = CowClient::with_transport(&spy); + + let source = { + let order = order.clone(); + src(move |_, _, _, _| ready_outcome(&order)) + }; + run(&host, &client, &source, &sample_tick()).unwrap(); + + let order_data = gpv2_to_order_data(&order).expect("known markers"); + let expected = CowIntentBody::V1(CowIntent::Signed(SignedOrder { + order: order_data_to_body(&order_data), + owner: sample_owner().into_array(), + signature: hex!("c0ffeec0ffeec0ffee").to_vec(), + })) + .to_bytes() + .expect("body encodes"); + + let calls = spy.calls.borrow(); + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].0, CowVenue::ID.as_str()); + assert_eq!(calls[0].1, expected, "the wire carries the intent body"); + assert!( + Journal::submitted(&host) + .contains(&submission_key(&CowVenue::ID, &expected)) + .unwrap(), + "the journal keys on the generic submission key", + ); + assert_eq!( + host.cow_api.call_count(), + 0, + "the legacy cow-api seam is never touched", + ); +} diff --git a/crates/videre-sdk/src/keeper.rs b/crates/videre-sdk/src/keeper.rs index 7255efd9..8ca91a3e 100644 --- a/crates/videre-sdk/src/keeper.rs +++ b/crates/videre-sdk/src/keeper.rs @@ -112,7 +112,7 @@ impl Keeper { report.unsigned.push(tx); continue; } - Err(fault) => retry_action(fault), + Err(fault) => retry_action(&fault), } } Sweep::WaitBlock => RetryAction::TryNextBlock, @@ -165,8 +165,10 @@ pub fn submission_key(venue: &VenueId, body: &[u8]) -> String { /// Fold a venue refusal into the retry action the ledger runs: the /// throttle hint becomes an epoch gate, transient failures retry next -/// block, and refusals no retry can cure drop the watch. -fn retry_action(fault: VenueFault) -> RetryAction { +/// block, and refusals no retry can cure drop the watch. Public so a +/// keeper sweeping outside [`Keeper::sweep`] folds refusals the same +/// way. +pub fn retry_action(fault: &VenueFault) -> RetryAction { match fault { VenueFault::RateLimited { retry_after_ms: Some(ms), diff --git a/crates/videre-sdk/src/lib.rs b/crates/videre-sdk/src/lib.rs index 99e16d9a..007eb2e2 100644 --- a/crates/videre-sdk/src/lib.rs +++ b/crates/videre-sdk/src/lib.rs @@ -87,7 +87,7 @@ pub use adapter::VenueAdapter; pub use body::{BodyError, IntentBody}; pub use client::{ClientError, HostVenues, Quoted, Venue, VenueClient, VenueId, VenueTransport}; pub use faults::VenueFault; -pub use keeper::{Keeper, Sweep, SweepReport}; +pub use keeper::{Keeper, Sweep, SweepReport, retry_action}; /// Derive [`IntentBody`] on the outer per-venue version enum. See /// [`videre_macros::IntentBody`]. pub use videre_macros::IntentBody; diff --git a/modules/twap-monitor/src/strategy.rs b/modules/twap-monitor/src/strategy.rs index 540a6c7e..d5426d58 100644 --- a/modules/twap-monitor/src/strategy.rs +++ b/modules/twap-monitor/src/strategy.rs @@ -24,7 +24,7 @@ use nexum_sdk::chain::{eth_call_params, parse_eth_call_result}; use nexum_sdk::events::Log; use nexum_sdk::host::{ChainError, Fault}; use nexum_sdk::keeper::{ConditionalSource, Tick, WatchRef, WatchSet}; -use shepherd_sdk::cow::{CowHost, events, run}; +use shepherd_sdk::cow::{CowApiTransport, CowClient, CowHost, events, run}; /// Block fields the poll path reads on every dispatch. pub struct BlockInfo { @@ -71,15 +71,17 @@ pub fn on_chain_logs(host: &H, logs: &[Log]) -> Result<(), Fault> { } /// Poll entry: run the keeper over every gate-ready watch through the -/// shared composition. The block timestamp arrives in milliseconds; the -/// tick carries Unix seconds. +/// shared composition, submitting through the typed client on the +/// transitional cow-api bridge. The block timestamp arrives in +/// milliseconds; the tick carries Unix seconds. pub fn on_block(host: &H, block: BlockInfo) -> Result<(), Fault> { let tick = Tick { chain_id: block.chain_id, block: block.number, epoch_s: block.timestamp / 1000, }; - run(host, &TwapSource, &tick) + let venue = CowClient::with_transport(CowApiTransport::new(host, tick.chain_id)); + run(host, &venue, &TwapSource, &tick) } // ---- indexing path ----