From e46ae5c7d6b0f5da38fc69f49e259aa14daa9548 Mon Sep 17 00:00:00 2001 From: mfw78 Date: Fri, 17 Jul 2026 20:38:05 +0000 Subject: [PATCH] modules: re-point ethflow-watcher onto the pool observe path The videre client face gains observe: put an externally-obtained receipt (an on-chain placement the pool never submitted) under the registry's status watch. The registry resolves the venue and admits the watch without touching the adapter; refusal at the watch cap is a typed unavailable. HostVenues and the typed VenueClient carry the new verb; the transport trait defaults it to unsupported so existing transports opt in. ethflow-watcher flips onto that seam: the module is now a the legacy cow-api extension. on_chain_logs computes each placement's orderbook UID and observes it at the cow venue; the registry polls the adapter and fans intent-status transitions back, and the module journals observed:{uid} on the first one. Observe-only is preserved: no call path reaches quote, submit, status, or cancel. The shepherd-backtest rlib replay drives the same strategy pair over a recording pool transport and delivers the open transition the registry would poll, keeping the Observed classification exact-UID strict. The ethflow boot e2e moves from the cow-api extension suite to the venue platform suite. --- Cargo.lock | 9 +- crates/shepherd-backtest/Cargo.toml | 4 +- crates/shepherd-backtest/src/main.rs | 4 +- crates/shepherd-backtest/src/replay.rs | 242 +++++--- crates/shepherd-backtest/src/report.rs | 11 +- crates/shepherd-cow-host/Cargo.toml | 1 - crates/shepherd-cow-host/tests/cow_boot.rs | 43 +- crates/videre-host/src/bindings.rs | 7 +- crates/videre-host/src/client.rs | 4 + crates/videre-host/src/registry.rs | 110 +++- crates/videre-host/tests/platform.rs | 38 ++ crates/videre-sdk/src/client.rs | 23 + modules/ethflow-watcher/Cargo.toml | 4 +- modules/ethflow-watcher/module.toml | 25 +- modules/ethflow-watcher/src/lib.rs | 97 ++-- modules/ethflow-watcher/src/strategy.rs | 610 +++++++++++---------- wit/videre-venue/venue.wit | 4 + 17 files changed, 766 insertions(+), 470 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index dddca7c6..2e31d334 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2300,12 +2300,12 @@ version = "0.1.0" dependencies = [ "alloy-primitives", "alloy-sol-types", + "cow-venue", "cowprotocol", "nexum-sdk", "nexum-sdk-test", - "shepherd-sdk", - "shepherd-sdk-test", "tracing", + "videre-sdk", "wit-bindgen 0.59.0", ] @@ -5198,12 +5198,14 @@ version = "0.1.0" dependencies = [ "anyhow", "clap", + "cow-venue", "ethflow-watcher", "hex", "nexum-sdk", + "nexum-sdk-test", "serde", "serde_json", - "shepherd-sdk-test", + "videre-sdk", ] [[package]] @@ -5212,7 +5214,6 @@ version = "0.2.0" dependencies = [ "alloy-chains", "alloy-primitives", - "alloy-rpc-types-eth", "anyhow", "cowprotocol", "http", diff --git a/crates/shepherd-backtest/Cargo.toml b/crates/shepherd-backtest/Cargo.toml index 2ec14aa6..e424d50c 100644 --- a/crates/shepherd-backtest/Cargo.toml +++ b/crates/shepherd-backtest/Cargo.toml @@ -15,8 +15,10 @@ path = "src/main.rs" # (alongside its wasm cdylib) specifically so this crate can drive # `strategy::on_chain_logs` directly without an embedded runtime. ethflow-watcher = { path = "../../modules/ethflow-watcher" } +cow-venue = { path = "../cow-venue", features = ["client"] } nexum-sdk = { path = "../nexum-sdk" } -shepherd-sdk-test = { path = "../shepherd-sdk-test" } +nexum-sdk-test = { path = "../nexum-sdk-test" } +videre-sdk = { path = "../videre-sdk" } anyhow.workspace = true clap.workspace = true diff --git a/crates/shepherd-backtest/src/main.rs b/crates/shepherd-backtest/src/main.rs index 45d446b6..6d0ff80d 100644 --- a/crates/shepherd-backtest/src/main.rs +++ b/crates/shepherd-backtest/src/main.rs @@ -3,8 +3,8 @@ //! Offline replay harness for Shepherd modules. Loads a fixtures //! JSON produced by `tools/backtest-collect/backtest_collect.py`, //! drives each on-chain event through the production strategy code -//! via `shepherd_sdk_test::MockHost`, classifies the result, and -//! emits a Markdown report at +//! over `nexum_sdk_test::MockHost` and a recording pool transport, +//! classifies the result, and emits a Markdown report at //! `docs/operations/backtest-reports/backtest-7d-YYYY-MM-DD.md`. //! //! ## Scope diff --git a/crates/shepherd-backtest/src/replay.rs b/crates/shepherd-backtest/src/replay.rs index 71a0aef1..36649f7b 100644 --- a/crates/shepherd-backtest/src/replay.rs +++ b/crates/shepherd-backtest/src/replay.rs @@ -1,32 +1,37 @@ -//! Per-event replay against `ethflow_watcher::strategy::on_chain_logs`. +//! Per-event replay against the ethflow-watcher strategy pair. //! -//! Each [`EthFlowFixture`] is driven through the production strategy -//! exactly the way the live engine does it: a fresh [`MockHost`] is -//! constructed, a catch-all 200 response is programmed for any -//! `cow_api_request` call (the observe+verify strategy GETs -//! `/api/v1/orders/{uid}` to confirm the orderbook has indexed the -//! order), and `strategy::on_chain_logs` is invoked with an alloy -//! `Log` reconstructed from the raw `eth_getLogs` payload. +//! Each [`EthFlowFixture`] is driven the way the live engine does it, +//! minus the wasm boundary: `strategy::on_chain_logs` runs over a +//! recording venue transport (the pool seam), then the status +//! transition the registry would poll for the watched receipt is +//! delivered through `strategy::on_intent_status`. In backtest context +//! all fixtures are confirmed real orders, so the simulated transition +//! is `open`. //! -//! The classification falls into one of the four buckets defined in -//! the issue: +//! The classification falls into one of four buckets: //! -//! - `Observed`: the strategy verified the order with exactly one -//! `GET /api/v1/orders/{uid}` and wrote `observed:{uid}` to the -//! local store. This is the success case under the observe+verify -//! strategy. -//! - `RejectedExpected`: the strategy returned without observing in a -//! documented case (reserved for fixtures where the mock returns 404 -//! — not applicable when all fixtures program 200). +//! - `Observed`: the strategy registered exactly one `cow` watch whose +//! receipt is the fixture's UID and wrote `observed:{uid}` on the +//! delivered transition. The success case. +//! - `RejectedExpected`: reserved for a documented skip path; not +//! emitted in the current batch. //! - `RejectedUnexpected`: the strategy returned Ok but the observe -//! contract was violated (no `observed:{uid}` marker, an unexpected -//! orderbook call shape, or a `submit_order` attempt); a follow-up -//! should be filed before the report closes. -//! - `StrategyError`: `on_chain_logs` returned `Err(fault)`. A test +//! contract was violated (a wrong watch shape, a non-observe venue +//! call, or a missing `observed:{uid}` marker); a follow-up should +//! be filed before the report closes. +//! - `StrategyError`: a strategy call returned `Err(fault)`. A test //! bug or an `unreachable!` we want to investigate. +use std::cell::RefCell; +use std::rc::Rc; + +use cow_venue::client::CowClient; use ethflow_watcher::strategy; -use shepherd_sdk_test::MockHost; +use nexum_sdk::status_body::{IntentStatus, StatusBody}; +use nexum_sdk_test::{CapturedEvents, MockHost, capture_tracing}; +use videre_sdk::client::{VenueId, VenueTransport}; +use videre_sdk::rt::complete; +use videre_sdk::{Quotation, SubmitOutcome, VenueFault}; use crate::fixtures::{EthFlowFixture, parse_address}; @@ -45,9 +50,8 @@ pub struct ReplayOutcome { #[derive(Debug, Clone, PartialEq, Eq)] pub enum Classification { Observed, - /// Reserved for any documented skip path (e.g. a fixture where the mock - /// returns 404 for an un-indexed order). Not emitted in the current batch; - /// retained so the acceptance-ratio formula is complete. + /// Reserved for any documented skip path. Not emitted in the current + /// batch; retained so the acceptance-ratio formula is complete. #[allow(dead_code)] RejectedExpected(String), RejectedUnexpected(String), @@ -74,16 +78,94 @@ impl Classification { } } -/// Replay one EthFlow fixture through the production strategy. +/// One recorded transport call: the verb, and for `observe` the routed +/// venue and receipt. +#[derive(Clone, Debug, PartialEq, Eq)] +enum Call { + Quote, + Submit, + Observe(String, Vec), + Status, + Cancel, +} + +impl Call { + fn label(&self) -> &'static str { + match self { + Call::Quote => "quote", + Call::Submit => "submit", + Call::Observe(..) => "observe", + Call::Status => "status", + Call::Cancel => "cancel", + } + } +} + +/// Records every pool call; `observe` accepts, the other verbs refuse. +/// Cloneable over shared state so the replay keeps a handle after one +/// moves into the client. +#[derive(Clone, Default)] +struct RecordingVenues { + calls: Rc>>, +} + +impl RecordingVenues { + fn calls(&self) -> Vec { + self.calls.borrow().clone() + } +} + +impl videre_sdk::client::sealed::SealedTransport for RecordingVenues {} + +impl VenueTransport for RecordingVenues { + async fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { + self.calls.borrow_mut().push(Call::Quote); + Err(VenueFault::Unsupported) + } + + async fn submit(&self, _venue: &VenueId, _body: Vec) -> Result { + self.calls.borrow_mut().push(Call::Submit); + Err(VenueFault::Unsupported) + } + + async fn observe(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault> { + self.calls + .borrow_mut() + .push(Call::Observe(venue.to_string(), receipt.to_vec())); + Ok(()) + } + + async fn status( + &self, + _venue: &VenueId, + _receipt: &[u8], + ) -> Result { + self.calls.borrow_mut().push(Call::Status); + Err(VenueFault::Unsupported) + } + + async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { + self.calls.borrow_mut().push(Call::Cancel); + Err(VenueFault::Unsupported) + } +} + +/// The `open` transition the registry would report for an indexed order. +fn open_status() -> Result, String> { + StatusBody { + status: IntentStatus::Open, + proof: None, + reason: None, + } + .encode() + .map_err(|e| format!("status body encode: {e}")) +} + +/// Replay one EthFlow fixture through the production strategy pair. pub fn replay_ethflow(fx: &EthFlowFixture, chain_id: u64) -> ReplayOutcome { let host = MockHost::new(); - - // Program a catch-all 200 response for any cow_api_request. In - // the observe+verify strategy the module GETs - // `/api/v1/orders/{uid}` to confirm the orderbook has indexed the - // order. In backtest context all fixtures are confirmed real orders, - // so the mock orderbook always returns 200 (indexed). - host.cow_api.respond_to_request(Ok("{}".to_string())); + let venues = RecordingVenues::default(); + let client = CowClient::with_transport(venues.clone()); // Reconstruct the log fields. Topics + data come straight from the // collector's `raw_log`; the contract address is the EthFlow @@ -120,18 +202,32 @@ pub fn replay_ethflow(fx: &EthFlowFixture, chain_id: u64) -> ReplayOutcome { } .into(); - // Drive the strategy. - let result = strategy::on_chain_logs(&host, chain_id, &[log]); - let log_lines: Vec = host - .logging - .lines() - .into_iter() - .map(|l| format!("[{:?}] {}", l.level, l.message)) - .collect(); + // Drive the strategy: register the watch, then deliver the status + // transition the registry would poll for the watched receipt. + let (result, logs) = capture_tracing(|| { + complete(strategy::on_chain_logs(&host, &client, chain_id, &[log])) + .ok_or_else(|| "strategy future suspended".to_owned()) + .and_then(|r| r.map_err(|e| e.to_string()))?; + let calls = venues.calls(); + let [Call::Observe(venue, receipt)] = calls.as_slice() else { + let shapes: Vec<&str> = calls.iter().map(Call::label).collect(); + return Err(format!( + "expected exactly one observe; saw [{}]", + shapes.join(", ") + )); + }; + if venue != "cow" { + return Err(format!("watch routed to venue `{venue}`, not `cow`")); + } + strategy::on_intent_status(&host, venue, receipt, &open_status()?) + .map_err(|e| e.to_string())?; + Ok(receipt.clone()) + }); + let log_lines = log_lines(&logs); let class = match result { - Err(e) => Classification::StrategyError(e.to_string()), - Ok(()) => classify_ok(&host, &fx.uid, &log_lines), + Err(detail) => classify_err(&venues, detail), + Ok(receipt) => classify_ok(&host, &receipt, &fx.uid, &log_lines), }; ReplayOutcome { @@ -143,6 +239,13 @@ pub fn replay_ethflow(fx: &EthFlowFixture, chain_id: u64) -> ReplayOutcome { } } +fn log_lines(logs: &CapturedEvents) -> Vec { + logs.events() + .into_iter() + .map(|e| format!("[{:?}] {}", e.level, e.message)) + .collect() +} + fn error_outcome(fx: &EthFlowFixture, reason: String) -> ReplayOutcome { ReplayOutcome { uid: fx.uid.clone(), @@ -153,31 +256,40 @@ fn error_outcome(fx: &EthFlowFixture, reason: String) -> ReplayOutcome { } } -/// Classify an `Ok(())` replay by inspecting the mock's recorded -/// side effects, independent of the strategy's own logging. +/// A failed replay is an observe-contract violation when the strategy +/// itself returned Ok but touched the pool wrongly; otherwise a +/// strategy error. +fn classify_err(venues: &RecordingVenues, detail: String) -> Classification { + if venues + .calls() + .iter() + .any(|c| !matches!(c, Call::Observe(..))) + { + return Classification::RejectedUnexpected(format!( + "strategy called a non-observe venue verb; observe-only must never submit ({detail})" + )); + } + if detail.starts_with("expected exactly one observe") + || detail.starts_with("watch routed to venue") + { + return Classification::RejectedUnexpected(detail); + } + Classification::StrategyError(detail) +} + +/// Classify an `Ok` replay by inspecting the recorded side effects, +/// independent of the strategy's own logging. /// /// `Observed` demands the full observe contract, not just the marker: -/// exactly one orderbook call, shaped `GET /api/v1/orders/{uid}`, -/// zero `submit_order` attempts, and the `observed:{uid}` store key. -/// The exact-UID match (never an `observed:` prefix scan) means a -/// compute_uid divergence between module and collector cannot produce -/// a false `Observed`. -fn classify_ok(host: &MockHost, uid: &str, log_lines: &[String]) -> Classification { - if host.cow_api.call_count() > 0 { - return Classification::RejectedUnexpected( - "strategy called submit_order; observe+verify must never submit".into(), - ); - } - let requests = host.cow_api.request_calls(); - let expected_path = format!("/api/v1/orders/{uid}"); - if requests.len() != 1 || requests[0].method != "GET" || requests[0].path != expected_path { - let shapes: Vec = requests - .iter() - .map(|r| format!("{} {}", r.method, r.path)) - .collect(); +/// the one watch's receipt renders to exactly the fixture UID (never a +/// prefix scan, so a compute_uid divergence between module and +/// collector cannot produce a false `Observed`), and the delivered +/// transition wrote the `observed:{uid}` store key. +fn classify_ok(host: &MockHost, receipt: &[u8], uid: &str, log_lines: &[String]) -> Classification { + let receipt_hex = format!("0x{}", hex::encode(receipt)); + if receipt_hex != uid { return Classification::RejectedUnexpected(format!( - "expected exactly one GET {expected_path}; saw [{}]", - shapes.join(", ") + "watched receipt {receipt_hex} does not match the collector UID" )); } if host diff --git a/crates/shepherd-backtest/src/report.rs b/crates/shepherd-backtest/src/report.rs index 9e64b9fd..a27b28ae 100644 --- a/crates/shepherd-backtest/src/report.rs +++ b/crates/shepherd-backtest/src/report.rs @@ -37,11 +37,12 @@ pub fn render(fx: &Fixtures, outcomes: &[ReplayOutcome], threshold: f64) -> Stri )); out.push_str( "Replays every collected EthFlow `OrderPlacement` event through the production \ - `ethflow_watcher::strategy::on_chain_logs` code path via `shepherd_sdk_test::MockHost`. \ - The orderbook is **never hit**: the MockHost programs a catch-all 200 for all \ - `cow_api_request` calls so the observe+verify strategy sees every fixture as \ - already indexed. Success is measured by whether the strategy wrote the exact \ - `observed:{uid}` marker to the local store after the 200 confirmation.\n\n", + `ethflow_watcher::strategy` pair (`on_chain_logs` then `on_intent_status`) over \ + `nexum_sdk_test::MockHost` and a recording pool transport. The orderbook is \ + **never hit**: the transport accepts every watch and the replay delivers the \ + `open` transition the registry would poll for it, so every fixture reads as \ + indexed. Success is measured by whether the strategy watched exactly the \ + fixture UID and wrote the exact `observed:{uid}` marker on the transition.\n\n", ); out.push_str("## Run metadata\n\n"); out.push_str("| Field | Value |\n|---|---|\n"); diff --git a/crates/shepherd-cow-host/Cargo.toml b/crates/shepherd-cow-host/Cargo.toml index a5113911..47ba5e1e 100644 --- a/crates/shepherd-cow-host/Cargo.toml +++ b/crates/shepherd-cow-host/Cargo.toml @@ -50,4 +50,3 @@ toml.workspace = true tokio.workspace = true wiremock.workspace = true tempfile.workspace = true -alloy-rpc-types-eth.workspace = true diff --git a/crates/shepherd-cow-host/tests/cow_boot.rs b/crates/shepherd-cow-host/tests/cow_boot.rs index 3953ab6a..b195b537 100644 --- a/crates/shepherd-cow-host/tests/cow_boot.rs +++ b/crates/shepherd-cow-host/tests/cow_boot.rs @@ -8,7 +8,6 @@ use std::path::{Path, PathBuf}; use std::sync::Arc; -use alloy_chains::Chain; use nexum_runtime::bindings::nexum; use nexum_runtime::engine_config::{EngineConfig, ModuleLimits}; use nexum_runtime::host::component::{Components, RuntimeTypes}; @@ -144,38 +143,6 @@ async fn boot_production_module( .expect("boot_single") } -/// ethflow-watcher imports `shepherd:cow/cow-api` and subscribes to logs; -/// it boots with the cow extension and a synthetic log is delivered. -#[tokio::test] -async fn e2e_ethflow_watcher_log_dispatch() { - let Some(wasm) = module_wasm_or_skip("ethflow-watcher") else { - return; - }; - let manifest = production_module_toml("modules/ethflow-watcher/module.toml"); - let engine = make_wasmtime_engine(); - let linker = make_linker(&engine); - let (_dir, store) = temp_local_store(); - - let mut supervisor = boot_production_module(&engine, &linker, &store, &wasm, &manifest).await; - assert_eq!(supervisor.alive_count(), 1); - - // A log with an unrecognised topic is silently skipped by the module's - // decoder (returns `None` from `decode_order_placement`), so the test - // only proves: supervisor delivered, module did not trap, module stayed - // alive. - let synthetic_log = alloy_rpc_types_eth::Log::default(); - let dispatched = supervisor - .dispatch_chain_log( - "ethflow-watcher", - Chain::from_id(SEPOLIA), - synthetic_log, - None, - ) - .await; - assert!(dispatched); - assert_eq!(supervisor.alive_count(), 1); -} - /// stop-loss imports `shepherd:cow/cow-api`; it boots with the cow /// extension and a block dispatch reaches it. #[tokio::test] @@ -195,17 +162,17 @@ async fn e2e_stop_loss_block_dispatch() { } /// The boot-order invariant, exercised (not merely asserted in prose): -/// a module that imports `shepherd:cow/cow-api` (ethflow-watcher) must -/// NOT boot when the cow extension is absent from the linker AND the +/// a module that imports `shepherd:cow/cow-api` (stop-loss) must NOT +/// boot when the cow extension is absent from the linker AND the /// capability registry. The paired linker-hook + capability-namespace /// registration is what makes the same module boot in the tests above; /// drop the pairing and boot fails. #[tokio::test] -async fn ethflow_watcher_without_cow_extension_fails_to_boot() { - let Some(wasm) = module_wasm_or_skip("ethflow-watcher") else { +async fn stop_loss_without_cow_extension_fails_to_boot() { + let Some(wasm) = module_wasm_or_skip("stop-loss") else { return; }; - let manifest = production_module_toml("modules/ethflow-watcher/module.toml"); + let manifest = production_module_toml("modules/examples/stop-loss/module.toml"); let engine = make_wasmtime_engine(); // Core-only: no cow linker hook, no cow capability namespace. let linker = build_linker::(&engine, &[]).expect("build_linker"); diff --git a/crates/videre-host/src/bindings.rs b/crates/videre-host/src/bindings.rs index c07618a9..06e2b887 100644 --- a/crates/videre-host/src/bindings.rs +++ b/crates/videre-host/src/bindings.rs @@ -157,7 +157,7 @@ mod value_flow_smoke { /// client interface and, transitively, the types interface and its /// value-flow dependency. The test names every generated type, case, and /// field by its plain Rust spelling, and a dummy `client` host impl pins -/// the four function signatures, so a keyword collision or an accidental +/// the five function signatures, so a keyword collision or an accidental /// signature change fails this build rather than a downstream binding. #[cfg(test)] mod client_smoke { @@ -193,6 +193,10 @@ mod client_smoke { Err(VenueError::UnknownVenue) } + fn observe(&mut self, _venue: String, _receipt: Vec) -> Result<(), VenueError> { + Err(VenueError::UnknownVenue) + } + fn status( &mut self, _venue: String, @@ -264,6 +268,7 @@ mod client_smoke { let mut client = DummyClient; assert!(client.quote(String::new(), Vec::new()).is_err()); assert!(client.submit(String::new(), Vec::new()).is_err()); + assert!(client.observe(String::new(), Vec::new()).is_err()); assert!(client.status(String::new(), Vec::new()).is_err()); assert!(client.cancel(String::new(), Vec::new()).is_err()); } diff --git a/crates/videre-host/src/client.rs b/crates/videre-host/src/client.rs index 2973b01e..cbe27e70 100644 --- a/crates/videre-host/src/client.rs +++ b/crates/videre-host/src/client.rs @@ -38,6 +38,10 @@ impl Host for HostState { .await } + async fn observe(&mut self, venue: String, receipt: Vec) -> Result<(), VenueError> { + registry(self)?.observe(&VenueId::from(venue), receipt) + } + async fn status( &mut self, venue: String, diff --git a/crates/videre-host/src/registry.rs b/crates/videre-host/src/registry.rs index 6a47fdf7..815205e6 100644 --- a/crates/videre-host/src/registry.rs +++ b/crates/videre-host/src/registry.rs @@ -7,7 +7,9 @@ //! header, run the guard interposition seam on it (advisory-only for now: //! see [`EgressGuard`]), and only then submit. //! Status and cancel are pass-throughs; they are not submissions, so they -//! skip the header, the guard, and the quota. +//! skip the header, the guard, and the quota. Observe puts an +//! externally-obtained receipt under status watch without touching the +//! adapter at all. //! //! Invocation is serialised per adapter through the supervised-actor //! primitive: each adapter sits behind its own [`ActorSlot`], so concurrent @@ -318,8 +320,8 @@ struct VenueRegistryInner { ledger: Mutex, watch_limit: WatchLimit, /// Receipts under status watch, appended by accepted submissions and - /// pruned as they reach a terminal status, expire, or overflow - /// [`WatchLimit`]. + /// explicit observes, pruned as they reach a terminal status, expire, + /// or overflow [`WatchLimit`]. watched: Mutex>, } @@ -481,11 +483,26 @@ impl VenueRegistry { adapter.quote(&body).await } - /// Put a `(venue, receipt)` pair under status watch. Idempotent: a - /// re-submitted receipt keeps its existing watch entry. Bounded: - /// expired entries evict first, and at the cap the new watch is - /// refused and logged rather than an existing live watch dropped. - fn watch(&self, venue: &VenueId, receipt: Vec) { + /// Put an externally-obtained `(venue, receipt)` pair under status + /// watch: the `observe` half of the client face, for receipts the + /// pool never submitted (an accepted submit is watched implicitly). + /// Not a submission: no header, no guard, no quota; the watch cap + /// bounds it. Idempotent. + pub fn observe(&self, venue: &VenueId, receipt: Vec) -> Result<(), VenueError> { + let _ = self.resolve(venue)?; + if self.watch(venue, receipt) { + Ok(()) + } else { + Err(VenueError::Unavailable("status watch set full".to_owned())) + } + } + + /// Put a `(venue, receipt)` pair under status watch, reporting + /// whether it is watched. Idempotent: a re-submitted receipt keeps + /// its existing watch entry. Bounded: expired entries evict first, + /// and at the cap the new watch is refused and logged rather than + /// an existing live watch dropped. + fn watch(&self, venue: &VenueId, receipt: Vec) -> bool { let (evicted, admitted) = { let mut watched = self.inner.watched.lock().expect("watch list poisoned"); let evicted = prune_expired(&mut watched); @@ -515,6 +532,7 @@ impl VenueRegistry { "status watch set full - transitions for this receipt will not be reported", ); } + admitted } /// Number of receipts currently under status watch. @@ -1444,6 +1462,82 @@ mod tests { assert_eq!(registry.watched_count(), 1); } + #[tokio::test] + async fn observe_watches_an_externally_obtained_receipt() { + let calls = Arc::new(StubCalls::default()); + let adapter = + StubAdapter::new(calls.clone()).with_status_script([Ok(IntentStatus::Fulfilled)]); + let registry = registry_with(SubmitQuota::default(), None, adapter); + + registry + .observe(&cow(), b"onchain".to_vec()) + .expect("observe succeeds"); + // Re-observing keeps the existing entry. + registry + .observe(&cow(), b"onchain".to_vec()) + .expect("observe is idempotent"); + assert_eq!(registry.watched_count(), 1); + // No adapter work happened at observe time. + assert_eq!(calls.status.load(Ordering::SeqCst), 0); + assert_eq!(calls.submit.load(Ordering::SeqCst), 0); + + // The watch polls like a submitted one: the terminal status + // reports once and prunes the entry. + let updates = registry.poll_status_transitions().await; + assert_eq!(updates.len(), 1); + assert_eq!(updates[0].receipt, b"onchain"); + assert_eq!(decoded(&updates[0]), plain(Lifecycle::Fulfilled)); + assert_eq!(registry.watched_count(), 0); + } + + #[test] + fn observe_rejects_an_unknown_venue() { + let registry = registry_with( + SubmitQuota::default(), + None, + StubAdapter::new(Arc::new(StubCalls::default())), + ); + assert!(matches!( + registry.observe(&VenueId::from("unlisted"), b"r".to_vec()), + Err(VenueError::UnknownVenue) + )); + assert_eq!(registry.watched_count(), 0); + } + + #[test] + fn observe_of_a_dead_venue_is_unavailable() { + let liveness = Liveness::default(); + let registry = VenueRegistryBuilder::new(SubmitQuota::default()).build(); + registry + .install( + cow(), + liveness.clone(), + StubAdapter::new(Arc::new(StubCalls::default())), + ) + .expect("install adapter"); + liveness.mark_dead(); + assert!(matches!( + registry.observe(&cow(), b"r".to_vec()), + Err(VenueError::Unavailable(_)) + )); + assert_eq!(registry.watched_count(), 0); + } + + #[test] + fn observe_at_the_watch_cap_is_refused_typedly() { + let limit = WatchLimit::new(1, Duration::from_secs(3600)); + let registry = + watch_bounded_registry(limit, StubAdapter::new(Arc::new(StubCalls::default()))); + + registry.observe(&cow(), b"a".to_vec()).expect("admitted"); + let err = registry + .observe(&cow(), b"b".to_vec()) + .expect_err("overflow refused"); + assert!(matches!(err, VenueError::Unavailable(_))); + // The live watch is kept; the overflow was refused. + assert_eq!(registry.watched_count(), 1); + } + #[tokio::test] async fn requires_signing_outcome_is_not_watched() { let calls = Arc::new(StubCalls::default()); diff --git a/crates/videre-host/tests/platform.rs b/crates/videre-host/tests/platform.rs index ace107a8..eadd9165 100644 --- a/crates/videre-host/tests/platform.rs +++ b/crates/videre-host/tests/platform.rs @@ -388,6 +388,44 @@ async fn e2e_intent_status_subscription_receives_polled_transitions() { ); } +/// ethflow-watcher (built by `#[videre_sdk::keeper]`) boots on the venue +/// platform with its shipped manifest and handles a delivered cow status +/// transition without trapping. +#[tokio::test] +async fn e2e_ethflow_watcher_boots_and_handles_intent_status() { + let Some(wasm) = module_wasm_or_skip("ethflow-watcher") else { + return; + }; + let manifest = workspace_path("modules/ethflow-watcher/module.toml"); + let videre = Arc::new(platform(&EngineConfig::default())); + let mut supervisor = boot_example(&videre, &wasm, &manifest).await; + assert_eq!(supervisor.alive_count(), 1); + assert!( + supervisor + .extension_subscription_kinds() + .contains(INTENT_STATUS) + ); + + let update = videre_host::IntentStatusUpdate { + venue: "cow".to_owned(), + receipt: vec![0xAB; 56], + status: nexum_status_body::StatusBody { + status: nexum_status_body::IntentStatus::Open, + proof: None, + reason: None, + } + .encode() + .expect("encode"), + }; + assert_eq!( + supervisor + .dispatch_extension_event(status_event(update)) + .await, + 1 + ); + assert_eq!(supervisor.alive_count(), 1); +} + /// The event-loop wiring, through the real seam: the platform's `events` /// source opens against the booted service map, its poll task drives the /// supervisor, and the module's handler observably ran (its log line is diff --git a/crates/videre-sdk/src/client.rs b/crates/videre-sdk/src/client.rs index c3e9b1af..63ff375d 100644 --- a/crates/videre-sdk/src/client.rs +++ b/crates/videre-sdk/src/client.rs @@ -103,6 +103,19 @@ pub trait VenueTransport: sealed::SealedTransport { body: Vec, ) -> impl Future>; + /// Put an externally-obtained receipt under the host's status + /// watch; an accepted submit is watched implicitly. Defaults to + /// `unsupported`: a transport that can watch foreign receipts opts + /// in. + fn observe( + &self, + venue: &VenueId, + receipt: &[u8], + ) -> impl Future> { + let _ = (venue, receipt); + async { Err(VenueFault::Unsupported) } + } + /// Report where a previously submitted intent is in its life. fn status( &self, @@ -137,6 +150,10 @@ impl VenueTransport for HostVenues { shims::submit(venue.as_str(), &body).map_err(VenueFault::from) } + async fn observe(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault> { + shims::observe(venue.as_str(), receipt).map_err(VenueFault::from) + } + async fn status(&self, venue: &VenueId, receipt: &[u8]) -> Result { shims::status(venue.as_str(), receipt).map_err(VenueFault::from) } @@ -208,6 +225,12 @@ impl VenueClient { Ok(self.transport.submit(&V::ID, bytes).await?) } + /// Put an externally-obtained receipt under the host's status + /// watch at the bound venue. + pub async fn observe(&self, receipt: &[u8]) -> Result<(), ClientError> { + Ok(self.transport.observe(&V::ID, receipt).await?) + } + /// Report where a previously submitted intent is in its life. pub async fn status(&self, receipt: &[u8]) -> Result { Ok(self.transport.status(&V::ID, receipt).await?) diff --git a/modules/ethflow-watcher/Cargo.toml b/modules/ethflow-watcher/Cargo.toml index 3beb4710..eff3f0ba 100644 --- a/modules/ethflow-watcher/Cargo.toml +++ b/modules/ethflow-watcher/Cargo.toml @@ -14,7 +14,8 @@ crate-type = ["cdylib", "rlib"] [dependencies] nexum-sdk = { path = "../../crates/nexum-sdk" } -shepherd-sdk = { path = "../../crates/shepherd-sdk" } +videre-sdk = { path = "../../crates/videre-sdk" } +cow-venue = { path = "../../crates/cow-venue", features = ["client", "assembly"] } 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"] } @@ -22,5 +23,4 @@ tracing = { version = "0.1", default-features = false } wit-bindgen = { version = "0.59", default-features = false, features = ["macros", "realloc"] } [dev-dependencies] -shepherd-sdk-test = { path = "../../crates/shepherd-sdk-test" } nexum-sdk-test = { path = "../../crates/nexum-sdk-test" } diff --git a/modules/ethflow-watcher/module.toml b/modules/ethflow-watcher/module.toml index 47cc57c0..ae9cb5f0 100644 --- a/modules/ethflow-watcher/module.toml +++ b/modules/ethflow-watcher/module.toml @@ -1,6 +1,6 @@ -# ethflow-watcher: see `CoWSwapEthFlow.OrderPlacement`, lift the embedded -# `GPv2OrderData` into an `OrderCreation`, and submit it via the CoW -# Protocol orderbook with the EIP-1271 signing scheme. +# ethflow-watcher: decode `CoWSwapEthFlow.OrderPlacement` logs, compute +# the orderbook UID, and put it under the pool's status watch via the +# cow venue adapter. Observe-only: the module never submits. [module] name = "ethflow-watcher" @@ -10,16 +10,13 @@ version = "0.1.0" component = "sha256:0000000000000000000000000000000000000000000000000000000000000000" [capabilities] -# Least-privilege: the module exercises logging, local-store and -# cow-api today; `chain` is listed as optional so a follow-up (e.g. -# adding an eth_call to read the EthFlow refund pointer) -# can use it without manifest churn, without widening the required -# grant for a capability the module does not call yet. -required = ["logging", "local-store", "cow-api"] -optional = ["chain"] +# Least-privilege: `client` for the pool observe path, `logging` for +# structured runtime logs, `local-store` for the `observed:` journal. +required = ["client", "logging", "local-store"] +optional = [] [capabilities.http] -# All outbound HTTP goes through `cow-api`; no direct `http` calls. +# All venue I/O rides the pool router; no direct `http` calls. allow = [] # --- subscriptions ------------------------------------------------------ @@ -39,3 +36,9 @@ kind = "chain-log" chain_id = 11155111 address = "0xbA3cB449bD2B4ADddBc894D8697F5170800EAdeC" event_signature = "0xcf5f9de2984132265203b5c335b25727702ca77262ff622e136baa7362bf1da9" + +# Status transitions the registry polls for watched receipts at the cow +# venue; the module journals `observed:{uid}` on the first one. +[[subscription]] +kind = "intent-status" +venue = "cow" diff --git a/modules/ethflow-watcher/src/lib.rs b/modules/ethflow-watcher/src/lib.rs index 8cca9f36..67c92f9e 100644 --- a/modules/ethflow-watcher/src/lib.rs +++ b/modules/ethflow-watcher/src/lib.rs @@ -1,23 +1,21 @@ //! # ethflow-watcher (Shepherd module) //! //! Subscribes to `CoWSwapOnchainOrders.OrderPlacement` logs from the -//! canonical CoWSwap EthFlow contracts and verifies the orderbook's -//! native indexer caught each placement via `GET /api/v1/orders/{uid}`. -//! See `strategy.rs` for the design rationale: the orderbook -//! backend indexes EthFlow `OrderPlacement` events server-side with -//! its own dual-validTo bookkeeping, so `POST /api/v1/orders` is -//! structurally the wrong endpoint for on-chain EthFlow orders. The -//! module observes and verifies, it does not submit. +//! canonical CoWSwap EthFlow contracts, computes each placement's +//! orderbook UID, and puts it under the pool's status watch through the +//! typed cow venue client. The registry polls the cow adapter and fans +//! transitions back as `intent-status` events; the module journals +//! `observed:{uid}` on the first one. Observe-only: it never submits +//! (see `strategy.rs`). //! //! ## Module layout //! -//! - `strategy.rs` holds the pure logic and unit tests against -//! `nexum_sdk::host::Host`. It does not know `wit-bindgen` -//! exists. -//! - `lib.rs` (this file) is the per-cdylib glue: wit-bindgen import -//! shims, the `WitBindgenHost` adapter that bridges the generated -//! free functions to the SDK traits, and the `Guest` impl that -//! delegates the `chain-logs` event variant to `strategy::on_chain_logs`. +//! - `strategy.rs` holds the pure logic and unit tests against the +//! `nexum_sdk::host` and `videre_sdk` client seams. It does not know +//! `wit-bindgen` exists. +//! - `lib.rs` (this file) is the per-cdylib glue: the +//! `#[videre_sdk::keeper]` handler impl that binds the world derived +//! from `module.toml` and delegates each event to `strategy`. // wit_bindgen::generate! expands to host-import shims whose arity // matches the WIT signatures, which can exceed clippy's @@ -25,59 +23,48 @@ #![cfg_attr(not(test), warn(unused_crate_dependencies))] #![allow(clippy::too_many_arguments)] -// The wit-bindgen-generated import shims only resolve against the -// engine's wasm component host - they have no native-target -// equivalent. Cfg-gate the entire glue layer so the `rlib` artefact -// (consumed by `shepherd-backtest`) carries just the -// strategy code without dangling `extern "C"` imports. The -// `use wit_bindgen as _` line below silences the unused-crate -// lint on native targets where the macro never expands. +// The keeper glue only resolves against the engine's wasm component +// host. Cfg-gate it so the `rlib` artefact (consumed by +// `shepherd-backtest`) carries just the strategy code without dangling +// `extern "C"` imports; the `use wit_bindgen as _` line silences the +// unused-crate lint on native targets where the macro never expands. #[cfg(not(target_arch = "wasm32"))] use wit_bindgen as _; -#[cfg(target_arch = "wasm32")] -wit_bindgen::generate!({ - path: [ - "../../wit/nexum-host", - "../../wit/shepherd-cow", - ], - world: "shepherd:cow/shepherd", - generate_all, -}); - pub mod strategy; -// `WitBindgenHost` and the fault and level `From` impls -// are generated below. Single source of truth in `nexum-sdk` + `shepherd-sdk`. -// Gated on `wasm32` so the strategy can be reused in native targets -// (e.g. the backtest replay harness in `crates/shepherd-backtest`). #[cfg(target_arch = "wasm32")] -use nexum::host::types; +mod glue { + use cow_venue::client::CowClient; -#[cfg(target_arch = "wasm32")] -shepherd_sdk::bind_cow_host_via_wit_bindgen!(); + use crate::strategy; -#[cfg(target_arch = "wasm32")] -struct EthFlowWatcher; + struct EthFlowWatcher; -#[cfg(target_arch = "wasm32")] -impl Guest for EthFlowWatcher { - fn init(_config: Vec<(String, String)>) -> Result<(), Fault> { - install_tracing(); - tracing::info!("ethflow-watcher init"); - Ok(()) - } + #[videre_sdk::keeper] + impl EthFlowWatcher { + fn init(_config: Vec<(String, String)>) -> Result<(), Fault> { + install_tracing(); + tracing::info!("ethflow-watcher init"); + Ok(()) + } - fn on_event(event: types::Event) -> Result<(), Fault> { - if let types::Event::ChainLogs(batch) = event { + async fn on_chain_logs(batch: nexum::host::types::ChainLogs) -> Result<(), Fault> { let logs: Vec = batch.logs.into_iter().map(Into::into).collect(); - strategy::on_chain_logs(&WitBindgenHost, batch.chain_id, &logs)?; + strategy::on_chain_logs(&WitBindgenHost, &CowClient::new(), batch.chain_id, &logs) + .await?; + Ok(()) + } + + fn on_intent_status(update: nexum::host::types::IntentStatusUpdate) -> Result<(), Fault> { + strategy::on_intent_status( + &WitBindgenHost, + &update.venue, + &update.receipt, + &update.status, + )?; + Ok(()) } - // Block / Tick / Message are not used by this module. - Ok(()) } } - -#[cfg(target_arch = "wasm32")] -export!(EthFlowWatcher); diff --git a/modules/ethflow-watcher/src/strategy.rs b/modules/ethflow-watcher/src/strategy.rs index 4d24db00..002a9984 100644 --- a/modules/ethflow-watcher/src/strategy.rs +++ b/modules/ethflow-watcher/src/strategy.rs @@ -1,11 +1,11 @@ //! Pure strategy logic for the ethflow-watcher module. //! -//! Every interaction with the world flows through the -//! `nexum_sdk::host::Host` trait seam - no direct calls to wit- -//! bindgen-generated free functions live here. The `lib.rs` glue -//! wraps a `WitBindgenHost` adapter around the per-cdylib wit-bindgen -//! imports and hands it to [`on_chain_logs`]; tests under `#[cfg(test)]` -//! drive the same function with `shepherd_sdk_test::MockHost`. +//! Every interaction with the world flows through two seams: the +//! `nexum_sdk::host` traits for the `observed:` journal and the typed +//! [`CowClient`] over [`VenueTransport`] for the pool. The `lib.rs` +//! glue binds both to the module's own imports; tests drive the same +//! functions with `nexum_sdk_test::MockHost` and an in-memory spy +//! transport. //! //! ## Design (redesign) //! @@ -13,36 +13,35 @@ //! to `/api/v1/orders` with the EthFlow contract as the EIP-1271 owner. //! Empirical evidence (2026-06-22 Sepolia soak) showed that path cannot //! succeed: the orderbook backend indexes EthFlow `OrderPlacement` -//! events natively and writes server-only fields (`onchainUser`, -//! `onchainOrderData`, `ethflowData.userValidTo`) the public POST body -//! does not carry. Submissions through `/api/v1/orders` are rejected -//! with `ExcessiveValidTo` even though the same UID is `fulfilled` on -//! the orderbook by the time we look. +//! events natively and writes server-only fields the public POST body +//! does not carry. //! -//! This strategy therefore **observes + verifies** instead of -//! submitting: +//! This strategy therefore **observes + verifies** through the pool: //! //! 1. Decode the `OrderPlacement` log against the canonical EthFlow //! contract addresses. -//! 2. Compute the orderbook UID from the on-chain order shape -//! (`OrderData::uid(domain, contract)`). -//! 3. GET `/api/v1/orders/{uid}` to confirm the orderbook indexer -//! picked up the placement. On 200, record `observed:{uid}` in the -//! keeper idempotency journal so log re-delivery is a no-op. On -//! 404, log at Info - typical indexer lag, do not write the marker -//! so the next re-delivery rechecks. Any other error is logged at -//! Warn for operator follow-up. +//! 2. Compute the orderbook UID from the on-chain order shape: the +//! externally-obtained receipt at the cow venue. +//! 3. `observe` the receipt through the pool router, which polls the +//! cow adapter's `status` and fans transitions back as +//! `intent-status` events. On the first one, record `observed:{uid}` +//! in the keeper idempotency journal so log re-delivery is a no-op. +//! A refused observe is logged and left unjournalled, so the next +//! re-delivery retries. use alloy_primitives::{Address, Bytes}; use alloy_sol_types::SolEvent; +use cow_venue::assembly; +use cow_venue::client::{CowClient, CowVenue}; use cowprotocol::{ Chain, CoWSwapOnchainOrders::OrderPlacement, ETH_FLOW_PRODUCTION, ETH_FLOW_STAGING, GPv2OrderData, OnchainSignature, OrderUid, }; use nexum_sdk::events::Log; -use nexum_sdk::host::Fault; +use nexum_sdk::host::{Fault, LocalStoreHost}; use nexum_sdk::keeper::Journal; -use shepherd_sdk::cow::{CowApiError, CowHost, events, gpv2_to_order_data}; +use nexum_sdk::status_body::StatusBody; +use videre_sdk::client::{Venue, VenueTransport}; /// Decoded payload of a `CoWSwapOnchainOrders.OrderPlacement` log. /// `GPv2OrderData` is ~300 bytes; box it so the struct stays @@ -70,16 +69,51 @@ pub(crate) struct DecodedPlacement { } /// Entry point: decode every `OrderPlacement` chain-log in a dispatch batch -/// and feed each decoded placement to the observe path. -pub fn on_chain_logs(host: &H, chain_id: u64, logs: &[Log]) -> Result<(), Fault> { +/// and put each decoded placement's UID under the pool's status watch. +pub async fn on_chain_logs( + host: &H, + venue: &CowClient, + chain_id: u64, + logs: &[Log], +) -> Result<(), Fault> { for log in logs { if let Some(placement) = decode_order_placement(log) { - observe_placement(host, chain_id, &placement)?; + observe_placement(host, venue, chain_id, &placement).await?; } } Ok(()) } +/// A registry status transition for a watched receipt. Foreign venues +/// are ignored; the first transition for a cow receipt records the +/// `observed:{uid}` marker (the orderbook indexed the placement), and +/// every transition is logged. +pub fn on_intent_status( + host: &H, + venue: &str, + receipt: &[u8], + status: &[u8], +) -> Result<(), Fault> { + if venue != CowVenue::ID.as_str() { + return Ok(()); + } + let Ok(uid) = OrderUid::try_from(receipt) else { + tracing::warn!( + "ethflow status update with a non-uid receipt ({} bytes)", + receipt.len(), + ); + return Ok(()); + }; + let body = StatusBody::decode(status).map_err(|e| Fault::InvalidInput(e.to_string()))?; + let uid_hex = format!("{uid}"); + let journal = Journal::observed(host); + if !journal.contains(&uid_hex)? { + journal.record(&uid_hex)?; + } + tracing::info!("ethflow observed {uid_hex}: {:?}", body.status); + Ok(()) +} + // ---- decode ---- /// Decode a raw event log against `CoWSwapOnchainOrders.OrderPlacement`. @@ -96,7 +130,7 @@ pub(crate) fn decode_order_placement(log: &Log) -> Option { if contract != ETH_FLOW_PRODUCTION && contract != ETH_FLOW_STAGING { return None; } - if log.topics().first() != Some(&events::ORDER_PLACEMENT.topic0) { + if log.topics().first() != Some(&OrderPlacement::SIGNATURE_HASH) { return None; } let decoded = OrderPlacement::decode_log(&log.inner).ok()?; @@ -109,57 +143,43 @@ pub(crate) fn decode_order_placement(log: &Log) -> Option { }) } -// ---- observe + verify (redesign) ---- +// ---- observe + verify (pool) ---- -/// Compute the orderbook UID for the placement and confirm the -/// orderbook's native EthFlow indexer picked it up. -fn observe_placement( +/// Compute the orderbook UID for the placement and put it under the +/// pool's status watch. A refused observe (venue down, watch set full) +/// is logged and left unjournalled, so re-delivery retries. +async fn observe_placement( host: &H, + venue: &CowClient, chain_id: u64, placement: &DecodedPlacement, ) -> Result<(), Fault> { - let uid_hex = match compute_uid(chain_id, placement) { - Some(uid) => format!("{uid}"), - None => { - tracing::warn!( - "ethflow uid build skipped (sender={:#x}): unsupported chain {chain_id} or unknown order marker", - placement.sender, - ); - return Ok(()); - } + let Some(uid) = compute_uid(chain_id, placement) else { + tracing::warn!( + "ethflow uid build skipped (sender={:#x}): unsupported chain {chain_id} or unknown order marker", + placement.sender, + ); + return Ok(()); }; + let uid_hex = format!("{uid}"); - // Idempotency: once verified, do not re-check on log re-delivery + // Idempotency: once observed, do not re-watch on log re-delivery // (engine restart, reorg replay, supervisor restart). let journal = Journal::observed(host); if journal.contains(&uid_hex)? { return Ok(()); } - let path = format!("/api/v1/orders/{uid_hex}"); - match host.cow_api_request(chain_id, "GET", &path, None) { - Ok(_) => { - journal.record(&uid_hex)?; + match venue.observe(uid.as_slice()).await { + Ok(()) => { tracing::info!( - "ethflow observed {uid_hex} (orderbook indexed, sender={:#x})", - placement.sender, - ); - } - Err(CowApiError::Http(http)) if http.status == 404 => { - // Indexer lag is expected immediately after the block lands - - // shepherd's WebSocket can deliver the log a few hundred - // milliseconds before the orderbook's own indexer commits. - // Do NOT write the marker so a later re-delivery (or a future - // block-tick poll) can recheck. Info keeps the soak dashboard - // quiet on normal lag. - tracing::info!( - "ethflow not yet indexed {uid_hex} (sender={:#x}); will recheck on re-delivery", + "ethflow watching {uid_hex} (sender={:#x})", placement.sender, ); } Err(err) => { tracing::warn!( - "ethflow indexer check failed {uid_hex}: {err} (sender={:#x})", + "ethflow watch failed {uid_hex}: {err} (sender={:#x})", placement.sender, ); } @@ -168,30 +188,140 @@ fn observe_placement( } /// Compute the canonical 56-byte orderbook UID for the placement. -/// `OrderData::uid` packs `digest || owner || valid_to`; the owner -/// input is the EthFlow contract (which signs via EIP-1271), not the -/// native-token sender. +/// The UID packs `digest || owner || valid_to`; the owner input is the +/// EthFlow contract (which signs via EIP-1271), not the native-token +/// sender. fn compute_uid(chain_id: u64, placement: &DecodedPlacement) -> Option { let chain = Chain::try_from(chain_id).ok()?; - let domain = chain.settlement_domain(); - let order_data = gpv2_to_order_data(&placement.order)?; - Some(order_data.uid(&domain, placement.contract)) + let order = assembly::gpv2_to_order_data(&placement.order)?; + Some(assembly::order_uid(chain, &order, placement.contract)) } #[cfg(test)] mod tests { - use super::*; + use std::cell::RefCell; + use std::collections::VecDeque; + use std::rc::Rc; + use alloy_primitives::{U256, address, hex}; use alloy_sol_types::SolValue; use cowprotocol::{BuyTokenDestination, OnchainSigningScheme, OrderKind, SellTokenSource}; use nexum_sdk::Level; - use nexum_sdk::host::{Fault, LocalStoreHost as _}; - use nexum_sdk_test::capture_tracing; - use shepherd_sdk::cow::HttpFailure; - use shepherd_sdk_test::MockHost; + use nexum_sdk::host::LocalStoreHost as _; + use nexum_sdk::status_body::IntentStatus as Lifecycle; + use nexum_sdk_test::{MockHost, capture_tracing}; + use videre_sdk::client::VenueId; + use videre_sdk::rt::complete; + use videre_sdk::{IntentStatus, Quotation, SubmitOutcome, VenueFault}; + + use super::*; const SEPOLIA: u64 = 11_155_111; + /// One recorded transport call: which verb, and for `observe` the + /// routed venue and receipt. + #[derive(Clone, Debug, Eq, PartialEq)] + enum Call { + Quote, + Submit, + Observe(String, Vec), + Status, + Cancel, + } + + /// Records every call; `observe` pops a scripted response and + /// defaults to accepted once the script drains. The other verbs + /// refuse: the observe-only strategy must never reach them. + /// Cloneable over shared state so the test keeps a handle after + /// one moves into the client. + #[derive(Clone, Default)] + struct SpyVenues { + calls: Rc>>, + observe_script: Rc>>>, + } + + impl SpyVenues { + fn script_observe(&self, result: Result<(), VenueFault>) { + self.observe_script.borrow_mut().push_back(result); + } + + fn calls(&self) -> Vec { + self.calls.borrow().clone() + } + + fn observe_count(&self) -> usize { + self.calls + .borrow() + .iter() + .filter(|c| matches!(c, Call::Observe(..))) + .count() + } + } + + impl videre_sdk::client::sealed::SealedTransport for SpyVenues {} + + impl VenueTransport for SpyVenues { + async fn quote(&self, _venue: &VenueId, _body: Vec) -> Result { + self.calls.borrow_mut().push(Call::Quote); + Err(VenueFault::Unsupported) + } + + async fn submit( + &self, + _venue: &VenueId, + _body: Vec, + ) -> Result { + self.calls.borrow_mut().push(Call::Submit); + Err(VenueFault::Unsupported) + } + + async fn observe(&self, venue: &VenueId, receipt: &[u8]) -> Result<(), VenueFault> { + self.calls + .borrow_mut() + .push(Call::Observe(venue.to_string(), receipt.to_vec())); + self.observe_script + .borrow_mut() + .pop_front() + .unwrap_or(Ok(())) + } + + async fn status( + &self, + _venue: &VenueId, + _receipt: &[u8], + ) -> Result { + self.calls.borrow_mut().push(Call::Status); + Err(VenueFault::Unsupported) + } + + async fn cancel(&self, _venue: &VenueId, _receipt: &[u8]) -> Result<(), VenueFault> { + self.calls.borrow_mut().push(Call::Cancel); + Err(VenueFault::Unsupported) + } + } + + /// Drive the async strategy on the synchronous test boundary. + fn run_logs( + host: &MockHost, + spy: &SpyVenues, + chain_id: u64, + logs: &[Log], + ) -> Result<(), Fault> { + let client = CowClient::with_transport(spy.clone()); + complete(on_chain_logs(host, &client, chain_id, logs)) + .expect("guest futures complete in one poll") + } + + fn open_status() -> Vec { + StatusBody { + status: Lifecycle::Open, + proof: None, + reason: None, + } + .encode() + .expect("status body encodes") + } + fn sample_order() -> GPv2OrderData { GPv2OrderData { sellToken: address!("6810e776880C02933D47DB1b9fc05908e5386b96"), @@ -246,11 +376,14 @@ mod tests { .into() } - fn computed_uid(placement: &DecodedPlacement) -> String { - format!( - "{}", - compute_uid(SEPOLIA, placement).expect("sepolia + canonical markers") - ) + fn sample_log() -> Log { + let (topics, data) = encode_log(&sample_event()); + make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data) + } + + fn sample_uid() -> OrderUid { + let placement = decode_order_placement(&sample_log()).expect("decode succeeds"); + compute_uid(SEPOLIA, &placement).expect("sepolia + canonical markers") } // ---- decode (invariants preserved) ---- @@ -258,9 +391,7 @@ mod tests { #[test] fn decodes_well_formed_placement() { let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let decoded = decode_order_placement(&log).expect("decode succeeds"); + let decoded = decode_order_placement(&sample_log()).expect("decode succeeds"); assert_eq!(decoded.contract, ETH_FLOW_PRODUCTION); assert_eq!(decoded.sender, event.sender); assert_eq!(decoded.signature.scheme, OnchainSigningScheme::Eip1271); @@ -294,11 +425,7 @@ mod tests { #[test] fn compute_uid_pins_owner_to_ethflow_contract_and_validto() { let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let decoded = decode_order_placement(&log).unwrap(); - - let uid = compute_uid(SEPOLIA, &decoded).expect("sepolia + canonical markers"); + let uid = sample_uid(); let bytes: [u8; 56] = uid.into(); // owner suffix (bytes 32..52) = EthFlow contract address. assert_eq!(&bytes[32..52], ETH_FLOW_PRODUCTION.as_slice()); @@ -311,273 +438,202 @@ mod tests { #[test] fn compute_uid_returns_none_on_unsupported_chain() { - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let decoded = decode_order_placement(&log).unwrap(); + let decoded = decode_order_placement(&sample_log()).unwrap(); assert!(compute_uid(9999, &decoded).is_none()); } - // ---- observe + verify dispatch (Host-trait integration) ---- + // ---- observe via the pool (transport integration) ---- - /// 200 from `GET /api/v1/orders/{uid}` → `observed:{uid}` written - /// + Info log + zero submit attempts. + /// A placement registers exactly one status watch at the cow venue + /// with the computed UID as the receipt, and journals nothing yet: + /// the marker waits for the first status transition. #[test] - fn placement_log_marks_observed_on_orderbook_200() { + fn placement_log_registers_the_uid_watch() { let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let placement = decode_order_placement(&log).unwrap(); - let uid = computed_uid(&placement); - - // Minimal stub of the orderbook's GET response - strategy only - // checks for 200 vs 404 vs other, the body is opaque to it. - host.cow_api.respond_to_request_for( - "GET", - format!("/api/v1/orders/{uid}"), - Ok(r#"{"status":"fulfilled"}"#.to_string()), - ); + let spy = SpyVenues::default(); + let uid = sample_uid(); - on_chain_logs(&host, SEPOLIA, &[log]).unwrap(); + run_logs(&host, &spy, SEPOLIA, &[sample_log()]).unwrap(); - assert!( - host.store - .snapshot() - .contains_key(&format!("observed:{uid}")), - "200 response must write observed:{{uid}} marker" - ); assert_eq!( - host.cow_api.request_calls().len(), - 1, - "exactly one orderbook GET per log" + spy.calls(), + vec![Call::Observe("cow".to_owned(), uid.as_slice().to_vec())], + "exactly one observe, nothing else", ); - assert_eq!( - host.cow_api.call_count(), - 0, - "observe path must never call submit_order" - ); - } - - /// 404 from `GET /api/v1/orders/{uid}` → no marker written + Info - /// log + the next re-delivery rechecks (no early dedup). - #[test] - fn placement_log_does_not_mark_observed_on_orderbook_404() { - let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let placement = decode_order_placement(&log).unwrap(); - let uid = computed_uid(&placement); - - host.cow_api - .respond_to_request(Err(CowApiError::Http(HttpFailure { - status: 404, - body: None, - }))); - - let (result, logs) = capture_tracing(|| on_chain_logs(&host, SEPOLIA, &[log])); - result.unwrap(); - assert!( - !host - .store - .snapshot() - .contains_key(&format!("observed:{uid}")), - "404 must NOT write observed: so re-delivery can recheck" - ); - assert_eq!( - host.cow_api.request_calls().len(), - 1, - "the orderbook GET was attempted" - ); - let ev = logs.expect_one(|e| e.message.contains("not yet indexed")); - assert_eq!( - ev.level, - Level::INFO, - "indexer lag is expected; Info keeps soak dashboards quiet" + host.store.snapshot().is_empty(), + "observed:{{uid}} waits for the first status transition", ); } - /// Non-404 error from the orderbook check → Warn log + no marker. + /// A refused observe warns, journals nothing, and the next + /// re-delivery retries the watch. #[test] - fn placement_log_warns_on_orderbook_other_error() { + fn watch_refusal_warns_and_redelivery_retries() { let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - - host.cow_api - .respond_to_request(Err(CowApiError::Fault(Fault::Unavailable( - "bad gateway".into(), - )))); + let spy = SpyVenues::default(); + spy.script_observe(Err(VenueFault::Unavailable("venue down".to_owned()))); - let (result, logs) = capture_tracing(|| on_chain_logs(&host, SEPOLIA, &[log])); + let (result, logs) = capture_tracing(|| run_logs(&host, &spy, SEPOLIA, &[sample_log()])); result.unwrap(); - assert!( - host.store.snapshot().is_empty(), - "non-404 error must not write any marker" - ); - assert_eq!( - host.cow_api.request_calls().len(), - 1, - "the orderbook GET was attempted" - ); - logs.expect_one(|e| e.level == Level::WARN && e.message.contains("indexer check failed")); + assert!(host.store.snapshot().is_empty()); + logs.expect_one(|e| e.level == Level::WARN && e.message.contains("watch failed")); + + // Unjournalled, so the re-delivered log observes again. + run_logs(&host, &spy, SEPOLIA, &[sample_log()]).unwrap(); + assert_eq!(spy.observe_count(), 2); } /// Idempotency: a placement that already has `observed:{uid}` in - /// local store does NOT trigger a fresh GET on re-delivery. + /// local store does NOT touch the pool on re-delivery. #[test] fn previously_observed_placement_is_skipped_on_redelivery() { let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let placement = decode_order_placement(&log).unwrap(); - let uid = computed_uid(&placement); + let spy = SpyVenues::default(); + let uid = sample_uid(); host.store .set(&format!("observed:{uid}"), b"") .expect("seed observed marker"); - on_chain_logs(&host, SEPOLIA, &[log]).unwrap(); + run_logs(&host, &spy, SEPOLIA, &[sample_log()]).unwrap(); - assert_eq!( - host.cow_api.request_calls().len(), - 0, - "observed:{{uid}} must short-circuit before the orderbook GET" - ); - assert_eq!( - host.cow_api.call_count(), - 0, - "and certainly no submit_order" + assert!( + spy.calls().is_empty(), + "observed:{{uid}} must short-circuit before the pool", ); } /// Defensive: unsupported chain id surfaces a Warn but does not - /// panic and does not touch the orderbook. + /// panic and does not touch the pool. #[test] - fn unsupported_chain_logs_warn_without_orderbook_call() { + fn unsupported_chain_logs_warn_without_venue_call() { let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); + let spy = SpyVenues::default(); // 9999 is not in cowprotocol::Chain. - let (result, logs) = capture_tracing(|| on_chain_logs(&host, 9999, &[log])); + let (result, logs) = capture_tracing(|| run_logs(&host, &spy, 9999, &[sample_log()])); result.unwrap(); - assert_eq!(host.cow_api.request_calls().len(), 0); - assert_eq!(host.cow_api.call_count(), 0); + assert!(spy.calls().is_empty()); assert!(host.store.snapshot().is_empty()); logs.expect_one(|e| { e.level == Level::WARN && e.message.contains("ethflow uid build skipped") }); } - /// Strategy must never call `submit_order` - the trait still - /// exposes it for other modules (twap-monitor legitimately - /// submits), but ethflow-watcher's observe design never does. - /// Belt-and-suspenders regression guard. + /// The strategy is observer-only: no call path reaches quote, + /// submit, status, or cancel. Belt-and-suspenders regression guard. #[test] - fn strategy_never_calls_submit_order() { + fn strategy_never_submits() { let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - host.cow_api.respond_to_request(Ok("{}".to_string())); + let spy = SpyVenues::default(); - on_chain_logs(&host, SEPOLIA, &[log]).unwrap(); + run_logs(&host, &spy, SEPOLIA, &[sample_log()]).unwrap(); - assert_eq!( - host.cow_api.call_count(), - 0, - "submit_order count must stay at zero - ethflow-watcher is observer-only" + assert!( + spy.calls().iter().all(|c| matches!(c, Call::Observe(..))), + "observe is the only verb the strategy may use", ); } - /// Guard: the `sol!` decoder's topic-0 matches the - /// `shepherd:cow/cow-events` package of record. A typo or ABI - /// drift would silently miss every EthFlow event. - #[test] - fn topic0_matches_order_placement_canonical_signature() { - assert_eq!( - OrderPlacement::SIGNATURE_HASH, - events::ORDER_PLACEMENT.topic0, - "sol! topic-0 must match the shepherd:cow/cow-events pin", - ); - } + // ---- intent-status transitions ---- - /// Stronger guard than the constant check above: read the shipped - /// `module.toml` and assert its pinned `event_signature` actually - /// equals the package-of-record topic-0 - catches a manifest/code - /// drift the decoder assertion cannot see. + /// The first cow transition journals `observed:{uid}` and logs it. #[test] - fn manifest_topic0_matches_order_placement_signature_hash() { - let manifest = include_str!("../module.toml"); - let expected = format!("{:#x}", events::ORDER_PLACEMENT.topic0); + fn status_update_journals_the_observed_marker() { + let host = MockHost::new(); + let uid = sample_uid(); + + let (result, logs) = + capture_tracing(|| on_intent_status(&host, "cow", uid.as_slice(), &open_status())); + result.unwrap(); + assert!( - manifest.contains(&expected), - "module.toml event_signature must equal the shepherd:cow/cow-events pin ({expected})", + host.store + .snapshot() + .contains_key(&format!("observed:{uid}")), + "the first transition must write observed:{{uid}}", ); + let ev = logs.expect_one(|e| e.message.contains("ethflow observed")); + assert_eq!(ev.level, Level::INFO); } - /// 429 (rate-limit) from the orderbook check → Warn log + no marker. - /// Verifies the strategy does not conflate 429 with 404 (which would - /// suppress the warning) and does not panic or return an error. + /// Later transitions keep the single marker and stay Ok. #[test] - fn placement_log_warns_on_429_rate_limit() { + fn repeated_transitions_keep_one_marker() { let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let placement = decode_order_placement(&log).unwrap(); - let uid = computed_uid(&placement); + let uid = sample_uid(); - host.cow_api - .respond_to_request(Err(CowApiError::Http(HttpFailure { - status: 429, - body: Some("Too Many Requests".to_string()), - }))); + on_intent_status(&host, "cow", uid.as_slice(), &open_status()).unwrap(); + let fulfilled = StatusBody { + status: Lifecycle::Fulfilled, + proof: None, + reason: None, + } + .encode() + .expect("status body encodes"); + on_intent_status(&host, "cow", uid.as_slice(), &fulfilled).unwrap(); - let (result, logs) = capture_tracing(|| on_chain_logs(&host, SEPOLIA, &[log])); - result.unwrap(); + assert_eq!(host.store.snapshot().len(), 1); + } - assert!( - !host - .store - .snapshot() - .contains_key(&format!("observed:{uid}")), - "429 must NOT write observed: marker" - ); - logs.expect_one(|e| e.level == Level::WARN && e.message.contains("indexer check failed")); + /// A transition from a foreign venue is not ethflow's: ignored. + #[test] + fn foreign_venue_status_update_is_ignored() { + let host = MockHost::new(); + on_intent_status(&host, "echo-venue", sample_uid().as_slice(), &open_status()).unwrap(); + assert!(host.store.snapshot().is_empty()); } - /// HTTP 200 with a malformed (non-JSON) body → `observed:{uid}` still - /// written. The strategy only inspects Ok vs Err, never parses the body, - /// so any successful response confirms indexer pickup regardless of body - /// content. + /// A cow receipt that is not a 56-byte UID warns without a marker. #[test] - fn placement_log_marks_observed_on_malformed_response_body() { + fn non_uid_receipt_warns_without_marker() { let host = MockHost::new(); - let event = sample_event(); - let (topics, data) = encode_log(&event); - let log = make_log(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let placement = decode_order_placement(&log).unwrap(); - let uid = computed_uid(&placement); + let (result, logs) = + capture_tracing(|| on_intent_status(&host, "cow", b"abc", &open_status())); + result.unwrap(); + assert!(host.store.snapshot().is_empty()); + logs.expect_one(|e| e.level == Level::WARN && e.message.contains("non-uid receipt")); + } - host.cow_api - .respond_to_request(Ok("not-valid-json{{{{".to_string())); + /// A status body the SDK cannot decode is a typed fault, never a + /// silent marker. + #[test] + fn malformed_status_body_is_a_typed_fault() { + let host = MockHost::new(); + let err = on_intent_status(&host, "cow", sample_uid().as_slice(), &[0xFF, 0x00]) + .expect_err("undecodable status body"); + assert!(matches!(err, Fault::InvalidInput(_))); + assert!(host.store.snapshot().is_empty()); + } - on_chain_logs(&host, SEPOLIA, &[log]).unwrap(); + // ---- package-of-record parity ---- + /// Guard: the `sol!` decoder's topic-0 matches the + /// `shepherd:cow/cow-events` package of record. A typo or ABI + /// drift would silently miss every EthFlow event. + #[test] + fn topic0_matches_the_cow_events_package_of_record() { + let wit = include_str!("../../../wit/shepherd-cow/cow-events.wit"); + let expected = format!("{:#x}", OrderPlacement::SIGNATURE_HASH); assert!( - host.store - .snapshot() - .contains_key(&format!("observed:{uid}")), - "200 with malformed body must still write observed: — strategy does not parse the response", + wit.contains(&expected), + "sol! topic-0 must match the shepherd:cow/cow-events pin ({expected})", + ); + } + + /// Read the shipped `module.toml` and assert its pinned + /// `event_signature` equals the decoder topic-0 - catches a + /// manifest/code drift the wit assertion cannot see. + #[test] + fn manifest_topic0_matches_order_placement_signature_hash() { + let manifest = include_str!("../module.toml"); + let expected = format!("{:#x}", OrderPlacement::SIGNATURE_HASH); + assert!( + manifest.contains(&expected), + "module.toml event_signature must equal the decoder topic-0 ({expected})", ); } } diff --git a/wit/videre-venue/venue.wit b/wit/videre-venue/venue.wit index 33e73f10..8a9685f2 100644 --- a/wit/videre-venue/venue.wit +++ b/wit/videre-venue/venue.wit @@ -7,6 +7,10 @@ interface client { quote: func(venue: string, body: list) -> result; submit: func(venue: string, body: list) -> result; + /// Put an externally-obtained receipt (e.g. an on-chain placement) + /// under the host's status watch; an accepted submit is watched + /// implicitly. Idempotent. + observe: func(venue: string, receipt: receipt) -> result<_, venue-error>; status: func(venue: string, receipt: receipt) -> result; cancel: func(venue: string, receipt: receipt) -> result<_, venue-error>; }