From 0d5ff1a9d66b4a84487f1543483e132d9a3601e3 Mon Sep 17 00:00:00 2001 From: brunota20 Date: Mon, 22 Jun 2026 21:45:10 -0300 Subject: [PATCH] feat(ethflow-watcher): observe + verify orderbook indexing (BLEU-833 redesign, COW-1076) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The original BLEU-833 design lifted the on-chain GPv2OrderData into an OrderCreation body and POSTed it to /api/v1/orders with the EthFlow contract as the EIP-1271 owner. Empirical evidence collected during a 2026-06-22 Sepolia soak shows that path cannot succeed: - The 3 OrderPlacement events the watcher observed during the run (creationDate 18:43, 18:53, 21:19 UTC, UIDs ending in ba3cb449...EAdec|ffffffff) are *fulfilled* on the orderbook with validTo = u32::MAX — the same shape the orderbook rejects when shepherd POSTs it. - The orderbook entries carry onchainUser, onchainOrderData and ethflowData.userValidTo (a real unix timestamp), and these fields do not exist on the OrderCreation request body in cowprotocol-orderbook. They are populated by the orderbook backend itself, which indexes the OrderPlacement event server-side and creates the order with dual-validTo bookkeeping. So /api/v1/orders is structurally the wrong endpoint for on-chain EthFlow placements: the orderbook's native indexer is the one that creates the order entry, and the public POST endpoint applies the generic validTo cap and rejects with ExcessiveValidTo every time. Continuing to submit produces only garbage outcomes per real EthFlow trade (misleading dropped/backoff markers and a misleading shepherd_cow_api_submit_total{outcome="err"} increment). This commit replaces the submit path with the contract the orderbook actually offers: shepherd decodes the OrderPlacement event, computes the canonical UID, GETs /api/v1/orders/{uid}, and records the outcome: - 200 → write observed:{uid}, log Info ("orderbook indexed"). - 404 → log Info ("not yet indexed; will recheck on re-delivery") and do not write the marker, so a future log re-delivery or a block-tick poll (out of scope here) can re-verify. - other err → log Warn with the host's code and message. Idempotency stays event-driven: once observed:{uid} is written, log re-delivery is a no-op. Block-tick-driven retries for stuck indexer lag are intentionally out of scope and tracked separately. Test surface: the existing BLEU-832 decode tests are preserved verbatim and the GPv2OrderData fixture is reused with the canonical EthFlow markers (OrderKind::SELL, SellTokenSource::ERC20, BuyTokenDestination::ERC20). Two new tests cover UID computation invariants: the owner suffix pins to the EthFlow contract and the validTo suffix pins to the on-chain order's validTo. A third asserts compute_uid returns None on unsupported chains so callers fall back to a Warn log rather than panic. This invalidates the design that PRs #10 (BLEU-833), #47 (COW-1074), #48 (COW-1075) and #49 (COW-1076) were all patching. After review, those PRs should be closed or substantially rewritten — none of the appData resolution / error-envelope forwarding / log-downgrade work is needed once the broken endpoint is no longer called. AI assistance disclosure: drafted by Claude (Opus 4.7, 1M context) following the soak inspection that surfaced the divergence. --- modules/ethflow-watcher/src/lib.rs | 213 ++++++++++++++++++++++++----- 1 file changed, 182 insertions(+), 31 deletions(-) diff --git a/modules/ethflow-watcher/src/lib.rs b/modules/ethflow-watcher/src/lib.rs index f042c154..98a94c46 100644 --- a/modules/ethflow-watcher/src/lib.rs +++ b/modules/ethflow-watcher/src/lib.rs @@ -11,20 +11,31 @@ wit_bindgen::generate!({ use alloy_primitives::{Address, B256, Bytes}; use alloy_sol_types::SolEvent; use cowprotocol::{ - CoWSwapOnchainOrders::OrderPlacement, ETH_FLOW_PRODUCTION, ETH_FLOW_STAGING, GPv2OrderData, - OnchainSignature, + BuyTokenDestination, Chain, CoWSwapOnchainOrders::OrderPlacement, ETH_FLOW_PRODUCTION, + ETH_FLOW_STAGING, GPv2OrderData, OnchainSignature, OrderData, OrderKind, OrderUid, + SellTokenSource, }; -use nexum::host::{logging, types}; +use nexum::host::{local_store, logging, types}; +use shepherd::cow::cow_api; -/// Fully decoded payload of a `CoWSwapOnchainOrders.OrderPlacement` log. -/// `GPv2OrderData` is ~300 bytes; box it so the struct stays cache- -/// friendly when it later lands in the BLEU-833 submission path. +/// Decoded payload of a `CoWSwapOnchainOrders.OrderPlacement` log. +/// `GPv2OrderData` is ~300 bytes; box it to keep the struct +/// cache-friendly when threaded through the observe path. #[derive(Debug)] -#[allow(dead_code)] // Fields consumed by BLEU-833. struct DecodedPlacement { + /// EthFlow contract that emitted the event. EIP-1271 owner of + /// the resulting orderbook entry — used as the UID `owner` input. + contract: Address, + /// Native-token seller that called `createOrder`. Logged for + /// operator diagnostics; not the orderbook owner. sender: Address, order: Box, + /// Refund pointer / opaque placer metadata. Recorded by the + /// orderbook indexer in `ethflowData.userValidTo`; not consumed + /// by this module. + #[allow(dead_code)] signature: OnchainSignature, + #[allow(dead_code)] data: Bytes, } @@ -39,11 +50,10 @@ impl Guest for EthFlowWatcher { fn on_event(event: types::Event) -> Result<(), HostError> { if let types::Event::Logs(logs) = event { for log in &logs { - if let Some(placement) = decode_order_placement(&log.address, &log.topics, &log.data) + if let Some(placement) = + decode_order_placement(&log.address, &log.topics, &log.data) { - log_placement(&placement); - // BLEU-833 will build OrderCreation + submit + apply - // OrderPostError::retry_hint right here. + observe_placement(log.chain_id, &placement)?; } } } @@ -52,8 +62,9 @@ impl Guest for EthFlowWatcher { } } -/// Decode a raw event log against `CoWSwapOnchainOrders.OrderPlacement`, -/// keeping the four fields the BLEU-833 submission path needs. +// ---- decode (BLEU-832) ---- + +/// Decode a raw event log against `CoWSwapOnchainOrders.OrderPlacement`. /// /// Returns `None` when: /// - the log's contract address is not one of the canonical `ETH_FLOW_*` @@ -62,9 +73,6 @@ impl Guest for EthFlowWatcher { /// leak through); /// - topic0 does not match the event signature; or /// - the ABI body fails to decode (truncated, wrong layout). -/// -/// Kept on plain slices so the host-free unit tests can call it without -/// wit-bindgen scaffolding. fn decode_order_placement( address: &[u8], topics: &[Vec], @@ -88,6 +96,7 @@ fn decode_order_placement( .collect(); let decoded = OrderPlacement::decode_raw_log(words, data).ok()?; Some(DecodedPlacement { + contract, sender: decoded.sender, order: Box::new(decoded.order), signature: decoded.signature, @@ -95,18 +104,108 @@ fn decode_order_placement( }) } -fn log_placement(p: &DecodedPlacement) { - logging::log( - logging::Level::Info, - &format!( - "ethflow OrderPlacement sender={:#x} sell={:#x} buy={:#x} valid_to={} sig_scheme={:?}", - p.sender, - p.order.sellToken, - p.order.buyToken, - p.order.validTo, - p.signature.scheme, - ), - ); +// ---- observe + verify (BLEU-833 revised — COW-1076) ---- + +/// Compute the orderbook UID for the placement and confirm the +/// orderbook's native EthFlow indexer picked it up. +/// +/// Background (COW-1076): the orderbook backend indexes +/// `OrderPlacement` events server-side and creates the order entry +/// with its dual-validTo bookkeeping (on-chain `validTo = u32::MAX` +/// for chain compatibility, off-chain `ethflowData.userValidTo` +/// derived from the embedded placer payload). The public +/// `POST /api/v1/orders` endpoint applies the generic validity cap +/// and rejects EthFlow shapes with `ExcessiveValidTo`; there is no +/// path through it that produces the same result as the native +/// indexer. This module therefore observes the event and verifies +/// the indexer caught it, instead of submitting. +fn observe_placement(chain_id: u64, placement: &DecodedPlacement) -> Result<(), HostError> { + let uid_hex = match compute_uid(chain_id, placement) { + Some(uid) => format!("{uid}"), + None => { + logging::log( + logging::Level::Warn, + &format!( + "ethflow uid build skipped (sender={:#x}): unsupported chain {chain_id} or unknown order marker", + placement.sender, + ), + ); + return Ok(()); + } + }; + + // Idempotency: once verified, do not re-check on log re-delivery + // (engine restart, reorg replay, supervisor restart). + if local_store::get(&format!("observed:{uid_hex}"))?.is_some() { + return Ok(()); + } + + let path = format!("/api/v1/orders/{uid_hex}"); + match cow_api::request(chain_id, "GET", &path, None) { + Ok(_) => { + local_store::set(&format!("observed:{uid_hex}"), &[])?; + logging::log( + logging::Level::Info, + &format!( + "ethflow observed {uid_hex} (orderbook indexed, sender={:#x})", + placement.sender, + ), + ); + } + 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 `observed:` so the next log re-delivery (or a + // future block-tick poll) can re-check; log at Info to keep + // the soak dashboard quiet on normal lag. + logging::log( + logging::Level::Info, + &format!( + "ethflow not yet indexed {uid_hex} (sender={:#x}); will recheck on re-delivery", + placement.sender, + ), + ); + } + Err(err) => { + logging::log( + logging::Level::Warn, + &format!( + "ethflow indexer check failed {uid_hex} ({}): {} (sender={:#x})", + err.code, err.message, placement.sender, + ), + ); + } + } + Ok(()) +} + +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)) +} + +/// Lift the on-chain `GPv2OrderData` enum-byte fields into the typed +/// `OrderData` the UID computation needs. Returns `None` if any of +/// the enum bytes carries an unrecognised marker (defensive — every +/// production EthFlow placement uses the canonical markers). +fn gpv2_to_order_data(gpv2: &GPv2OrderData) -> Option { + Some(OrderData { + sell_token: gpv2.sellToken, + buy_token: gpv2.buyToken, + receiver: (gpv2.receiver != Address::ZERO).then_some(gpv2.receiver), + sell_amount: gpv2.sellAmount, + buy_amount: gpv2.buyAmount, + valid_to: gpv2.validTo, + app_data: gpv2.appData, + fee_amount: gpv2.feeAmount, + kind: OrderKind::from_contract_bytes(gpv2.kind)?, + partially_fillable: gpv2.partiallyFillable, + sell_token_balance: SellTokenSource::from_contract_bytes(gpv2.sellTokenBalance)?, + buy_token_balance: BuyTokenDestination::from_contract_bytes(gpv2.buyTokenBalance)?, + }) } export!(EthFlowWatcher); @@ -128,10 +227,13 @@ mod tests { validTo: 1_700_000_000, appData: B256::repeat_byte(0xaa), feeAmount: U256::ZERO, - kind: B256::repeat_byte(0xbb), + // Canonical sell-side / ERC20 markers — cowprotocol exposes + // them as B256 associated constants matching the on-chain + // `keccak256("sell")` / `keccak256("erc20")` shapes. + kind: OrderKind::SELL, partiallyFillable: false, - sellTokenBalance: B256::repeat_byte(0xcc), - buyTokenBalance: B256::repeat_byte(0xdd), + sellTokenBalance: SellTokenSource::ERC20, + buyTokenBalance: BuyTokenDestination::ERC20, } } @@ -165,6 +267,8 @@ mod tests { (topics, data) } + // ---- decode tests (BLEU-832) ---- + #[test] fn decodes_well_formed_placement() { let (sender, event) = sample_event(); @@ -172,6 +276,7 @@ mod tests { let address = ETH_FLOW_PRODUCTION.as_slice(); let decoded = decode_order_placement(address, &topics, &data).expect("decode succeeds"); + assert_eq!(decoded.contract, ETH_FLOW_PRODUCTION); assert_eq!(decoded.sender, sender); assert_eq!(decoded.order.sellToken, event.order.sellToken); assert_eq!(decoded.order.buyAmount, event.order.buyAmount); @@ -232,4 +337,50 @@ mod tests { let (_, data) = encode_log(&sample_event().1); assert!(decode_order_placement(ETH_FLOW_PRODUCTION.as_slice(), &[], &data).is_none()); } + + // ---- UID computation (observe path) ---- + + /// Sanity check that `compute_uid` produces the canonical 56-byte + /// shape `digest || owner || valid_to`. We do not pin the exact UID + /// (it depends on `Chain::SEPOLIA::settlement_domain()` and would + /// drift if cowprotocol's domain separator changes); instead we + /// assert structural invariants the orderbook joins on. + #[test] + 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(); + + // 11155111 = Sepolia chain ID; supported by cowprotocol::Chain. + let uid = compute_uid(11_155_111, &decoded).expect("known chain + 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 big-endian of the order's + // on-chain validTo. EthFlow uses u32::MAX in production; this + // sample uses 1_700_000_000. + assert_eq!( + u32::from_be_bytes(bytes[52..56].try_into().unwrap()), + event.order.validTo, + ); + } + + #[test] + 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(); + // Chain id 9999 is not in `cowprotocol::Chain`. + assert!(compute_uid(9999, &decoded).is_none()); + } + + #[test] + fn gpv2_to_order_data_rejects_unknown_kind_marker() { + let mut order = sample_order(); + order.kind = B256::repeat_byte(0xff); + assert!(gpv2_to_order_data(&order).is_none()); + } }