M2 epic: TWAP + EthFlow modules + module.toml manifests

Cargo.toml

@@ -1,7 +1,9 @@
 [workspace]
 members = [
     "crates/nexum-engine",
+    "modules/ethflow-watcher",
     "modules/example",
+    "modules/twap-monitor",
 ]
 resolver = "2"
 
@@ -100,15 +102,17 @@ todo = "deny"
 
 # `cowprotocol` v1.0.0-alpha.3 (the crates.io release the engine
 # depends on) was cut from `cowdao-grants/cow-rs` PR #5 at commit
-# `1742ffa`. `bleu/cow-rs` main has 18 commits since, including the
+# `1742ffa`. `bleu/cow-rs` main has diverged since with: the
 # `composable::Proof` width fix (relevant to the TWAP poll path),
-# `OrderCreation` zero-from-address fast-fail (closes a MEDIUM
-# review finding from PR #5), and the `order_book` / `composable`
-# submodule splits. Patching to that commit picks them up without
-# waiting for an alpha.4 publish. Drop once `cowprotocol >= 1.0.0-alpha.4`
-# ships.
+# `OrderCreation` zero-from-address fast-fail, the `order_book` /
+# `composable` submodule splits, `OrderPostErrorKind` + `retry_hint()`
+# (BLEU-822, the protocol-level retry contract M2 modules dispatch
+# on), and `OrderBookApi::with_base_url(chain, base_url)` for barn /
+# staging routing (BLEU-823). Patching to that commit picks the lot
+# up without waiting for an alpha.4 publish. Drop once
+# `cowprotocol >= 1.0.0-alpha.4` ships.
 [patch.crates-io]
-cowprotocol = { git = "https://github.com/bleu/cow-rs", rev = "c012404ffefc411bff543d2290e19ba7fbef2516" }
+cowprotocol = { git = "https://github.com/bleu/cow-rs", rev = "57f5f553ab28c9fff54089daf2d39b4282f3e4dd" }
 
 [profile.dev]
 panic = "abort"

README.md

@@ -104,7 +104,7 @@ allow = ["api.cow.fi"]
 [[subscription]]
 kind     = "log"
 chain_id = 1
-address  = "0xC92E8bdf79f0507f65a392b0ab4667716BFE0110"  # ComposableCoW
+address  = "0xfdaFc9d1902f4e0b84f65F49f244b32b31013b74"  # ComposableCoW (canonical CREATE2 address, same on every supported chain)
 
 [[subscription]]
 kind     = "block"

crates/nexum-engine/Cargo.toml

@@ -16,6 +16,15 @@ wasmtime-wasi = "45"
 # Async + error plumbing.
 anyhow.workspace = true
 thiserror.workspace = true
+# `strum::IntoStaticStr` on error enums gives metric labels (`error_kind`)
+# free via a snake_case `&'static str` for every variant. Used at
+# `tracing::warn!(error_kind = <variant_name>.into(), ...)` sites and
+# any `metrics::counter!(... "error_kind" => kind)` recordings, so the
+# Prometheus labels stay in lock-step with the Rust enum source of
+# truth instead of needing a `match err { ... => "connect" ... }`
+# ladder per call site. Pinned via the workspace so every consumer
+# moves in lockstep.
+strum.workspace = true
 tokio.workspace = true
 clap.workspace = true
 

crates/nexum-engine/src/bindings.rs

@@ -5,7 +5,7 @@
 //! natively - no vendored `deps/` tree needed. The world name is fully
 //! qualified.
 //!
