[review-mirror] M4 epic (upstream nullislabs/shepherd#20)

.gitignore

@@ -29,3 +29,14 @@ skills-lock.json
 
 # Engine runtime state (default state_dir from engine.toml).
 data/
+
+# E2E automation: rendered configs with embedded RPC keys + script state
+# never get committed.
+*.local.toml
+scripts/.state
+scripts/.env
+
+# Generated reports under e2e-reports/ (operator commits the filled-in ones
+# manually via `git add -f`).
+docs/operations/e2e-reports/engine-*.log
+docs/operations/e2e-reports/metrics-*.txt

Cargo.toml

@@ -8,7 +8,12 @@ members = [
     "modules/examples/balance-tracker",
     "modules/examples/price-alert",
     "modules/examples/stop-loss",
+    "modules/fixtures/flaky-bomb",
+    "modules/fixtures/fuel-bomb",
+    "modules/fixtures/memory-bomb",
     "modules/twap-monitor",
+    "tools/load-gen",
+    "tools/orderbook-mock",
 ]
 resolver = "2"
 

crates/nexum-engine/Cargo.toml

@@ -36,7 +36,13 @@ serde_json.workspace = true
 # Observability. `tracing` replaces the prior `eprintln!` debug log
 # so the engine can drop into a structured log pipeline in production.
 tracing.workspace = true
-tracing-subscriber = { workspace = true, default-features = false, features = ["fmt", "env-filter", "ansi"] }
+tracing-subscriber.workspace = true
+
+# Prometheus exporter (COW-1034). `metrics` is the facade every
+# recording site (dispatch, host backends) calls; the exporter
+# crate installs the recorder + binds the `/metrics` HTTP listener.
+metrics = "0.24"
+metrics-exporter-prometheus = { version = "0.17", default-features = false, features = ["http-listener"] }
 
 # `cow-api` backend. cowprotocol pulls `OrderBookApi`, `OrderCreation`,
 # `OrderUid`, the orderbook base URL table per `Chain`, and the typed

crates/nexum-engine/src/cli.rs

@@ -13,7 +13,17 @@ use clap::Parser;
 
 /// Parsed CLI surface.
 ///
-/// `nexum-engine [<wasm-path> [<manifest-path>]] [--engine-config <path>]`
+/// `nexum-engine [<wasm-path> [<manifest-path>]] [--engine-config <path>] [--pretty-logs]`
+///
+/// Positional `<wasm-path>` is a backwards-compat shortcut that
+/// synthesises a one-module engine config. Production deployments pass
+/// `--engine-config` and declare modules in TOML.
+///
+/// `--pretty-logs` selects the human-readable tracing formatter (the
+/// historical 0.1 default). Without the flag the engine emits JSON
+/// log lines per the COW-1035 structured-logging contract: a single
+/// `jq` / Loki / Grafana stream reconstructs the full timeline of
+/// any dispatch, host call, or order submission.
 #[derive(Parser, Debug, Default)]
 #[command(
     name = "nexum-engine",
@@ -35,4 +45,9 @@ pub struct Cli {
     /// documented in `engine_config::load_or_default`.
     #[arg(long = "engine-config")]
     pub engine_config: Option<PathBuf>,
+
+    /// Use the human-readable tracing formatter instead of the
+    /// default JSON formatter (COW-1035 structured-logging contract).
+    #[arg(long = "pretty-logs")]
+    pub pretty_logs: bool,
 }

crates/nexum-engine/src/engine_config.rs

@@ -87,23 +87,63 @@ pub struct EngineSection {
     /// `info` when absent; `RUST_LOG` overrides at process start.
     #[serde(default = "default_log_level")]
     pub log_level: String,
+    /// Prometheus metrics exporter wiring (COW-1034). Absent table =
+    /// disabled (the engine still installs the recorder so call sites
+    /// stay live but no HTTP listener binds).
+    #[serde(default)]
+    pub metrics: MetricsSection,
 }
 
 impl Default for EngineSection {
     fn default() -> Self {
         Self {
             state_dir: default_state_dir(),
             log_level: default_log_level(),
+            metrics: MetricsSection::default(),
         }
     }
 }
 
+/// `[engine.metrics]` config. When `enabled = true` the engine starts
+/// a Prometheus HTTP exporter on `bind_addr` and serves `/metrics`.
+///
+/// Default: disabled. Operators opt in explicitly so the M3 / M4
+/// runbook smoke runs do not bind a port unintentionally.
+#[derive(Debug, Deserialize)]
+pub struct MetricsSection {
+    #[serde(default)]
+    pub enabled: bool,
+    /// IPv4 / IPv6 socket address to bind. Default `127.0.0.1:9100`.
+    #[serde(default = "default_metrics_bind")]
+    pub bind_addr: String,
+}
+
+impl Default for MetricsSection {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            bind_addr: default_metrics_bind(),
+        }
+    }
+}
+
+fn default_metrics_bind() -> String {
+    "127.0.0.1:9100".to_owned()
+}
+
 #[derive(Debug, Deserialize)]
 pub struct ChainConfig {
     /// JSON-RPC endpoint. `ws://` and `wss://` engage alloy's pubsub
     /// transport (required for `eth_subscribe`); `http://` and `https://`
     /// engage the HTTP transport (request/response only).
     pub rpc_url: String,
+    /// Optional CoW orderbook base URL override for this chain. When
+    /// absent (the common case), the host uses the canonical
+    /// `api.cow.fi/{slug}/api/v1` URL from `cowprotocol::Chain`. Set
+    /// this to point at a staging/barn instance or a local mock (e.g.
+    /// `tools/orderbook-mock` for the COW-1079 load test).
+    #[serde(default)]
+    pub orderbook_url: Option<String>,
 }
 
 fn default_state_dir() -> PathBuf {

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

@@ -29,21 +29,28 @@ pub struct OrderBookPool {
     http: reqwest::Client,
 }
 
+/// Canonical CoW Protocol chain set the engine ships clients for.
+///
+/// Both `Default::default()` and `OrderBookPool::from_config` walk
+/// this single source of truth so a new chain joining CoW protocol
+/// only needs a one-line addition here instead of two parallel
+/// arrays.
+const DEFAULT_CHAINS: &[Chain] = &[
+    Chain::Mainnet,
+    Chain::Gnosis,
+    Chain::Sepolia,
+    Chain::ArbitrumOne,
+    Chain::Base,
+];
+
 impl Default for OrderBookPool {
     /// Build a pool covering every `cowprotocol::Chain` variant. Each entry
     /// uses the canonical `api.cow.fi/{slug}/api/v1` base URL from the SDK.
     /// Override individual entries via `OrderBookApi::new_with_base_url` for
     /// barn or staging targets.
     fn default() -> Self {
         let http = reqwest::Client::new();
-        let chains = [
-            Chain::Mainnet,
-            Chain::Gnosis,
-            Chain::Sepolia,
-            Chain::ArbitrumOne,
-            Chain::Base,
-        ];
-        let clients = chains
+        let clients = DEFAULT_CHAINS
             .iter()
             .map(|c| (c.id(), OrderBookApi::new(*c)))
             .collect();
@@ -52,6 +59,36 @@ impl Default for OrderBookPool {
 }
 
 impl OrderBookPool {
+    /// Build a pool from engine config, honouring any
+    /// `[chains.<id>] orderbook_url = "..."` overrides. Chains
+    /// without an override fall back to the canonical
+    /// `cowprotocol::Chain` URLs (same as [`OrderBookPool::default`]).
+    ///
+    /// Used by the load test (COW-1079) to point all submissions at
+    /// `tools/orderbook-mock`, and by staging/barn deployments that
+    /// run against a non-production orderbook.
+    pub fn from_config(cfg: &crate::engine_config::EngineConfig) -> Self {
+        let http = reqwest::Client::new();
+        let mut clients: BTreeMap<u64, OrderBookApi> = DEFAULT_CHAINS
+            .iter()
+            .map(|c| (c.id(), OrderBookApi::new(*c)))
+            .collect();
+        for (chain_id, chain_cfg) in &cfg.chains {
+            if let Some(url) = chain_cfg.orderbook_url.as_deref() {
+                match url.parse::<url::Url>() {
+                    Ok(parsed) => {
+                        tracing::info!(chain_id, url, "cow-api: orderbook URL override");
+                        clients.insert(*chain_id, OrderBookApi::new_with_base_url(parsed));
+                    }
+                    Err(e) => {
+                        tracing::warn!(chain_id, url, error = %e, "cow-api: bad orderbook_url, falling back to canonical");
+                    }
+                }
+            }
+        }
+        Self { clients, http }
+    }
+
     /// Look up the client for a chain.
     pub fn get(&self, chain_id: u64) -> Result<&OrderBookApi, CowApiError> {
         self.clients

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

@@ -147,6 +147,40 @@ async fn request_rejects_unknown_chain() {
     assert!(matches!(err, CowApiError::UnknownChain(99_999)));
 }
 
+#[tokio::test]
+async fn submit_order_propagates_orderbook_envelope() {
+    // The orderbook rejects with a typed envelope. The pool must
+    // surface `cowprotocol::Error::OrderbookApi { status, api }`
+    // so the WIT adapter can forward `api` to `HostError.data`
+    // (COW-1075). The string `DuplicatedOrder` is what the live
+    // Sepolia orderbook returns for an already-submitted order;
+    // it parses as `ApiError` even though `OrderPostErrorKind`
+    // falls back to `Unknown` for the spelling.
+    let mock = MockServer::start().await;
+    let envelope = r#"{"errorType":"DuplicatedOrder","description":"order already exists"}"#;
+    Mock::given(method("POST"))
+        .and(path("/api/v1/orders"))
+        .respond_with(ResponseTemplate::new(400).set_body_string(envelope))
+        .expect(1)
+        .mount(&mock)
+        .await;
+
+    let pool = pool_with_mainnet_at(&mock);
+    let err = pool
+        .submit_order_json(Chain::Mainnet.id(), sample_order_json().as_bytes())
+        .await
+        .expect_err("orderbook 400 surfaces as error");
+
+    match err {
+        CowApiError::Orderbook(cowprotocol::Error::OrderbookApi { status, api }) => {
+            assert_eq!(status, 400);
+            assert_eq!(api.error_type, "DuplicatedOrder");
+            assert_eq!(api.description, "order already exists");
+        }
+        other => panic!("expected OrderbookApi envelope, got {other:?}"),
+    }
+}
+
 #[tokio::test]
 async fn submit_order_propagates_orderbook_response() {
     let mock = MockServer::start().await;

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

@@ -18,6 +18,7 @@ impl nexum::host::chain::Host for HostState {
     ) -> Result<String, HostError> {
         let start = Instant::now();
         tracing::debug!(chain_id, %method, "chain::request");
+        let method_label = method.clone();
         let result = match self.chain.request(chain_id, method, params).await {
             Ok(body) => Ok(body),
             Err(ProviderError::UnknownChain(id)) => Err(HostError {
@@ -27,23 +28,31 @@ impl nexum::host::chain::Host for HostState {
                 message: format!("chain {id} has no engine.toml RPC entry"),
                 data: None,
             }),
-            Err(ProviderError::InvalidParams { detail, .. }) => Err(HostError {
+            Err(err @ ProviderError::InvalidParams { .. }) => Err(HostError {
                 domain: "chain".into(),
                 kind: HostErrorKind::InvalidInput,
                 code: -32602,
-                message: detail,
+                message: err.to_string(),
                 data: None,
             }),
-            Err(ProviderError::Rpc { detail, .. }) => Err(HostError {
+            Err(err @ ProviderError::Rpc { .. }) => Err(HostError {
                 domain: "chain".into(),
                 kind: HostErrorKind::Internal,
                 code: -32603,
-                message: detail,
+                message: err.to_string(),
                 data: None,
             }),
             Err(err) => Err(internal_error("chain", err.to_string())),
         };
         tracing::trace!(elapsed_ms = ?start.elapsed(), "chain::request done");
+        let outcome = if result.is_ok() { "ok" } else { "err" };
+        metrics::counter!(
+            "shepherd_chain_request_total",
+            "chain_id" => chain_id.to_string(),
+            "method" => method_label,
+            "outcome" => outcome,
+        )
+        .increment(1);
         result
     }