diff --git a/modules/ethflow-watcher/Cargo.toml b/modules/ethflow-watcher/Cargo.toml index aaad1967..260fbda1 100644 --- a/modules/ethflow-watcher/Cargo.toml +++ b/modules/ethflow-watcher/Cargo.toml @@ -13,8 +13,6 @@ shepherd-sdk = { path = "../../crates/shepherd-sdk" } cowprotocol = { version = "1.0.0-alpha.3", default-features = false } alloy-primitives = { version = "1.5", default-features = false, features = ["std"] } alloy-sol-types = { version = "1.5", default-features = false, features = ["std"] } -serde_json = { version = "1", default-features = false, features = ["alloc"] } -thiserror = "2" wit-bindgen = { version = "0.57", default-features = false, features = ["macros", "realloc"] } [dev-dependencies] diff --git a/modules/ethflow-watcher/src/lib.rs b/modules/ethflow-watcher/src/lib.rs index 5a70ca72..dd217e91 100644 --- a/modules/ethflow-watcher/src/lib.rs +++ b/modules/ethflow-watcher/src/lib.rs @@ -1,10 +1,13 @@ //! # ethflow-watcher (Shepherd module) //! //! Subscribes to `CoWSwapOnchainOrders.OrderPlacement` logs from the -//! CoWSwap EthFlow contracts and resubmits each placed order through -//! the orderbook API with `Signature::Eip1271`. The EthFlow contract -//! is the EIP-1271 verifier, so the `from` field on the resubmission -//! is the contract address (not the original native-token seller). +//! 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 (COW-1076): 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. //! //! ## Module layout (BLEU-855) //! diff --git a/modules/ethflow-watcher/src/strategy.rs b/modules/ethflow-watcher/src/strategy.rs index ee668466..0bc3ae02 100644 --- a/modules/ethflow-watcher/src/strategy.rs +++ b/modules/ethflow-watcher/src/strategy.rs @@ -1,19 +1,44 @@ //! Pure strategy logic for the ethflow-watcher module. //! //! Every interaction with the world flows through the -//! `shepherd_sdk::host::Host` trait seam - no direct calls to wit- +//! `shepherd_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_logs`]; tests under `#[cfg(test)]` -//! hand the same function a `shepherd_sdk_test::MockHost`. +//! drive the same function with `shepherd_sdk_test::MockHost`. +//! +//! ## Design (COW-1076 redesign) +//! +//! The original BLEU-833 design POSTed each on-chain `OrderPlacement` +//! 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. +//! +//! This strategy therefore **observes + verifies** instead of +//! submitting: +//! +//! 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, mark `observed:{uid}` 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. use alloy_primitives::{Address, B256, Bytes}; use alloy_sol_types::SolEvent; use cowprotocol::{ Chain, CoWSwapOnchainOrders::OrderPlacement, ETH_FLOW_PRODUCTION, ETH_FLOW_STAGING, - GPv2OrderData, OnchainSignature, OnchainSigningScheme, OrderCreation, OrderUid, Signature, + GPv2OrderData, OnchainSignature, OrderUid, }; -use shepherd_sdk::cow::{RetryAction, classify_api_error, gpv2_to_order_data}; +use shepherd_sdk::cow::gpv2_to_order_data; use shepherd_sdk::host::{Host, HostError, LogLevel}; /// Fields the strategy needs from a wit-bindgen `log`. Borrowed slices @@ -25,32 +50,37 @@ pub struct LogView<'a> { pub data: &'a [u8], } -/// Fully decoded payload of a `CoWSwapOnchainOrders.OrderPlacement` -/// log. `GPv2OrderData` is ~300 bytes; box it so the struct stays -/// cache-friendly through the submit path. +/// Decoded payload of a `CoWSwapOnchainOrders.OrderPlacement` log. +/// `GPv2OrderData` is ~300 bytes; box it so the struct stays +/// cache-friendly when threaded through the observe path. #[derive(Debug)] pub(crate) struct DecodedPlacement { - /// EthFlow contract that emitted the event - also the EIP-1271 - /// verifier `from` for the submitted `OrderCreation`. + /// EthFlow contract that emitted the event — also the EIP-1271 + /// owner of the resulting orderbook entry, used as the UID + /// `owner` input. pub(crate) contract: Address, - /// Original native-token seller - logged for diagnostics; the - /// orderbook's `from` is the contract (EIP-1271 owner), not this. + /// Original native-token seller. Logged for operator diagnostics; + /// not the orderbook owner. pub(crate) sender: Address, pub(crate) order: Box, + /// Decoded signature. Recorded by the orderbook indexer itself; + /// not consumed by the observe path. + #[allow(dead_code)] pub(crate) signature: OnchainSignature, - /// Refund pointer / opaque placer metadata. Not consumed by the - /// submit path today, but the field is part of the BLEU-832 - /// decoder contract. + /// Refund pointer / opaque placer metadata embedded in the + /// `OrderPlacement` event. The orderbook indexer derives + /// `ethflowData.userValidTo` from this blob; we keep it on the + /// struct for parity with the BLEU-832 decoder contract. #[allow(dead_code)] pub(crate) data: Bytes, } /// Entry point: decode every `OrderPlacement` log in a dispatch batch -/// and feed the decoded placement to the submit path. +/// and feed each decoded placement to the observe path. pub fn on_logs(host: &H, logs: &[LogView<'_>]) -> Result<(), HostError> { for log in logs { if let Some(placement) = decode_order_placement(log.address, log.topics, log.data) { - submit_placement(host, log.chain_id, &placement)?; + observe_placement(host, log.chain_id, &placement)?; } } Ok(()) @@ -62,9 +92,9 @@ pub fn on_logs(host: &H, logs: &[LogView<'_>]) -> Result<(), HostError> /// /// Returns `None` when: /// - the log's contract address is neither `ETH_FLOW_PRODUCTION` nor -/// `ETH_FLOW_STAGING` (defensive - the host's `[[subscription]]` -/// filter already pins the address, but a misconfigured engine -/// could still leak through); +/// `ETH_FLOW_STAGING` (defensive — the host's `[[subscription]]` +/// filter already pins the address, but a misconfigured engine could +/// still leak through); /// - topic0 does not match the event signature; or /// - the ABI body fails to decode. pub(crate) fn decode_order_placement( @@ -98,225 +128,84 @@ pub(crate) fn decode_order_placement( }) } -// ---- BLEU-833: submit + retry ---- - -#[derive(Debug, thiserror::Error)] -pub(crate) enum BuildError { - #[error("GPv2OrderData carried an unknown enum marker")] - UnknownMarker, - #[error("OnchainSignature carried an unknown scheme variant")] - UnknownSignatureScheme, - #[error("chain {0} is not supported by cowprotocol")] - UnsupportedChain(u64), - #[error(transparent)] - Cowprotocol(#[from] cowprotocol::Error), -} - -/// Lift `OnchainSignature` into the orderbook-typed `Signature`. The -/// EthFlow contract is the EIP-1271 verifier, so the `data` blob is -/// the raw verifier bytes; for `PreSign` the orderbook accepts an -/// empty payload. -fn to_signature(sig: &OnchainSignature) -> Option { - // sol! adds a hidden `__Invalid` variant on every Solidity enum, - // so exhaustive patterns require a wildcard; we surface it as - // `None` (caller falls back to skipping the placement) rather - // than panic. - match sig.scheme { - OnchainSigningScheme::Eip1271 => Some(Signature::Eip1271(sig.data.to_vec())), - OnchainSigningScheme::PreSign => Some(Signature::PreSign), - _ => None, - } -} +// ---- observe + verify (BLEU-833 redesign, COW-1076) ---- -/// Assemble `(OrderCreation, OrderUid)` from a placement. `from` is -/// the EthFlow contract (EIP-1271 owner). -/// -/// `app_data_json` is the canonical JSON document whose -/// `keccak256` matches `placement.order.appData`. The caller -/// resolves it via [`shepherd_sdk::cow::resolve_app_data`] (or -/// any equivalent path); passing a mismatching string makes -/// `from_signed_order_data` reject with "app_data JSON digest -/// does not match signed app_data hash" (COW-1074). -pub(crate) fn build_eth_flow_creation( - chain_id: u64, - placement: &DecodedPlacement, - app_data_json: String, -) -> Result<(OrderCreation, OrderUid), BuildError> { - let chain = Chain::try_from(chain_id).map_err(|_| BuildError::UnsupportedChain(chain_id))?; - let domain = chain.settlement_domain(); - let order_data = gpv2_to_order_data(&placement.order).ok_or(BuildError::UnknownMarker)?; - let uid = order_data.uid(&domain, placement.contract); - let signature = to_signature(&placement.signature).ok_or(BuildError::UnknownSignatureScheme)?; - let creation = OrderCreation::from_signed_order_data( - &order_data, - signature, - placement.contract, - app_data_json, - None, - )?; - Ok((creation, uid)) -} - -fn submit_placement( +/// Compute the orderbook UID for the placement and confirm the +/// orderbook's native EthFlow indexer picked it up. +fn observe_placement( host: &H, chain_id: u64, placement: &DecodedPlacement, ) -> Result<(), HostError> { - // COW-1074: cow-swap UI (and other clients) sign EthFlow - // placements with a non-empty `appData` hash pointing at a JSON - // document held by the orderbook's app_data registry. Resolve - // it before assembling the submission body; on 404 (orderbook - // doesn't mirror this hash) log a Warn and drop the placement - // — there is no path to recover without operator intervention. - let app_data_json = match shepherd_sdk::cow::resolve_app_data( - host, - chain_id, - &placement.order.appData.0, - ) { - Ok(json) => json, - Err(err) if err.code == 404 => { + let uid_hex = match compute_uid(chain_id, placement) { + Some(uid) => format!("{uid}"), + None => { host.log( LogLevel::Warn, &format!( - "ethflow submit skipped (sender={:#x}): appData hash not mirrored on orderbook", + "ethflow uid build skipped (sender={:#x}): unsupported chain {chain_id} or unknown order marker", placement.sender, ), ); return Ok(()); } - Err(err) => { - host.log( - LogLevel::Warn, - &format!( - "ethflow submit skipped (sender={:#x}): appData resolve failed ({}): {}", - placement.sender, err.code, err.message, - ), - ); - return Ok(()); - } }; - let (creation, uid) = match build_eth_flow_creation(chain_id, placement, app_data_json) { - Ok(x) => x, - Err(e) => { + // Idempotency: once verified, do not re-check on log re-delivery + // (engine restart, reorg replay, supervisor restart). + if host.get(&format!("observed:{uid_hex}"))?.is_some() { + return Ok(()); + } + + let path = format!("/api/v1/orders/{uid_hex}"); + match host.cow_api_request(chain_id, "GET", &path, None) { + Ok(_) => { + host.set(&format!("observed:{uid_hex}"), b"")?; host.log( - LogLevel::Warn, + LogLevel::Info, &format!( - "ethflow submit skipped (sender={:#x}): {e}", - placement.sender + "ethflow observed {uid_hex} (orderbook indexed, sender={:#x})", + placement.sender, ), ); - return Ok(()); } - }; - let uid_hex = format!("{uid}"); - - // Idempotency. A host reconnect or engine restart may replay the - // same OrderPlacement log; without the guard we would attempt a - // second submit, the orderbook would reject `DuplicateOrder` - // (permanent), and we would end up with both `submitted:` AND - // `dropped:` written for the same UID. `backoff:` is *not* a - // short-circuit - a previous transient error deserves a fresh - // attempt on re-delivery. - match prior_outcome(host, &uid_hex)? { - PriorOutcome::Submitted => { - host.log( - LogLevel::Info, - &format!("ethflow {uid_hex} already submitted; skipping"), - ); - return Ok(()); - } - PriorOutcome::Dropped => { + Err(err) if err.code == 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. host.log( LogLevel::Info, - &format!("ethflow {uid_hex} previously dropped; skipping"), + &format!( + "ethflow not yet indexed {uid_hex} (sender={:#x}); will recheck on re-delivery", + placement.sender, + ), ); - return Ok(()); } - PriorOutcome::None | PriorOutcome::Backoff => {} - } - - let body = match serde_json::to_vec(&creation) { - Ok(b) => b, - Err(e) => { + Err(err) => { host.log( - LogLevel::Error, - &format!("OrderCreation JSON encode failed: {e}"), + LogLevel::Warn, + &format!( + "ethflow indexer check failed {uid_hex} ({}): {} (sender={:#x})", + err.code, err.message, placement.sender, + ), ); - return Ok(()); - } - }; - match host.submit_order(chain_id, &body) { - Ok(server_uid) => { - // Persist under the server-supplied UID so downstream - // observers (cow-tooling, dune) join on the same key. The - // client UID we just computed should equal it; a Warn is - // worth a closer look if not (domain/owner divergence). - if server_uid != uid_hex { - host.log( - LogLevel::Warn, - &format!("ethflow uid drift: local={uid_hex} server={server_uid}"), - ); - } - host.set(&format!("submitted:{server_uid}"), b"")?; - // Clear any backoff: marker a prior transient error left - // behind; the terminal `submitted:` flag supersedes it. - let _ = host.delete(&format!("backoff:{server_uid}")); - host.log(LogLevel::Info, &format!("ethflow submitted {server_uid}")); } - Err(err) => apply_submit_retry(host, &err, &uid_hex)?, } Ok(()) } -/// Which terminal / transient marker (if any) the local store carries -/// for `uid_hex`. The submit path short-circuits on `Submitted` / -/// `Dropped`; `Backoff` still proceeds with a fresh attempt; `None` -/// means a clean first try. -#[derive(Debug, Eq, PartialEq)] -enum PriorOutcome { - None, - Submitted, - Backoff, - Dropped, -} - -fn prior_outcome(host: &H, uid_hex: &str) -> Result { - if host.get(&format!("submitted:{uid_hex}"))?.is_some() { - return Ok(PriorOutcome::Submitted); - } - if host.get(&format!("dropped:{uid_hex}"))?.is_some() { - return Ok(PriorOutcome::Dropped); - } - if host.get(&format!("backoff:{uid_hex}"))?.is_some() { - return Ok(PriorOutcome::Backoff); - } - Ok(PriorOutcome::None) -} - -fn apply_submit_retry(host: &H, err: &HostError, uid_hex: &str) -> Result<(), HostError> { - match classify_api_error(err.data.as_deref()) { - RetryAction::TryNextBlock | RetryAction::Backoff { .. } => { - host.set(&format!("backoff:{uid_hex}"), b"")?; - host.log( - LogLevel::Warn, - &format!("ethflow backoff {uid_hex} ({}): {}", err.code, err.message), - ); - } - RetryAction::Drop => { - host.set(&format!("dropped:{uid_hex}"), b"")?; - // Clear `backoff:` if a prior transient attempt left it - // behind - the terminal `dropped:` flag now supersedes - // it, and we want at most one outcome marker per UID at - // rest. - let _ = host.delete(&format!("backoff:{uid_hex}")); - host.log( - LogLevel::Warn, - &format!("ethflow dropped {uid_hex} ({}): {}", err.code, err.message), - ); - } - } - Ok(()) +/// 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. +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)) } #[cfg(test)] @@ -324,13 +213,13 @@ mod tests { use super::*; use alloy_primitives::{U256, address, hex}; use alloy_sol_types::SolValue; - use cowprotocol::{BuyTokenDestination, OrderKind, SellTokenSource}; - use shepherd_sdk::host::{HostErrorKind as Kind, LocalStoreHost as _}; + use cowprotocol::{BuyTokenDestination, OnchainSigningScheme, OrderKind, SellTokenSource}; + use shepherd_sdk::host::{HostError as SdkHostError, HostErrorKind, LocalStoreHost as _}; use shepherd_sdk_test::MockHost; const SEPOLIA: u64 = 11_155_111; - fn submittable_order() -> GPv2OrderData { + fn sample_order() -> GPv2OrderData { GPv2OrderData { sellToken: address!("6810e776880C02933D47DB1b9fc05908e5386b96"), buyToken: address!("DAE5F1590db13E3B40423B5b5c5fbf175515910b"), @@ -347,23 +236,10 @@ mod tests { } } - fn well_formed_placement() -> DecodedPlacement { - DecodedPlacement { - contract: ETH_FLOW_PRODUCTION, - sender: address!("00112233445566778899aabbccddeeff00112233"), - order: Box::new(submittable_order()), - signature: OnchainSignature { - scheme: OnchainSigningScheme::Eip1271, - data: hex!("c0ffeec0ffeec0ffee").to_vec().into(), - }, - data: Bytes::new(), - } - } - - fn sample_event_for_decode() -> OrderPlacement { + fn sample_event() -> OrderPlacement { OrderPlacement { sender: address!("00112233445566778899aabbccddeeff00112233"), - order: submittable_order(), + order: sample_order(), signature: OnchainSignature { scheme: OnchainSigningScheme::Eip1271, data: hex!("c0ffeec0ffeec0ffee").to_vec().into(), @@ -398,11 +274,18 @@ mod tests { } } - // ---- existing pure tests preserved from BLEU-832/833 ---- + fn computed_uid(placement: &DecodedPlacement) -> String { + format!( + "{}", + compute_uid(SEPOLIA, placement).expect("sepolia + canonical markers") + ) + } + + // ---- decode (BLEU-832 invariants preserved) ---- #[test] fn decodes_well_formed_placement() { - let event = sample_event_for_decode(); + let event = sample_event(); let (topics, data) = encode_log(&event); let decoded = decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data) .expect("decode succeeds"); @@ -413,365 +296,247 @@ mod tests { #[test] fn rejects_unrelated_contract_address() { - let event = sample_event_for_decode(); + let event = sample_event(); let (topics, data) = encode_log(&event); let stranger = address!("dead00000000000000000000000000000000dead"); assert!(decode_order_placement(stranger.as_slice(), &topics, &data).is_none()); } #[test] - fn build_eip1271_creation_has_contract_as_from() { - let placement = well_formed_placement(); - let (creation, uid) = build_eth_flow_creation( - 11_155_111, - &placement, - cowprotocol::EMPTY_APP_DATA_JSON.to_string(), - ) - .expect("build succeeds"); - assert_eq!(creation.from, placement.contract); - assert_eq!(creation.signing_scheme, cowprotocol::SigningScheme::Eip1271); - assert_eq!( - creation.signature.to_bytes(), - placement.signature.data.to_vec(), - ); - assert_eq!(&uid.as_slice()[32..52], placement.contract.as_slice()); - assert_eq!( - &uid.as_slice()[52..56], - &placement.order.validTo.to_be_bytes(), + fn rejects_wrong_topic_signature() { + let event = sample_event(); + let (_, data) = encode_log(&event); + let bad_topic = vec![0xaa_u8; 32]; + let sender_topic = vec![0u8; 32]; + assert!( + decode_order_placement( + ETH_FLOW_PRODUCTION.as_slice(), + &[bad_topic, sender_topic], + &data, + ) + .is_none() ); } - #[test] - fn build_presign_emits_presign_scheme() { - let mut placement = well_formed_placement(); - placement.signature = OnchainSignature { - scheme: OnchainSigningScheme::PreSign, - data: Bytes::new(), - }; - let (creation, _) = - build_eth_flow_creation(1, &placement, cowprotocol::EMPTY_APP_DATA_JSON.to_string()) - .expect("build succeeds"); - assert_eq!(creation.signing_scheme, cowprotocol::SigningScheme::PreSign); - assert!(creation.signature.to_bytes().is_empty()); - } + // ---- UID computation ---- #[test] - fn build_rejects_unsupported_chain() { - let placement = well_formed_placement(); - let err = build_eth_flow_creation( - 0xdead_beef, - &placement, - cowprotocol::EMPTY_APP_DATA_JSON.to_string(), - ) - .unwrap_err(); - assert!(matches!(err, BuildError::UnsupportedChain(0xdead_beef))); - } + fn compute_uid_pins_owner_to_ethflow_contract_and_validto() { + let event = sample_event(); + let (topics, data) = encode_log(&event); + let decoded = + decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); - #[test] - fn build_rejects_unknown_kind_marker() { - let mut placement = well_formed_placement(); - placement.order.kind = B256::repeat_byte(0x42); - let err = - build_eth_flow_creation(1, &placement, cowprotocol::EMPTY_APP_DATA_JSON.to_string()) - .unwrap_err(); - assert!(matches!(err, BuildError::UnknownMarker)); + let uid = compute_uid(SEPOLIA, &decoded).expect("sepolia + canonical markers"); + let bytes: [u8; 56] = uid.into(); + // owner suffix (bytes 32..52) = EthFlow contract address. + assert_eq!(&bytes[32..52], ETH_FLOW_PRODUCTION.as_slice()); + // valid_to suffix (bytes 52..56) = u32 BE of the on-chain validTo. + assert_eq!( + u32::from_be_bytes(bytes[52..56].try_into().unwrap()), + event.order.validTo, + ); } #[test] - fn build_rejects_non_empty_app_data() { - let mut placement = well_formed_placement(); - placement.order.appData = B256::repeat_byte(0xee); - let err = - build_eth_flow_creation(1, &placement, cowprotocol::EMPTY_APP_DATA_JSON.to_string()) - .unwrap_err(); - assert!(matches!(err, BuildError::Cowprotocol(_))); + fn compute_uid_returns_none_on_unsupported_chain() { + let event = sample_event(); + let (topics, data) = encode_log(&event); + let decoded = + decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); + assert!(compute_uid(9999, &decoded).is_none()); } - // ---- BLEU-855: MockHost dispatch tests ---- - - fn programmed_uid(placement: &DecodedPlacement) -> String { - let (_creation, uid) = build_eth_flow_creation( - SEPOLIA, - placement, - cowprotocol::EMPTY_APP_DATA_JSON.to_string(), - ) - .unwrap(); - format!("{uid}") - } + // ---- observe + verify dispatch (Host-trait integration) ---- + /// 200 from `GET /api/v1/orders/{uid}` → `observed:{uid}` written + /// + Info log + zero submit attempts. #[test] - fn placement_log_submits_order_and_persists_submitted_uid() { + fn placement_log_marks_observed_on_orderbook_200() { let host = MockHost::new(); - let event = sample_event_for_decode(); + let event = sample_event(); let (topics, data) = encode_log(&event); let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); let placement = decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); - let uid = programmed_uid(&placement); - host.cow_api.respond(Ok(uid.clone())); + 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()), + ); on_logs(&host, &[view]).unwrap(); - assert_eq!(host.cow_api.call_count(), 1); assert!( host.store .snapshot() - .contains_key(&format!("submitted:{uid}")) + .contains_key(&format!("observed:{uid}")), + "200 response must write observed:{{uid}} marker" ); - assert!( - !host - .store - .snapshot() - .contains_key(&format!("backoff:{uid}")) + assert_eq!( + host.cow_api.request_calls().len(), + 1, + "exactly one orderbook GET per log" ); - assert!(host.logging.contains(&format!("ethflow submitted {uid}"))); - } - - #[test] - fn redelivered_placement_is_skipped_via_submitted_uid_dedup() { - // BLEU-833 / commit c5e4d7d regression guard: a host - // reconnect or engine restart that replays the same - // OrderPlacement log must not double-submit. - let host = MockHost::new(); - let event = sample_event_for_decode(); - let (topics, data) = encode_log(&event); - let view1 = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - let placement = - decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); - let uid = programmed_uid(&placement); - host.cow_api.respond(Ok(uid.clone())); - - on_logs(&host, &[view1]).unwrap(); - assert_eq!(host.cow_api.call_count(), 1); - - let view2 = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - on_logs(&host, &[view2]).unwrap(); - assert_eq!( host.cow_api.call_count(), - 1, - "redelivered placement must not resubmit" + 0, + "observe path must never call submit_order" + ); + assert!( + host.logging + .contains(&format!("ethflow observed {uid} (orderbook indexed")) ); - assert!(host.logging.contains("already submitted")); } - /// COW-1074: an OrderPlacement carrying a non-empty `appData` - /// hash triggers a `cow_api_request` against - /// `/api/v1/app_data/{hex}`; the resolved JSON is passed to - /// `build_eth_flow_creation` so the digest matches and the - /// submit succeeds. Before this PR every non-empty placement - /// (cow-swap UI style) was rejected client-side with "app_data - /// JSON digest does not match signed app_data hash". + /// 404 from `GET /api/v1/orders/{uid}` → no marker written + Info + /// log + the next re-delivery rechecks (no early dedup). #[test] - fn placement_with_non_empty_app_data_resolves_then_submits() { - use alloy_primitives::keccak256; + fn placement_log_does_not_mark_observed_on_orderbook_404() { let host = MockHost::new(); - - let app_data_json = r#"{"version":"1.1.0","metadata":{"partnerId":"shepherd-e2e"}}"#; - let app_data_hash = keccak256(app_data_json.as_bytes()); - - // Build a placement event with the non-empty appData hash. - let mut event = sample_event_for_decode(); - event.order.appData = app_data_hash; + let event = sample_event(); let (topics, data) = encode_log(&event); let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); let placement = decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); - // Compute the UID against the resolved (non-empty) JSON so we - // can program cow_api.respond with the matching value. - let (_creation, uid_obj) = - build_eth_flow_creation(SEPOLIA, &placement, app_data_json.to_string()) - .expect("build with resolved app data"); - let uid = format!("{uid_obj}"); - host.cow_api.respond(Ok(uid.clone())); - - // Mirror the orderbook's /api/v1/app_data/{hex} response shape. - let envelope = format!( - r#"{{"fullAppData":{}}}"#, - serde_json::Value::String(app_data_json.to_string()), - ); - host.cow_api.respond_to_request_for( - "GET", - format!( - "/api/v1/app_data/0x{}", - alloy_primitives::hex::encode(app_data_hash) - ), - Ok(envelope), - ); + let uid = computed_uid(&placement); + + host.cow_api.respond_to_request(Err(SdkHostError { + domain: "cow-api".into(), + kind: HostErrorKind::Unavailable, + code: 404, + message: "Not Found".into(), + data: None, + })); on_logs(&host, &[view]).unwrap(); - assert_eq!( - host.cow_api.request_calls().len(), - 1, - "exactly one /app_data resolve" - ); - assert_eq!(host.cow_api.call_count(), 1, "exactly one orderbook submit"); assert!( - host.store + !host + .store .snapshot() - .contains_key(&format!("submitted:{uid}")), - "submitted:{{uid}} marker must be written after a successful resolve+submit" + .contains_key(&format!("observed:{uid}")), + "404 must NOT write observed: so re-delivery can recheck" + ); + let lines: Vec<_> = host + .logging + .lines() + .into_iter() + .filter(|l| l.message.contains("not yet indexed")) + .collect(); + assert_eq!(lines.len(), 1); + assert_eq!( + lines[0].level, + LogLevel::Info, + "indexer lag is expected; Info keeps soak dashboards quiet" ); - assert!(host.logging.contains(&format!("ethflow submitted {uid}"))); } - /// COW-1074: orderbook 404s the appData hash → strategy logs a - /// Warn and drops the placement (no submit attempt, no marker). + /// Non-404 error from the orderbook check → Warn log + no marker. #[test] - fn placement_skips_submit_when_app_data_hash_not_mirrored() { - use alloy_primitives::keccak256; + fn placement_log_warns_on_orderbook_other_error() { let host = MockHost::new(); - - let mut event = sample_event_for_decode(); - event.order.appData = keccak256(b"unknown-document"); + let event = sample_event(); let (topics, data) = encode_log(&event); let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - host.cow_api - .respond_to_request(Err(shepherd_sdk::host::HostError { - domain: "cow-api".into(), - kind: shepherd_sdk::host::HostErrorKind::Unavailable, - code: 404, - message: "Not Found".into(), - data: None, - })); + host.cow_api.respond_to_request(Err(SdkHostError { + domain: "cow-api".into(), + kind: HostErrorKind::Internal, + code: 502, + message: "bad gateway".into(), + data: None, + })); on_logs(&host, &[view]).unwrap(); - assert_eq!(host.cow_api.call_count(), 0, "no submit attempt on 404"); - let store = host.store.snapshot(); - assert!(!store.keys().any(|k| k.starts_with("submitted:"))); - assert!(!store.keys().any(|k| k.starts_with("dropped:"))); - assert!(host.logging.contains("appData hash not mirrored")); + assert!( + host.store.snapshot().is_empty(), + "non-404 error must not write any marker" + ); + let lines: Vec<_> = host + .logging + .lines() + .into_iter() + .filter(|l| l.message.contains("indexer check failed")) + .collect(); + assert_eq!(lines.len(), 1); + assert_eq!(lines[0].level, LogLevel::Warn); } + /// Idempotency: a placement that already has `observed:{uid}` in + /// local store does NOT trigger a fresh GET on re-delivery. #[test] - fn submit_transient_error_writes_backoff_marker_and_returns() { + fn previously_observed_placement_is_skipped_on_redelivery() { let host = MockHost::new(); - let event = sample_event_for_decode(); + let event = sample_event(); let (topics, data) = encode_log(&event); let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); let placement = decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); - let uid = programmed_uid(&placement); - - // InsufficientFee classifies as TryNextBlock per cowprotocol's - // retry_hint; ethflow-watcher treats every retriable - // classification as a backoff: marker (next event will retry, - // not next block). - let api_body = serde_json::json!({ - "errorType": "InsufficientFee", - "description": "fee too low", - }) - .to_string(); - host.cow_api.respond(Err(HostError { - domain: "cow-api".into(), - kind: Kind::Denied, - code: 400, - message: "InsufficientFee".into(), - data: Some(api_body), - })); + let uid = computed_uid(&placement); + + host.store + .set(&format!("observed:{uid}"), b"") + .expect("seed observed marker"); on_logs(&host, &[view]).unwrap(); - assert!( - host.store - .snapshot() - .contains_key(&format!("backoff:{uid}")) - ); - assert!( - !host - .store - .snapshot() - .contains_key(&format!("submitted:{uid}")) + assert_eq!( + host.cow_api.request_calls().len(), + 0, + "observed:{{uid}} must short-circuit before the orderbook GET" ); - assert!( - !host - .store - .snapshot() - .contains_key(&format!("dropped:{uid}")) + assert_eq!( + host.cow_api.call_count(), + 0, + "and certainly no submit_order" ); - assert!(host.logging.contains("ethflow backoff")); } + /// Defensive: unsupported chain id surfaces a Warn but does not + /// panic and does not touch the orderbook. #[test] - fn submit_permanent_error_persists_dropped_uid_and_clears_backoff() { + fn unsupported_chain_logs_warn_without_orderbook_call() { let host = MockHost::new(); - let event = sample_event_for_decode(); + let event = sample_event(); let (topics, data) = encode_log(&event); - let placement = - decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data).unwrap(); - let uid = programmed_uid(&placement); - - // Pre-seed a backoff: marker (prior transient attempt). A - // permanent failure on the retry must drop the order AND - // clear the stale backoff: row so we never have both at rest. - host.store.set(&format!("backoff:{uid}"), b"").unwrap(); - - let api_body = serde_json::json!({ - "errorType": "InvalidSignature", - "description": "bad sig", - }) - .to_string(); - host.cow_api.respond(Err(HostError { - domain: "cow-api".into(), - kind: Kind::Denied, - code: 400, - message: "InvalidSignature".into(), - data: Some(api_body), - })); + let view = LogView { + chain_id: 9999, // not in cowprotocol::Chain + address: ETH_FLOW_PRODUCTION.as_slice(), + topics: &topics, + data: &data, + }; - let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); on_logs(&host, &[view]).unwrap(); - assert!( - host.store - .snapshot() - .contains_key(&format!("dropped:{uid}")) - ); - assert!( - !host - .store - .snapshot() - .contains_key(&format!("backoff:{uid}")), - "terminal `dropped:` must clear stale `backoff:` marker" - ); - assert!(host.logging.contains("ethflow dropped")); + assert_eq!(host.cow_api.request_calls().len(), 0); + assert_eq!(host.cow_api.call_count(), 0); + assert!(host.logging.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. #[test] - fn eip1271_signature_shape_round_trips_through_submit_body() { - // Snapshot the JSON the host receives so reviewers can confirm - // the signing scheme / signature wire shape stays stable. The - // orderbook is strict about both fields. + fn strategy_never_calls_submit_order() { let host = MockHost::new(); - let event = sample_event_for_decode(); + let event = sample_event(); let (topics, data) = encode_log(&event); let view = placement_log_view(ETH_FLOW_PRODUCTION.as_slice(), &topics, &data); - host.cow_api.respond(Ok("0xfeedface".to_string())); + host.cow_api.respond_to_request(Ok("{}".to_string())); on_logs(&host, &[view]).unwrap(); - let body_json = host - .cow_api - .last_body_as_json() - .expect("body was submitted"); - // OrderCreation serialises signingScheme as a lowercase string - // and signature as a hex-prefixed bytes blob. - assert_eq!(body_json["signingScheme"].as_str(), Some("eip1271")); - let sig_hex = body_json["signature"] - .as_str() - .expect("signature is a string"); - assert!(sig_hex.starts_with("0x")); assert_eq!( - sig_hex, "0xc0ffeec0ffeec0ffee", - "EIP-1271 signature blob must be passed through verbatim" - ); - // EthFlow contract is the orderbook `from`, not the original sender. - assert_eq!( - body_json["from"].as_str(), - Some(&*format!("{:#x}", ETH_FLOW_PRODUCTION)) + host.cow_api.call_count(), + 0, + "submit_order count must stay at zero — ethflow-watcher is observer-only" ); } }