feat: headless-host metrics and stress fleet

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"
@@ -30,6 +31,7 @@ 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"
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "io-util", "io-std", "net", "time", "signal", "process"] }
 tokio-stream = { version = "0.1", features = ["sync"] }

rust/crates/truapi-host-cli/e2e/fleet.sh

@@ -0,0 +1,92 @@
+#!/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)
+#   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-<epoch>)
+#   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.
+# 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="${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="$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)"
+    [ -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"; 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" || 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"
+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" || true; done
+echo "fleet complete -> $METRICS_JSONL"

rust/crates/truapi-host-cli/src/frame_server.rs

@@ -5,8 +5,10 @@
 //! 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};
 use futures_util::{SinkExt, StreamExt};
@@ -17,13 +19,26 @@ use tokio_tungstenite::tungstenite::Message;
 use tracing::{debug, info};
 use truapi_server::{FrameSink, PairingHostRuntime, ProductContext};
 
-/// Frame sink that writes each outgoing protocol frame as one binary message.
+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`.
+type OutcomeMap = Arc<Mutex<HashMap<String, Outcome>>>;
+
+/// 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<Message>,
+    outcomes: OutcomeMap,
 }
 
 impl FrameSink for WsFrameSink {
     fn emit_frame(&self, frame: Vec<u8>) {
+        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));
     }
 }
@@ -46,15 +61,17 @@ pub async fn accept_loop(
     runtime: Arc<PairingHostRuntime>,
     product_id: String,
     listener: TcpListener,
+    metrics: Arc<MetricsRecorder>,
 ) -> Result<()> {
     let bound = listener.local_addr()?;
     info!(%bound, %product_id, "product frame server listening");
     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 +82,7 @@ async fn serve_connection(
     runtime: Arc<PairingHostRuntime>,
     product_id: String,
     stream: TcpStream,
+    metrics: Arc<MetricsRecorder>,
 ) -> Result<()> {
     let ws = accept_async(stream).await?;
     let (mut write, mut read) = ws.split();
@@ -80,23 +98,31 @@ 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)) => {
-                if let Err(err) = product_runtime.receive_frame(bytes.to_vec()).await {
+                let class = classify_frame(&bytes);
+                let started = Instant::now();
+                let result = product_runtime.receive_frame(bytes.to_vec()).await;
+                record_frame(&metrics, &class, started, &result, &outcomes);
+                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 bytes = text.as_bytes().to_vec();
+                let class = classify_frame(&bytes);
+                let started = Instant::now();
+                let result = product_runtime.receive_frame(bytes).await;
+                record_frame(&metrics, &class, started, &result, &outcomes);
+                if let Err(err) = result {
                     debug!(%err, "product runtime rejected text frame");
                 }
             }
@@ -110,3 +136,30 @@ async fn serve_connection(
     let _ = writer.await;
     Ok(())
 }
+
+/// 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<E: std::fmt::Display>(
+    metrics: &MetricsRecorder,
+    class: &FrameClass,
+    started: Instant,
+    result: &Result<(), E>,
+    outcomes: &OutcomeMap,
+) {
+    let latency_ms = started.elapsed().as_secs_f64() * 1000.0;
+    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_CLASS.to_string())),
+            Some(other) => (other, None),
+            None => (Outcome::Success, None),
+        },
+    };
+    metrics.record(class.category, &class.op, latency_ms, outcome, error_class);
+}