From a5d2becae60b594af4b6ea2bf2c95e51d9311f8b Mon Sep 17 00:00:00 2001 From: Eugenio Paluello Date: Tue, 7 Jul 2026 16:25:19 +0200 Subject: [PATCH 1/5] feat(truapi-host-cli): emit per-frame HostMetricRecord metrics (opt-in JSONL) --- Cargo.lock | 99 +++++++++ rust/crates/truapi-host-cli/Cargo.toml | 2 + .../truapi-host-cli/src/frame_server.rs | 38 +++- rust/crates/truapi-host-cli/src/main.rs | 6 + rust/crates/truapi-host-cli/src/metrics.rs | 192 ++++++++++++++++++ 5 files changed, 331 insertions(+), 6 deletions(-) create mode 100644 rust/crates/truapi-host-cli/src/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index cfe3de7c..dd38ab04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -64,6 +64,15 @@ version = "0.2.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" +[[package]] +name = "android_system_properties" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +dependencies = [ + "libc", +] + [[package]] name = "anstream" version = "1.0.0" @@ -693,6 +702,17 @@ dependencies = [ "rand_core 0.10.1", ] +[[package]] +name = "chrono" +version = "0.4.45" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" +dependencies = [ + "iana-time-zone", + "num-traits", + "windows-link", +] + [[package]] name = "cipher" version = "0.4.4" @@ -1734,6 +1754,30 @@ dependencies = [ "tracing", ] +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + [[package]] name = "icu_collections" version = "2.2.0" @@ -3981,6 +4025,7 @@ dependencies = [ "async-trait", "bip39", "blake2-rfc", + "chrono", "clap", "frame-metadata", "futures", @@ -3990,6 +4035,7 @@ dependencies = [ "reqwest", "rustls", "scale-info", + "serde", "serde_json", "sp-crypto-hashing", "tokio", @@ -4568,12 +4614,65 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.45.0" diff --git a/rust/crates/truapi-host-cli/Cargo.toml b/rust/crates/truapi-host-cli/Cargo.toml index ba7e7eff..e78c9205 100644 --- a/rust/crates/truapi-host-cli/Cargo.toml +++ b/rust/crates/truapi-host-cli/Cargo.toml @@ -30,7 +30,9 @@ scale-info = { version = "2.11", default-features = false, features = ["decode"] sp-crypto-hashing = "0.1" reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } rustls = { version = "0.23", default-features = false, features = ["ring"] } +serde = { version = "1", features = ["derive"] } serde_json = "1" +chrono = { version = "0.4", default-features = false, features = ["clock"] } tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "io-util", "io-std", "net", "time", "signal", "process"] } tokio-stream = { version = "0.1", features = ["sync"] } tokio-tungstenite = { version = "0.24", features = ["connect", "rustls-tls-webpki-roots"] } diff --git a/rust/crates/truapi-host-cli/src/frame_server.rs b/rust/crates/truapi-host-cli/src/frame_server.rs index 15a130ac..22a8b4fb 100644 --- a/rust/crates/truapi-host-cli/src/frame_server.rs +++ b/rust/crates/truapi-host-cli/src/frame_server.rs @@ -7,6 +7,7 @@ use std::net::SocketAddr; use std::sync::Arc; +use std::time::Instant; use anyhow::{Context, Result}; use futures_util::{SinkExt, StreamExt}; @@ -17,6 +18,8 @@ use tokio_tungstenite::tungstenite::Message; use tracing::{debug, info}; use truapi_server::{FrameSink, PairingHostRuntime, ProductContext}; +use crate::metrics::{Category, MetricsRecorder, Outcome}; + /// Frame sink that writes each outgoing protocol frame as one binary message. struct WsFrameSink { outbound: mpsc::UnboundedSender, @@ -46,6 +49,7 @@ pub async fn accept_loop( runtime: Arc, product_id: String, listener: TcpListener, + metrics: Arc, ) -> Result<()> { let bound = listener.local_addr()?; info!(%bound, %product_id, "product frame server listening"); @@ -53,8 +57,9 @@ pub async fn accept_loop( let (stream, peer) = listener.accept().await?; let runtime = runtime.clone(); let product_id = product_id.clone(); + let metrics = metrics.clone(); tokio::task::spawn_local(async move { - if let Err(err) = serve_connection(runtime, product_id, stream).await { + if let Err(err) = serve_connection(runtime, product_id, stream, metrics).await { debug!(%peer, %err, "frame connection ended"); } }); @@ -65,6 +70,7 @@ async fn serve_connection( runtime: Arc, product_id: String, stream: TcpStream, + metrics: Arc, ) -> Result<()> { let ws = accept_async(stream).await?; let (mut write, mut read) = ws.split(); @@ -88,15 +94,18 @@ async fn serve_connection( while let Some(message) = read.next().await { match message { Ok(Message::Binary(bytes)) => { - if let Err(err) = product_runtime.receive_frame(bytes.to_vec()).await { + let started = Instant::now(); + let result = product_runtime.receive_frame(bytes.to_vec()).await; + record_frame(&metrics, started, &result); + if let Err(err) = result { debug!(%err, "product runtime rejected frame"); } } Ok(Message::Text(text)) => { - if let Err(err) = product_runtime - .receive_frame(text.as_bytes().to_vec()) - .await - { + let started = Instant::now(); + let result = product_runtime.receive_frame(text.as_bytes().to_vec()).await; + record_frame(&metrics, started, &result); + if let Err(err) = result { debug!(%err, "product runtime rejected text frame"); } } @@ -110,3 +119,20 @@ async fn serve_connection( let _ = writer.await; Ok(()) } + +/// Record one product-frame operation: latency from receive to dispatch +/// completion (`receive_frame` awaits the handler), and its outcome. The wire +/// id is not decoded yet, so the category is the coarse `Frame`; decoding it +/// into per-method categories is the next step. +fn record_frame( + metrics: &MetricsRecorder, + started: Instant, + result: &Result<(), E>, +) { + let latency_ms = started.elapsed().as_secs_f64() * 1000.0; + let (outcome, error_class) = match result { + Ok(()) => (Outcome::Success, None), + Err(err) => (Outcome::Error, Some(err.to_string())), + }; + metrics.record(Category::Frame, "product_frame", latency_ms, outcome, error_class); +} diff --git a/rust/crates/truapi-host-cli/src/main.rs b/rust/crates/truapi-host-cli/src/main.rs index f1c9050e..9a22c973 100644 --- a/rust/crates/truapi-host-cli/src/main.rs +++ b/rust/crates/truapi-host-cli/src/main.rs @@ -13,6 +13,7 @@ mod alloc; mod attestation; mod chain; mod frame_server; +mod metrics; mod platform; mod script_runner; @@ -370,6 +371,10 @@ async fn run_pairing_host( let frame_url = format!("ws://{}", listener.local_addr()?); println!("FRAMES_LISTENING {frame_url}"); + // Per-operation metrics for the stress/sim layer. Opt-in via `METRICS_JSONL`; + // a no-op recorder otherwise, so default behaviour is unchanged. + let metrics = metrics::MetricsRecorder::from_env(); + let local = tokio::task::LocalSet::new(); let status = local .run_until(async move { @@ -377,6 +382,7 @@ async fn run_pairing_host( runtime, product_id.clone(), listener, + metrics, )); let status = script_runner::run(&frame_url, &product_id, &script).await; server.abort(); diff --git a/rust/crates/truapi-host-cli/src/metrics.rs b/rust/crates/truapi-host-cli/src/metrics.rs new file mode 100644 index 00000000..06d476be --- /dev/null +++ b/rust/crates/truapi-host-cli/src/metrics.rs @@ -0,0 +1,192 @@ +//! Headless-host metrics: one raw per-operation event, written as JSONL. +//! +//! First slice of the Host SDK simulation/stress layer. We only emit raw +//! events here; percentiles and aggregation are computed downstream (e.g. +//! product-loadtest ingest). Opt-in: without `METRICS_JSONL` set, recording is +//! a no-op and the host behaves exactly as before. + +use std::fs::OpenOptions; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use serde::Serialize; + +/// What kind of host operation a record measures. +/// +/// `Frame` is the coarse label for a whole product request frame at the +/// WebSocket boundary, where the wire id is not yet decoded. Decoding the +/// `ProtocolMessage` wire id into the fine-grained categories below (Signing, +/// Storage, ChainRpc, ...) is the next step and needs a small decode export +/// from `truapi-server`. +// The fine-grained variants are the full metric schema, used once the wire id +// is decoded (next step); `Frame` is all v1 emits today. +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Category { + Frame, + Pairing, + Signing, + Subscription, + HostCallback, + ChainRpc, + Storage, + Permission, + Memory, + Session, +} + +/// Terminal outcome of a measured operation. `Skipped` is part of the schema +/// for operations that never ran; v1 emits only `Success`/`Error`. +#[allow(dead_code)] +#[derive(Debug, Clone, Copy, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Outcome { + Success, + Error, + Skipped, +} + +/// One per-operation host metric event. Serialises to camelCase JSON. +#[derive(Debug, Clone, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct HostMetricRecord { + pub ts: String, + pub run_id: String, + pub vu_index: u32, + pub category: Category, + pub op: String, + pub latency_ms: f64, + pub outcome: Outcome, + #[serde(skip_serializing_if = "Option::is_none")] + pub error_class: Option, +} + +/// Appends `HostMetricRecord`s as newline-delimited JSON. +struct JsonlSink { + path: PathBuf, +} + +impl JsonlSink { + fn new(path: impl AsRef) -> Self { + Self { + path: path.as_ref().to_path_buf(), + } + } + + fn append(&self, record: &HostMetricRecord) -> std::io::Result<()> { + let line = serde_json::to_string(record) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(&self.path)?; + file.write_all(line.as_bytes())?; + file.write_all(b"\n") + } +} + +/// Records per-operation metrics for one run. Cheap to clone through an `Arc`. +/// +/// Configured from the environment: `METRICS_JSONL` (output path; unset means +/// recording is disabled), `RUN_ID` (defaults to `local`), `VU_INDEX` +/// (defaults to `0`). +pub struct MetricsRecorder { + sink: Option, + run_id: String, + vu_index: u32, +} + +impl MetricsRecorder { + /// Build a recorder from the environment. Returns an `Arc` so it can be + /// shared across connections. + pub fn from_env() -> Arc { + let sink = std::env::var("METRICS_JSONL") + .ok() + .filter(|p| !p.is_empty()) + .map(JsonlSink::new); + let run_id = std::env::var("RUN_ID") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "local".to_string()); + let vu_index = std::env::var("VU_INDEX") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(0); + Arc::new(Self { + sink, + run_id, + vu_index, + }) + } + + /// Record one operation. No-op when no sink is configured. + pub fn record( + &self, + category: Category, + op: &str, + latency_ms: f64, + outcome: Outcome, + error_class: Option, + ) { + let Some(sink) = &self.sink else { + return; + }; + let record = HostMetricRecord { + ts: chrono::Utc::now().to_rfc3339(), + run_id: self.run_id.clone(), + vu_index: self.vu_index, + category, + op: op.to_string(), + latency_ms, + outcome, + error_class, + }; + if let Err(err) = sink.append(&record) { + tracing::warn!(%err, "failed to append host metric record"); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn sample() -> HostMetricRecord { + HostMetricRecord { + ts: "2026-07-07T00:00:00+00:00".into(), + run_id: "run-1".into(), + vu_index: 0, + category: Category::Frame, + op: "product_frame".into(), + latency_ms: 12.5, + outcome: Outcome::Success, + error_class: None, + } + } + + #[test] + fn serialises_to_camel_case_snake_enums() { + let v = serde_json::to_value(sample()).unwrap(); + assert_eq!(v["runId"], "run-1"); + assert_eq!(v["vuIndex"], 0); + assert_eq!(v["category"], "frame"); + assert_eq!(v["outcome"], "success"); + assert_eq!(v["latencyMs"], 12.5); + assert!(v.get("errorClass").is_none(), "None error_class is skipped"); + } + + #[test] + fn appends_one_line_per_record() { + let path = std::env::temp_dir().join("truapi-host-cli-metrics-unit.jsonl"); + let _ = std::fs::remove_file(&path); + let sink = JsonlSink::new(&path); + sink.append(&sample()).unwrap(); + sink.append(&sample()).unwrap(); + let body = std::fs::read_to_string(&path).unwrap(); + assert_eq!(body.lines().count(), 2); + let first: serde_json::Value = serde_json::from_str(body.lines().next().unwrap()).unwrap(); + assert_eq!(first["op"], "product_frame"); + } +} From 1f39102f7eb5f0c41d80e81816e295de641c7379 Mon Sep 17 00:00:00 2001 From: Eugenio Paluello Date: Tue, 7 Jul 2026 16:56:22 +0200 Subject: [PATCH 2/5] feat(truapi-host-cli): classify frame metrics by wire id (per-method category/op) --- .../truapi-host-cli/src/frame_server.rs | 22 ++++--- rust/crates/truapi-host-cli/src/metrics.rs | 64 +++++++++++++++++++ 2 files changed, 77 insertions(+), 9 deletions(-) diff --git a/rust/crates/truapi-host-cli/src/frame_server.rs b/rust/crates/truapi-host-cli/src/frame_server.rs index 22a8b4fb..6ed2f82a 100644 --- a/rust/crates/truapi-host-cli/src/frame_server.rs +++ b/rust/crates/truapi-host-cli/src/frame_server.rs @@ -18,7 +18,7 @@ use tokio_tungstenite::tungstenite::Message; use tracing::{debug, info}; use truapi_server::{FrameSink, PairingHostRuntime, ProductContext}; -use crate::metrics::{Category, MetricsRecorder, Outcome}; +use crate::metrics::{Category, MetricsRecorder, Outcome, classify_frame}; /// Frame sink that writes each outgoing protocol frame as one binary message. struct WsFrameSink { @@ -94,17 +94,20 @@ async fn serve_connection( while let Some(message) = read.next().await { match message { Ok(Message::Binary(bytes)) => { + let (category, op) = classify_frame(&bytes); let started = Instant::now(); let result = product_runtime.receive_frame(bytes.to_vec()).await; - record_frame(&metrics, started, &result); + record_frame(&metrics, category, &op, started, &result); if let Err(err) = result { debug!(%err, "product runtime rejected frame"); } } Ok(Message::Text(text)) => { + let bytes = text.as_bytes().to_vec(); + let (category, op) = classify_frame(&bytes); let started = Instant::now(); - let result = product_runtime.receive_frame(text.as_bytes().to_vec()).await; - record_frame(&metrics, started, &result); + let result = product_runtime.receive_frame(bytes).await; + record_frame(&metrics, category, &op, started, &result); if let Err(err) = result { debug!(%err, "product runtime rejected text frame"); } @@ -120,12 +123,13 @@ async fn serve_connection( Ok(()) } -/// Record one product-frame operation: latency from receive to dispatch -/// completion (`receive_frame` awaits the handler), and its outcome. The wire -/// id is not decoded yet, so the category is the coarse `Frame`; decoding it -/// into per-method categories is the next step. +/// Record one product-frame operation: its method-derived `category`/`op` (from +/// the decoded wire id), the latency from receive to dispatch completion +/// (`receive_frame` awaits the handler), and its outcome. fn record_frame( metrics: &MetricsRecorder, + category: Category, + op: &str, started: Instant, result: &Result<(), E>, ) { @@ -134,5 +138,5 @@ fn record_frame( Ok(()) => (Outcome::Success, None), Err(err) => (Outcome::Error, Some(err.to_string())), }; - metrics.record(Category::Frame, "product_frame", latency_ms, outcome, error_class); + metrics.record(category, op, latency_ms, outcome, error_class); } diff --git a/rust/crates/truapi-host-cli/src/metrics.rs b/rust/crates/truapi-host-cli/src/metrics.rs index 06d476be..5653bbed 100644 --- a/rust/crates/truapi-host-cli/src/metrics.rs +++ b/rust/crates/truapi-host-cli/src/metrics.rs @@ -149,10 +149,74 @@ impl MetricsRecorder { } } +/// Classify a raw inbound frame into a metric `(category, op)` by decoding its +/// wire discriminant and looking the method up in the core's wire table. +/// +/// `op` is the trait method name (e.g. `signing_sign_raw`); `category` is a +/// coarse bucket from the method's namespace, or `Subscription` for any +/// subscription frame. Falls back to `(Frame, "product_frame")` when the frame +/// does not decode or the id is unknown. +pub fn classify_frame(frame: &[u8]) -> (Category, String) { + use parity_scale_codec::Decode; + use truapi_server::frame::ProtocolMessage; + use truapi_server::generated::wire_table::{WIRE_TABLE, WireKind}; + + let Ok(message) = ProtocolMessage::decode(&mut &*frame) else { + return (Category::Frame, "product_frame".to_string()); + }; + let id = message.payload.id; + for entry in WIRE_TABLE { + let (is_subscription, matches) = match &entry.kind { + WireKind::Request(r) => (false, r.request_id == id || r.response_id == id), + WireKind::Subscription(s) => ( + true, + s.start_id == id + || s.stop_id == id + || s.interrupt_id == id + || s.receive_id == id, + ), + }; + if matches { + let category = if is_subscription { + Category::Subscription + } else { + category_for_method(entry.method) + }; + return (category, entry.method.to_string()); + } + } + (Category::Frame, "product_frame".to_string()) +} + +/// Map a method name to a coarse metric category by its namespace prefix. +fn category_for_method(method: &str) -> Category { + match method.split('_').next().unwrap_or("") { + "signing" | "entropy" | "statement" => Category::Signing, + "chain" => Category::ChainRpc, + "local" => Category::Storage, + "permissions" => Category::Permission, + "account" => Category::Pairing, + "system" => Category::Session, + "notifications" | "preimage" | "theme" | "resource" | "coin" | "payment" | "chat" => { + Category::HostCallback + } + _ => Category::Frame, + } +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn category_maps_by_namespace() { + assert!(matches!(category_for_method("signing_sign_raw"), Category::Signing)); + assert!(matches!(category_for_method("chain_call"), Category::ChainRpc)); + assert!(matches!(category_for_method("local_storage_write"), Category::Storage)); + assert!(matches!(category_for_method("account_request_login"), Category::Pairing)); + assert!(matches!(category_for_method("mystery_method"), Category::Frame)); + } + fn sample() -> HostMetricRecord { HostMetricRecord { ts: "2026-07-07T00:00:00+00:00".into(), From 355f310523cfc0ab4125af86a1660ada4b296f17 Mon Sep 17 00:00:00 2001 From: Eugenio Paluello Date: Tue, 7 Jul 2026 17:50:09 +0200 Subject: [PATCH 3/5] feat(truapi-host-cli): metric outcome reflects response Result, not just dispatch --- .../truapi-host-cli/src/frame_server.rs | 51 ++++++++++----- rust/crates/truapi-host-cli/src/metrics.rs | 62 +++++++++++++++++-- 2 files changed, 95 insertions(+), 18 deletions(-) diff --git a/rust/crates/truapi-host-cli/src/frame_server.rs b/rust/crates/truapi-host-cli/src/frame_server.rs index 6ed2f82a..c7d6936a 100644 --- a/rust/crates/truapi-host-cli/src/frame_server.rs +++ b/rust/crates/truapi-host-cli/src/frame_server.rs @@ -5,8 +5,9 @@ //! binary messages. One binary WS message carries exactly one SCALE //! `ProtocolMessage`, matching the browser transport's framing. +use std::collections::HashMap; use std::net::SocketAddr; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Instant; use anyhow::{Context, Result}; @@ -18,15 +19,26 @@ use tokio_tungstenite::tungstenite::Message; use tracing::{debug, info}; use truapi_server::{FrameSink, PairingHostRuntime, ProductContext}; -use crate::metrics::{Category, MetricsRecorder, Outcome, classify_frame}; +use crate::metrics::{MetricsRecorder, Outcome, classify_frame, response_outcome}; -/// Frame sink that writes each outgoing protocol frame as one binary message. +/// True per-request outcomes, decoded from response frames as they are emitted +/// and consumed by the request loop, keyed by `request_id`. +type OutcomeMap = Arc>>; + +/// Frame sink that writes each outgoing protocol frame as one binary message, +/// and records the true outcome of response frames for the metrics layer. struct WsFrameSink { outbound: mpsc::UnboundedSender, + outcomes: OutcomeMap, } impl FrameSink for WsFrameSink { fn emit_frame(&self, frame: Vec) { + if let Some((request_id, outcome)) = response_outcome(&frame) { + if let Ok(mut map) = self.outcomes.lock() { + map.insert(request_id, outcome); + } + } let _ = self.outbound.send(Message::Binary(frame)); } } @@ -86,28 +98,30 @@ async fn serve_connection( let product = ProductContext::new(product_id) .map_err(|err| anyhow::anyhow!("invalid product id: {err}"))?; + let outcomes: OutcomeMap = Arc::new(Mutex::new(HashMap::new())); let sink = Arc::new(WsFrameSink { outbound: outbound_tx.clone(), + outcomes: outcomes.clone(), }); let product_runtime = runtime.product_runtime(product, sink); while let Some(message) = read.next().await { match message { Ok(Message::Binary(bytes)) => { - let (category, op) = classify_frame(&bytes); + let class = classify_frame(&bytes); let started = Instant::now(); let result = product_runtime.receive_frame(bytes.to_vec()).await; - record_frame(&metrics, category, &op, started, &result); + record_frame(&metrics, &class, started, &result, &outcomes); if let Err(err) = result { debug!(%err, "product runtime rejected frame"); } } Ok(Message::Text(text)) => { let bytes = text.as_bytes().to_vec(); - let (category, op) = classify_frame(&bytes); + let class = classify_frame(&bytes); let started = Instant::now(); let result = product_runtime.receive_frame(bytes).await; - record_frame(&metrics, category, &op, started, &result); + record_frame(&metrics, &class, started, &result, &outcomes); if let Err(err) = result { debug!(%err, "product runtime rejected text frame"); } @@ -123,20 +137,29 @@ async fn serve_connection( Ok(()) } -/// Record one product-frame operation: its method-derived `category`/`op` (from -/// the decoded wire id), the latency from receive to dispatch completion -/// (`receive_frame` awaits the handler), and its outcome. +/// Record one product-frame operation: its method-derived `category`/`op`, the +/// latency from receive to dispatch completion, and its true outcome. +/// +/// A dispatch-level failure (`receive_frame` returns Err) is an error. On a +/// clean dispatch, the outcome is the one the response frame carried (captured +/// by the sink under this request's id): a domain error in the response is an +/// error even though `receive_frame` returned Ok. If no response was captured +/// (e.g. a subscription start), the clean dispatch counts as success. fn record_frame( metrics: &MetricsRecorder, - category: Category, - op: &str, + class: &crate::metrics::FrameClass, started: Instant, result: &Result<(), E>, + outcomes: &OutcomeMap, ) { let latency_ms = started.elapsed().as_secs_f64() * 1000.0; let (outcome, error_class) = match result { - Ok(()) => (Outcome::Success, None), Err(err) => (Outcome::Error, Some(err.to_string())), + Ok(()) => match outcomes.lock().ok().and_then(|mut m| m.remove(&class.request_id)) { + Some(Outcome::Error) => (Outcome::Error, Some("response_error".to_string())), + Some(other) => (other, None), + None => (Outcome::Success, None), + }, }; - metrics.record(category, op, latency_ms, outcome, error_class); + metrics.record(class.category, &class.op, latency_ms, outcome, error_class); } diff --git a/rust/crates/truapi-host-cli/src/metrics.rs b/rust/crates/truapi-host-cli/src/metrics.rs index 5653bbed..4a81df63 100644 --- a/rust/crates/truapi-host-cli/src/metrics.rs +++ b/rust/crates/truapi-host-cli/src/metrics.rs @@ -156,14 +156,26 @@ impl MetricsRecorder { /// coarse bucket from the method's namespace, or `Subscription` for any /// subscription frame. Falls back to `(Frame, "product_frame")` when the frame /// does not decode or the id is unknown. -pub fn classify_frame(frame: &[u8]) -> (Category, String) { +/// A decoded inbound frame's metric identity. +pub struct FrameClass { + pub category: Category, + pub op: String, + pub request_id: String, +} + +pub fn classify_frame(frame: &[u8]) -> FrameClass { use parity_scale_codec::Decode; use truapi_server::frame::ProtocolMessage; use truapi_server::generated::wire_table::{WIRE_TABLE, WireKind}; let Ok(message) = ProtocolMessage::decode(&mut &*frame) else { - return (Category::Frame, "product_frame".to_string()); + return FrameClass { + category: Category::Frame, + op: "product_frame".to_string(), + request_id: String::new(), + }; }; + let request_id = message.request_id; let id = message.payload.id; for entry in WIRE_TABLE { let (is_subscription, matches) = match &entry.kind { @@ -182,10 +194,52 @@ pub fn classify_frame(frame: &[u8]) -> (Category, String) { } else { category_for_method(entry.method) }; - return (category, entry.method.to_string()); + return FrameClass { + category, + op: entry.method.to_string(), + request_id, + }; + } + } + FrameClass { + category: Category::Frame, + op: "product_frame".to_string(), + request_id, + } +} + +/// Decode a frame emitted back to the product and extract the true operation +/// outcome, keyed by `request_id`. Returns `Some` for response frames and for +/// error interrupts; `None` for streamed subscription items or undecodable +/// frames (whose outcome is left to the dispatch result). +/// +/// Request/response payloads are versioned as `[version_index][result_disc]..` +/// where `result_disc` is 0 for Ok and 1 for Err (truapi-server `frame.rs`), so +/// a domain error in the response is caught even though `receive_frame` still +/// returns Ok for it. +pub fn response_outcome(frame: &[u8]) -> Option<(String, Outcome)> { + use parity_scale_codec::Decode; + use truapi_server::frame::ProtocolMessage; + use truapi_server::generated::wire_table::{WIRE_TABLE, WireKind}; + + let message = ProtocolMessage::decode(&mut &*frame).ok()?; + let id = message.payload.id; + for entry in WIRE_TABLE { + match &entry.kind { + WireKind::Request(r) if r.response_id == id => { + let outcome = match message.payload.value.get(1) { + Some(1) => Outcome::Error, + _ => Outcome::Success, + }; + return Some((message.request_id, outcome)); + } + WireKind::Subscription(s) if s.interrupt_id == id => { + return Some((message.request_id, Outcome::Error)); + } + _ => {} } } - (Category::Frame, "product_frame".to_string()) + None } /// Map a method name to a coarse metric category by its namespace prefix. From d4914ce8d635d416ab479596bd6f3dc0f95807ab Mon Sep 17 00:00:00 2001 From: Eugenio Paluello Date: Tue, 7 Jul 2026 18:11:41 +0200 Subject: [PATCH 4/5] feat(truapi-host-cli): fleet runner (N VUs on a ramp, paired via signing-bot, per-VU metrics) --- rust/crates/truapi-host-cli/e2e/fleet.sh | 69 ++++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100755 rust/crates/truapi-host-cli/e2e/fleet.sh diff --git a/rust/crates/truapi-host-cli/e2e/fleet.sh b/rust/crates/truapi-host-cli/e2e/fleet.sh new file mode 100755 index 00000000..2dca8f2e --- /dev/null +++ b/rust/crates/truapi-host-cli/e2e/fleet.sh @@ -0,0 +1,69 @@ +#!/usr/bin/env bash +# Fleet runner (first slice): N virtual users, each a pairing host on its own +# port, paired against the signing-bot (the persona pool), started on a ramp, +# all emitting per-VU metrics (distinct VU_INDEX) to one shared JSONL. +# +# VUS=3 RAMP=3 e2e/fleet.sh # 3 VUs, 3s apart, battery.ts each +# SCRIPT=path/to/flow.ts VUS=5 e2e/fleet.sh +# +# Env: +# VUS number of virtual users (default 3) +# RAMP seconds between VU starts (default 3) +# SCRIPT product script each VU runs (default battery.ts) +# BASE_PORT first frame-server port; VU i uses BASE_PORT+i (default 9955) +# BOT signing-bot base URL (default http://localhost:3737) +# NETWORK pairing network (default paseo-next-v2) +# RUN_ID shared run id (default fleet-) +# METRICS_JSONL shared metrics sink (default /tmp/fleet-metrics.jsonl) +# +# Each VU pairs against the bot, which auto-provisions an attested user and +# signs, so scale is bounded by the bot's per-user attestation, not the host. +set -uo pipefail + +ROOT="$(cd "$(dirname "$0")/../../../.." && pwd)" +BIN="$ROOT/target/debug/truapi-host" +SCRIPT="${SCRIPT:-$ROOT/rust/crates/truapi-host-cli/js/scripts/battery.ts}" +VUS="${VUS:-3}" +RAMP="${RAMP:-3}" +BASE_PORT="${BASE_PORT:-9955}" +BOT="${BOT:-http://localhost:3737}" +NETWORK="${NETWORK:-paseo-next-v2}" +export RUN_ID="${RUN_ID:-fleet-$(date +%s)}" +export METRICS_JSONL="${METRICS_JSONL:-/tmp/fleet-metrics.jsonl}" + +[ -x "$BIN" ] || { echo "missing $BIN — build first (cargo build -p truapi-host-cli)" >&2; exit 2; } +: > "$METRICS_JSONL" + +# One VU: start its pairing host, hand the deeplink to the bot, wait for it. +run_vu() { + local i="$1" port="$2" log + log="$(mktemp)" + VU_INDEX="$i" "$BIN" pairing-host --product-id headless-playground.dot \ + --script "$SCRIPT" --frame-listen "127.0.0.1:$port" --auto-accept >"$log" 2>&1 & + local ph=$! + local deeplink="" + for _ in $(seq 1 240); do + deeplink="$(grep -m1 -oE 'PAIRING_DEEPLINK .+' "$log" | cut -d' ' -f2- || true)" + [ -n "$deeplink" ] && break + kill -0 "$ph" 2>/dev/null || break + sleep 0.5 + done + if [ -z "$deeplink" ]; then + echo "VU$i: no deeplink (host died?)"; tail -3 "$log"; kill "$ph" 2>/dev/null; rm -f "$log"; return 1 + fi + curl -s -m 180 -X POST "$BOT/api/pair" -H 'content-type: application/json' \ + -d "{\"handshake\":\"$deeplink\",\"network\":\"$NETWORK\"}" -o /dev/null \ + -w "VU$i: bot /api/pair=%{http_code}\n" + wait "$ph"; echo "VU$i: pairing host exit=$?" + rm -f "$log" +} + +echo "fleet: VUS=$VUS ramp=${RAMP}s script=$(basename "$SCRIPT") run_id=$RUN_ID" +pids=() +for i in $(seq 0 $((VUS - 1))); do + run_vu "$i" "$((BASE_PORT + i))" & + pids+=($!) + sleep "$RAMP" +done +for p in "${pids[@]}"; do wait "$p"; done +echo "fleet complete -> $METRICS_JSONL" From c9d02de1db579294859196e93bf91c9fca600c33 Mon Sep 17 00:00:00 2001 From: Eugenio Paluello Date: Wed, 8 Jul 2026 11:25:13 +0200 Subject: [PATCH 5/5] fix(truapi-host-cli): address review (clippy CI errors, async interrupt mislabel, wire-layout test, doc/fleet cleanup) --- rust/crates/truapi-host-cli/Cargo.toml | 2 +- rust/crates/truapi-host-cli/e2e/fleet.sh | 61 +++++++--- .../truapi-host-cli/src/frame_server.rs | 30 ++--- rust/crates/truapi-host-cli/src/metrics.rs | 113 +++++++++++------- 4 files changed, 128 insertions(+), 78 deletions(-) diff --git a/rust/crates/truapi-host-cli/Cargo.toml b/rust/crates/truapi-host-cli/Cargo.toml index e78c9205..ef892d28 100644 --- a/rust/crates/truapi-host-cli/Cargo.toml +++ b/rust/crates/truapi-host-cli/Cargo.toml @@ -20,6 +20,7 @@ anyhow = "1" async-trait = "0.1" bip39 = "2" blake2-rfc = { version = "0.2", default-features = false } +chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = { version = "4", features = ["derive"] } frame-metadata = { version = "23", default-features = false, features = ["std", "current", "decode"] } futures = "0.3" @@ -32,7 +33,6 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus rustls = { version = "0.23", default-features = false, features = ["ring"] } serde = { version = "1", features = ["derive"] } serde_json = "1" -chrono = { version = "0.4", default-features = false, features = ["clock"] } tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "io-util", "io-std", "net", "time", "signal", "process"] } tokio-stream = { version = "0.1", features = ["sync"] } tokio-tungstenite = { version = "0.24", features = ["connect", "rustls-tls-webpki-roots"] } diff --git a/rust/crates/truapi-host-cli/e2e/fleet.sh b/rust/crates/truapi-host-cli/e2e/fleet.sh index 2dca8f2e..18529526 100755 --- a/rust/crates/truapi-host-cli/e2e/fleet.sh +++ b/rust/crates/truapi-host-cli/e2e/fleet.sh @@ -7,40 +7,57 @@ # SCRIPT=path/to/flow.ts VUS=5 e2e/fleet.sh # # Env: -# VUS number of virtual users (default 3) -# RAMP seconds between VU starts (default 3) -# SCRIPT product script each VU runs (default battery.ts) -# BASE_PORT first frame-server port; VU i uses BASE_PORT+i (default 9955) -# BOT signing-bot base URL (default http://localhost:3737) -# NETWORK pairing network (default paseo-next-v2) -# RUN_ID shared run id (default fleet-) -# METRICS_JSONL shared metrics sink (default /tmp/fleet-metrics.jsonl) +# VUS number of virtual users (default 3) +# RAMP seconds between VU starts (default 3) +# SCRIPT product script each VU runs (default battery.ts) +# PRODUCT_ID product id each VU serves (default headless-playground.dot) +# BASE_PORT first frame-server port; VU i uses BASE_PORT+i (default 9955) +# SIGNER_BOT_BASE_URL signing-bot base URL (default http://localhost:3737) +# SIGNER_BOT_NETWORK pairing network (default paseo-next-v2) +# SIGNER_BOT_SVC_TOKEN bearer token; sent only if set (a local dev bot needs none) +# RUN_ID shared run id (default fleet-) +# METRICS_JSONL shared metrics sink (default /tmp/fleet-metrics.jsonl) # # Each VU pairs against the bot, which auto-provisions an attested user and # signs, so scale is bounded by the bot's per-user attestation, not the host. -set -uo pipefail +# The default target is an unauthenticated local bot; point it at an +# authenticated one by setting SIGNER_BOT_BASE_URL and SIGNER_BOT_SVC_TOKEN. +set -euo pipefail ROOT="$(cd "$(dirname "$0")/../../../.." && pwd)" BIN="$ROOT/target/debug/truapi-host" SCRIPT="${SCRIPT:-$ROOT/rust/crates/truapi-host-cli/js/scripts/battery.ts}" VUS="${VUS:-3}" RAMP="${RAMP:-3}" +PRODUCT_ID="${PRODUCT_ID:-headless-playground.dot}" BASE_PORT="${BASE_PORT:-9955}" -BOT="${BOT:-http://localhost:3737}" -NETWORK="${NETWORK:-paseo-next-v2}" +BOT="${SIGNER_BOT_BASE_URL:-http://localhost:3737}" +NETWORK="${SIGNER_BOT_NETWORK:-paseo-next-v2}" export RUN_ID="${RUN_ID:-fleet-$(date +%s)}" export METRICS_JSONL="${METRICS_JSONL:-/tmp/fleet-metrics.jsonl}" [ -x "$BIN" ] || { echo "missing $BIN — build first (cargo build -p truapi-host-cli)" >&2; exit 2; } + +# Per-VU host pids and logs live here; the trap kills leaked hosts and clears +# the logs on any exit, including Ctrl-C mid-ramp. +WORKDIR="$(mktemp -d)" +cleanup() { + for f in "$WORKDIR"/*.pid; do + [ -f "$f" ] && kill "$(cat "$f")" 2>/dev/null || true + done + rm -rf "$WORKDIR" +} +trap cleanup EXIT + : > "$METRICS_JSONL" # One VU: start its pairing host, hand the deeplink to the bot, wait for it. run_vu() { - local i="$1" port="$2" log - log="$(mktemp)" - VU_INDEX="$i" "$BIN" pairing-host --product-id headless-playground.dot \ + local i="$1" port="$2" log="$WORKDIR/vu-$1.log" + VU_INDEX="$i" "$BIN" pairing-host --product-id "$PRODUCT_ID" \ --script "$SCRIPT" --frame-listen "127.0.0.1:$port" --auto-accept >"$log" 2>&1 & local ph=$! + echo "$ph" >"$WORKDIR/vu-$i.pid" local deeplink="" for _ in $(seq 1 240); do deeplink="$(grep -m1 -oE 'PAIRING_DEEPLINK .+' "$log" | cut -d' ' -f2- || true)" @@ -49,13 +66,19 @@ run_vu() { sleep 0.5 done if [ -z "$deeplink" ]; then - echo "VU$i: no deeplink (host died?)"; tail -3 "$log"; kill "$ph" 2>/dev/null; rm -f "$log"; return 1 + echo "VU$i: no deeplink (host died?)"; tail -3 "$log"; return 1 + fi + local auth=() + if [ -n "${SIGNER_BOT_SVC_TOKEN:-}" ]; then + auth=(-H "authorization: Bearer $SIGNER_BOT_SVC_TOKEN") fi curl -s -m 180 -X POST "$BOT/api/pair" -H 'content-type: application/json' \ + ${auth[@]+"${auth[@]}"} \ -d "{\"handshake\":\"$deeplink\",\"network\":\"$NETWORK\"}" -o /dev/null \ - -w "VU$i: bot /api/pair=%{http_code}\n" - wait "$ph"; echo "VU$i: pairing host exit=$?" - rm -f "$log" + -w "VU$i: bot /api/pair=%{http_code}\n" || true + local rc=0 + wait "$ph" || rc=$? + echo "VU$i: pairing host exit=$rc" } echo "fleet: VUS=$VUS ramp=${RAMP}s script=$(basename "$SCRIPT") run_id=$RUN_ID" @@ -65,5 +88,5 @@ for i in $(seq 0 $((VUS - 1))); do pids+=($!) sleep "$RAMP" done -for p in "${pids[@]}"; do wait "$p"; done +for p in "${pids[@]}"; do wait "$p" || true; done echo "fleet complete -> $METRICS_JSONL" diff --git a/rust/crates/truapi-host-cli/src/frame_server.rs b/rust/crates/truapi-host-cli/src/frame_server.rs index c7d6936a..5a1c0a92 100644 --- a/rust/crates/truapi-host-cli/src/frame_server.rs +++ b/rust/crates/truapi-host-cli/src/frame_server.rs @@ -19,7 +19,7 @@ use tokio_tungstenite::tungstenite::Message; use tracing::{debug, info}; use truapi_server::{FrameSink, PairingHostRuntime, ProductContext}; -use crate::metrics::{MetricsRecorder, Outcome, classify_frame, response_outcome}; +use crate::metrics::{FrameClass, MetricsRecorder, Outcome, classify_frame, response_outcome}; /// True per-request outcomes, decoded from response frames as they are emitted /// and consumed by the request loop, keyed by `request_id`. @@ -34,10 +34,10 @@ struct WsFrameSink { impl FrameSink for WsFrameSink { fn emit_frame(&self, frame: Vec) { - if let Some((request_id, outcome)) = response_outcome(&frame) { - if let Ok(mut map) = self.outcomes.lock() { - map.insert(request_id, outcome); - } + if let Some((request_id, outcome)) = response_outcome(&frame) + && let Ok(mut map) = self.outcomes.lock() + { + map.insert(request_id, outcome); } let _ = self.outbound.send(Message::Binary(frame)); } @@ -137,17 +137,17 @@ async fn serve_connection( Ok(()) } -/// Record one product-frame operation: its method-derived `category`/`op`, the -/// latency from receive to dispatch completion, and its true outcome. -/// -/// A dispatch-level failure (`receive_frame` returns Err) is an error. On a -/// clean dispatch, the outcome is the one the response frame carried (captured -/// by the sink under this request's id): a domain error in the response is an -/// error even though `receive_frame` returned Ok. If no response was captured -/// (e.g. a subscription start), the clean dispatch counts as success. +/// Error class recorded when dispatch succeeded but the response frame carried +/// a domain error. This is an emitted-schema value consumed by downstream ingest. +const RESPONSE_ERROR_CLASS: &str = "response_error"; + +/// Record one product-frame op: its `(category, op)`, receive-to-dispatch +/// latency, and true outcome. A clean dispatch (`Ok`) still counts as an +/// error when the captured response frame carried a domain error; a dispatch +/// with no captured response counts as success. fn record_frame( metrics: &MetricsRecorder, - class: &crate::metrics::FrameClass, + class: &FrameClass, started: Instant, result: &Result<(), E>, outcomes: &OutcomeMap, @@ -156,7 +156,7 @@ fn record_frame( let (outcome, error_class) = match result { Err(err) => (Outcome::Error, Some(err.to_string())), Ok(()) => match outcomes.lock().ok().and_then(|mut m| m.remove(&class.request_id)) { - Some(Outcome::Error) => (Outcome::Error, Some("response_error".to_string())), + Some(Outcome::Error) => (Outcome::Error, Some(RESPONSE_ERROR_CLASS.to_string())), Some(other) => (other, None), None => (Outcome::Success, None), }, diff --git a/rust/crates/truapi-host-cli/src/metrics.rs b/rust/crates/truapi-host-cli/src/metrics.rs index 4a81df63..6ef8a874 100644 --- a/rust/crates/truapi-host-cli/src/metrics.rs +++ b/rust/crates/truapi-host-cli/src/metrics.rs @@ -1,9 +1,8 @@ //! Headless-host metrics: one raw per-operation event, written as JSONL. //! -//! First slice of the Host SDK simulation/stress layer. We only emit raw -//! events here; percentiles and aggregation are computed downstream (e.g. -//! product-loadtest ingest). Opt-in: without `METRICS_JSONL` set, recording is -//! a no-op and the host behaves exactly as before. +//! Raw events only; percentiles and aggregation are computed downstream. +//! Opt-in: without `METRICS_JSONL` set, recording is a no-op and the host +//! behaves exactly as before. use std::fs::OpenOptions; use std::io::Write; @@ -15,12 +14,8 @@ use serde::Serialize; /// What kind of host operation a record measures. /// /// `Frame` is the coarse label for a whole product request frame at the -/// WebSocket boundary, where the wire id is not yet decoded. Decoding the -/// `ProtocolMessage` wire id into the fine-grained categories below (Signing, -/// Storage, ChainRpc, ...) is the next step and needs a small decode export -/// from `truapi-server`. -// The fine-grained variants are the full metric schema, used once the wire id -// is decoded (next step); `Frame` is all v1 emits today. +/// WebSocket boundary; the fine-grained variants below are decoded from the +/// wire id. The `#[allow(dead_code)]` covers variants not yet emitted. #[allow(dead_code)] #[derive(Debug, Clone, Copy, Serialize)] #[serde(rename_all = "snake_case")] @@ -37,8 +32,7 @@ pub enum Category { Session, } -/// Terminal outcome of a measured operation. `Skipped` is part of the schema -/// for operations that never ran; v1 emits only `Success`/`Error`. +/// Terminal outcome of a measured operation. #[allow(dead_code)] #[derive(Debug, Clone, Copy, Serialize)] #[serde(rename_all = "snake_case")] @@ -76,8 +70,7 @@ impl JsonlSink { } fn append(&self, record: &HostMetricRecord) -> std::io::Result<()> { - let line = serde_json::to_string(record) - .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; + let line = serde_json::to_string(record).map_err(std::io::Error::other)?; let mut file = OpenOptions::new() .create(true) .append(true) @@ -99,8 +92,7 @@ pub struct MetricsRecorder { } impl MetricsRecorder { - /// Build a recorder from the environment. Returns an `Arc` so it can be - /// shared across connections. + /// Build a recorder from the environment (see the type's env-var list). pub fn from_env() -> Arc { let sink = std::env::var("METRICS_JSONL") .ok() @@ -149,13 +141,10 @@ impl MetricsRecorder { } } -/// Classify a raw inbound frame into a metric `(category, op)` by decoding its -/// wire discriminant and looking the method up in the core's wire table. -/// -/// `op` is the trait method name (e.g. `signing_sign_raw`); `category` is a -/// coarse bucket from the method's namespace, or `Subscription` for any -/// subscription frame. Falls back to `(Frame, "product_frame")` when the frame -/// does not decode or the id is unknown. +/// Op recorded for a frame we could not classify against the wire table. +/// This is an emitted-schema value consumed by downstream ingest. +const UNCLASSIFIED_OP: &str = "product_frame"; + /// A decoded inbound frame's metric identity. pub struct FrameClass { pub category: Category, @@ -163,6 +152,9 @@ pub struct FrameClass { pub request_id: String, } +/// Classify a raw inbound frame into `(category, op)` via the wire table. +/// Falls back to `(Frame, "product_frame")` when it does not decode or the id +/// is unknown. pub fn classify_frame(frame: &[u8]) -> FrameClass { use parity_scale_codec::Decode; use truapi_server::frame::ProtocolMessage; @@ -171,7 +163,7 @@ pub fn classify_frame(frame: &[u8]) -> FrameClass { let Ok(message) = ProtocolMessage::decode(&mut &*frame) else { return FrameClass { category: Category::Frame, - op: "product_frame".to_string(), + op: UNCLASSIFIED_OP.to_string(), request_id: String::new(), }; }; @@ -209,14 +201,13 @@ pub fn classify_frame(frame: &[u8]) -> FrameClass { } /// Decode a frame emitted back to the product and extract the true operation -/// outcome, keyed by `request_id`. Returns `Some` for response frames and for -/// error interrupts; `None` for streamed subscription items or undecodable -/// frames (whose outcome is left to the dispatch result). +/// outcome, keyed by `request_id`. Returns `Some` for response frames; `None` +/// for subscription frames or undecodable frames (whose outcome is left to the +/// dispatch result). /// -/// Request/response payloads are versioned as `[version_index][result_disc]..` -/// where `result_disc` is 0 for Ok and 1 for Err (truapi-server `frame.rs`), so -/// a domain error in the response is caught even though `receive_frame` still -/// returns Ok for it. +/// The response payload's result discriminant (`value[1]`: 0 = Ok, 1 = Err; +/// see truapi-server `frame.rs`) is the true outcome, so a domain error is +/// caught even though `receive_frame` returned Ok. pub fn response_outcome(frame: &[u8]) -> Option<(String, Outcome)> { use parity_scale_codec::Decode; use truapi_server::frame::ProtocolMessage; @@ -225,24 +216,22 @@ pub fn response_outcome(frame: &[u8]) -> Option<(String, Outcome)> { let message = ProtocolMessage::decode(&mut &*frame).ok()?; let id = message.payload.id; for entry in WIRE_TABLE { - match &entry.kind { - WireKind::Request(r) if r.response_id == id => { - let outcome = match message.payload.value.get(1) { - Some(1) => Outcome::Error, - _ => Outcome::Success, - }; - return Some((message.request_id, outcome)); - } - WireKind::Subscription(s) if s.interrupt_id == id => { - return Some((message.request_id, Outcome::Error)); - } - _ => {} + if let WireKind::Request(r) = &entry.kind + && r.response_id == id + { + let outcome = match message.payload.value.get(1) { + Some(1) => Outcome::Error, + _ => Outcome::Success, + }; + return Some((message.request_id, outcome)); } } None } /// Map a method name to a coarse metric category by its namespace prefix. +/// The buckets are provisional and unused in v1 (only `Frame` is emitted); +/// revisit them when the fine-grained category path goes live. fn category_for_method(method: &str) -> Category { match method.split('_').next().unwrap_or("") { "signing" | "entropy" | "statement" => Category::Signing, @@ -307,4 +296,42 @@ mod tests { let first: serde_json::Value = serde_json::from_str(body.lines().next().unwrap()).unwrap(); assert_eq!(first["op"], "product_frame"); } + + // Pins the versioned payload layout `response_outcome` reads (`value[1]`). + // Built with the real core encoders, so it trips if truapi-server changes + // the response wire shape (see the `TODO(shared-core-wire)` in frame.rs). + #[test] + fn response_outcome_reads_versioned_result_discriminant() { + use parity_scale_codec::Encode; + use truapi_server::frame::{ + Payload, ProtocolMessage, encode_versioned_err_payload, + encode_versioned_unit_ok_payload, + }; + use truapi_server::generated::wire_table::{WIRE_TABLE, WireKind}; + + let response_id = WIRE_TABLE + .iter() + .find_map(|entry| match &entry.kind { + WireKind::Request(r) => Some(r.response_id), + _ => None, + }) + .expect("wire table has at least one request method"); + let frame = |request_id: &str, value: Vec| { + ProtocolMessage { + request_id: request_id.to_string(), + payload: Payload { id: response_id, value }, + } + .encode() + }; + + let (id, outcome) = response_outcome(&frame("req-ok", encode_versioned_unit_ok_payload(1))) + .expect("response frame is classified"); + assert_eq!(id, "req-ok"); + assert!(matches!(outcome, Outcome::Success)); + + let (id, outcome) = response_outcome(&frame("req-err", encode_versioned_err_payload(0u32, 1))) + .expect("response frame is classified"); + assert_eq!(id, "req-err"); + assert!(matches!(outcome, Outcome::Error)); + } }