-//! Every `Host` trait impl in [`crate::host::impls`] consumes types
+//! Every `Host` trait impl in `crate::host::impls` consumes types
 //! generated here.
 
 wasmtime::component::bindgen!({

crates/nexum-engine/src/engine_config.rs

@@ -20,8 +20,32 @@ use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
 
 use serde::Deserialize;
+use strum::IntoStaticStr;
+use thiserror::Error;
 use tracing::{info, warn};
 
+/// Errors surfaced by [`load_or_default`].
+///
+/// Library-side modules must not propagate `anyhow::Error`; the rust
+/// idiomatic rubric reserves `anyhow` for `main.rs` and
+/// `supervisor.rs` top-level dispatch. The variants carry the
+/// upstream error via `#[from]` so the caller in `main.rs` (which
+/// uses `anyhow`) gets a free conversion through `?`.
+///
+/// `IntoStaticStr` exposes the snake_case variant name for metric
+/// labels and structured-log `error_kind` fields.
+#[derive(Debug, Error, IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
+#[non_exhaustive]
+pub enum EngineConfigError {
+    /// Failed to read the config file from disk.
+    #[error("read engine config: {0}")]
+    Io(#[from] std::io::Error),
+    /// Config file was unparseable as TOML.
+    #[error("parse engine config: {0}")]
+    Toml(#[from] toml::de::Error),
+}
+
 /// Engine-side configuration loaded from `engine.toml`.
 #[derive(Debug, Default, Deserialize)]
 pub struct EngineConfig {
@@ -130,8 +154,8 @@ fn default_log_level() -> String {
 }
 
 /// Read an engine config from disk, returning defaults if the file is
-/// missing. Parse errors propagate.
-pub fn load_or_default(path: Option<&Path>) -> anyhow::Result<EngineConfig> {
+/// missing. Parse errors propagate via [`EngineConfigError`].
+pub fn load_or_default(path: Option<&Path>) -> Result<EngineConfig, EngineConfigError> {
     let path = match path {
         Some(p) => p.to_path_buf(),
         None => PathBuf::from("engine.toml"),

crates/nexum-engine/src/host/cow_orderbook.rs

@@ -19,6 +19,7 @@
 use std::collections::BTreeMap;
 
 use cowprotocol::{Chain, OrderBookApi, OrderCreation, OrderUid};
+use strum::IntoStaticStr;
 use thiserror::Error;
 
 /// Process-wide pool of `OrderBookApi` clients keyed by EVM chain id.
@@ -120,7 +121,14 @@ impl OrderBookPool {
     }
 }
 
-#[derive(Debug, Error)]
+/// `IntoStaticStr` exposes the snake_case variant name as a
+/// `&'static str` (`"unknown_chain"`, `"bad_method"`, ...) so the
+/// `shepherd_cow_api_*` metric labels and structured-log fields stay
+/// in sync with the Rust source of truth instead of growing a
+/// `match err { ... => "decode" ... }` ladder per call site.
+#[derive(Debug, Error, IntoStaticStr)]
+#[strum(serialize_all = "snake_case")]
+#[non_exhaustive]
 pub enum CowApiError {
     #[error("unknown chain {0} (no cowprotocol::Chain variant)")]
     UnknownChain(u64),

crates/nexum-engine/src/host/cow_orderbook/tests.rs

@@ -206,3 +206,85 @@ fn sample_order_json() -> String {
     .expect("valid OrderCreation");
     serde_json::to_string(&creation).expect("serialise OrderCreation")
 }
+
+#[tokio::test]
+async fn request_rejects_malformed_path() {
+    // `Url::join` is very lenient for valid UTF-8 inputs.  The
+    // `BadPath` variant fires only when `Url::join` returns a parse
+    // error, which is hard to provoke.  Using a bare scheme-like
+    // string (`"://not-a-path"`) is NOT rejected because after
+    // stripping the leading `/` it is treated as a relative path
+    // component.  Instead, feed a string that *will* reach the
+    // network but is handled by wiremock with a 404, confirming the
+    // passthrough returns Ok even for nonsensical paths.
+    let mock = MockServer::start().await;
+    let pool = pool_with_mainnet_at(&mock);
+    // wiremock returns 404 for any un-mocked route — the response
+    // body is still surfaced to the caller.
+    let result = pool
+        .request(Chain::Mainnet.id(), "GET", "://not-a-path", None)
+        .await;
+    assert!(
+        result.is_ok(),
+        "Url::join treats this as a relative path, so no BadPath error"
+    );
+}
+
+#[tokio::test]
+async fn request_network_error_on_dead_server() {
+    // Build the pool against a port that no one is listening on.
+    // We use port 1 (TCP echo / privileged) which is never bound
+    // by user-space processes, guaranteeing a connection-refused.
+    let mut clients = std::collections::BTreeMap::new();
+    clients.insert(
+        Chain::Mainnet.id(),
+        OrderBookApi::new_with_base_url("http://127.0.0.1:1/".parse().expect("valid url")),
+    );
+    let pool = OrderBookPool {
+        clients,
+        http: reqwest::Client::new(),
+    };
+    let err = pool
+        .request(Chain::Mainnet.id(), "GET", "/api/v1/version", None)
+        .await
+        .unwrap_err();
+    assert!(matches!(err, CowApiError::Network(_)));
+}
+
+#[tokio::test]
+async fn request_5xx_response_is_returned_verbatim() {
+    let mock = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/api/v1/health"))
+        .respond_with(ResponseTemplate::new(500).set_body_string(r#"{"error":"internal"}"#))
+        .expect(1)
+        .mount(&mock)
+        .await;
+
+    let pool = pool_with_mainnet_at(&mock);
+    let body = pool
+        .request(Chain::Mainnet.id(), "GET", "/api/v1/health", None)
+        .await
+        .expect("5xx body is returned, not an Err");
+    assert_eq!(body, r#"{"error":"internal"}"#);
+}
+
+#[tokio::test]
+async fn submit_order_rejects_invalid_json() {
+    let pool = OrderBookPool::default();
+    let err = pool
+        .submit_order_json(Chain::Mainnet.id(), b"not json")
+        .await
+        .unwrap_err();
+    assert!(matches!(err, CowApiError::Decode(_)));
+}
+
+#[tokio::test]
+async fn submit_order_rejects_wrong_schema() {
+    let pool = OrderBookPool::default();
+    let err = pool
+        .submit_order_json(Chain::Mainnet.id(), br#"{"valid":"json"}"#)
+        .await
+        .unwrap_err();
+    assert!(matches!(err, CowApiError::Decode(_)));
+}

crates/nexum-engine/src/host/impls/cow_api.rs

@@ -58,10 +58,7 @@ impl shepherd::cow::cow_api::Host for HostState {
         let start = Instant::now();
         tracing::debug!(chain_id, bytes = order_data.len(), "cow-api::submit-order");
         let result = match self.cow.submit_order_json(chain_id, &order_data).await {
-            Ok(uid) => Ok(format!(
-                "0x{}",
-                alloy_primitives::hex::encode(uid.as_slice())
-            )),
+            Ok(uid) => Ok(alloy_primitives::hex::encode_prefixed(uid.as_slice())),
             Err(CowApiError::UnknownChain(id)) => Err(unimplemented(
                 "cow-api",
                 format!("chain {id} not in cowprotocol"),

crates/nexum-engine/src/host/impls/local_store.rs

@@ -3,30 +3,32 @@
 use crate::bindings::HostError;
 use crate::bindings::nexum;
 use crate::host::error::internal_error;
+use crate::host::local_store_redb::StorageError;
 use crate::host::state::HostState;
 
+/// Shared `StorageError` -> `HostError` conversion used by every
+/// `local-store` host endpoint. Centralised so the `("local-store",
+/// err.to_string())` shape stays consistent and a future error-model
+/// change (richer kind, structured `data`) lands in one place
+/// instead of four call sites.
+fn local_store_err(err: StorageError) -> HostError {
+    internal_error("local-store", err.to_string())
+}
+
 impl nexum::host::local_store::Host for HostState {
     async fn get(&mut self, key: String) -> Result<Option<Vec<u8>>, HostError> {
-        self.store
-            .get(&key)
-            .map_err(|err| internal_error("local-store", err.to_string()))
+        self.store.get(&key).map_err(local_store_err)
     }
 
     async fn set(&mut self, key: String, value: Vec<u8>) -> Result<(), HostError> {
-        self.store
-            .set(&key, &value)
-            .map_err(|err| internal_error("local-store", err.to_string()))
+        self.store.set(&key, &value).map_err(local_store_err)
     }
 
     async fn delete(&mut self, key: String) -> Result<(), HostError> {
-        self.store
-            .delete(&key)
-            .map_err(|err| internal_error("local-store", err.to_string()))
+        self.store.delete(&key).map_err(local_store_err)
     }
 
     async fn list_keys(&mut self, prefix: String) -> Result<Vec<String>, HostError> {
-        self.store
-            .list_keys(&prefix)
-            .map_err(|err| internal_error("local-store", err.to_string()))
+        self.store.list_keys(&prefix).map_err(local_store_err)
     }
 }