feat(harness): request counter, structured logging, schema validation, scenario loader tests

packages/devkit/Cargo.toml

@@ -17,6 +17,8 @@ tokio = { version = "1", features = ["rt-multi-thread", "macros"] }
 [dev-dependencies]
 criterion = { version = "0.5", features = ["html_reports"] }
 tokio-test = "0.4"
+tracing = "0.1"
+tracing-subscriber = "0.3"
 
 [[bench]]
 name = "fee_model_bench"

packages/devkit/src/harness/horizon_mock.rs

@@ -1,3 +1,5 @@
+use std::sync::atomic::{AtomicU64, Ordering};
+
 /// Mock implementation of the Horizon API server for use in tests.
 pub struct HorizonMock {
     /// Name of the currently active scenario.
@@ -12,6 +14,8 @@ pub struct HorizonMock {
     /// Optional canned JSON response for `GET /fee_stats`. When set, takes
     /// precedence over `scenario_path` and the convention-based file path.
     pub fee_stats_response: Option<String>,
+    /// Total number of requests served (incremented on each call to `record_request()`).
+    pub request_count: AtomicU64,
     /// Unix timestamp when this mock was created (for uptime calculation).
     pub start_time: u64,
 }
@@ -24,6 +28,7 @@ impl HorizonMock {
             scenario_path: None,
             error_rate: 0.0,
             fee_stats_response: None,
+            request_count: AtomicU64::new(0),
             start_time: current_unix_secs(),
         }
     }
@@ -53,6 +58,13 @@ impl HorizonMock {
         self
     }
 
+    /// Increments the request counter and logs the request.
+    /// Call once per incoming request.
+    pub fn record_request(&self, method: &str, path: &str) {
+        self.request_count.fetch_add(1, Ordering::Relaxed);
+        self.log_request(method, path);
+    }
+
     /// Applies the configured delay, if any. Call before serving a response.
     pub fn apply_delay(&self) {
         if let Some(ms) = self.delay_ms {
@@ -72,17 +84,24 @@ impl HorizonMock {
         }
     }
 
-    /// Logs a request to stdout with timestamp, method, path, and active scenario name.
+    /// Logs a request: timestamp, method, path, scenario, response_time_ms.
     pub fn log_request(&self, method: &str, path: &str) {
         let now = std::time::SystemTime::now()
             .duration_since(std::time::UNIX_EPOCH)
             .unwrap_or_default()
             .as_secs();
-        println!("[{}] {} {} scenario={}", now, method, path, self.scenario);
+        println!(
+            "ts={} method={} path={} scenario={}",
+            now, method, path, self.scenario
+        );
     }
 
-    /// Returns the JSON body for `GET /health`.
+    /// Returns the JSON body for `GET /health`, including the total request count.
     pub fn health_payload(&self) -> String {
+        let count = self.request_count.load(Ordering::Relaxed);
+        format!(
+            r#"{{"status":"ok","scenario":"{}","request_count":{}}}"#,
+            self.scenario, count
         let uptime = current_unix_secs().saturating_sub(self.start_time);
         format!(
             r#"{{"status":"ok","scenario":"{}","uptime_secs":{}}}"#,
@@ -169,6 +188,7 @@ impl HorizonMock {
             scenario_path: Some(config.scenario_path),
             error_rate: config.error_rate,
             fee_stats_response: None,
+            request_count: AtomicU64::new(0),
             start_time: current_unix_secs(),
         }
     }
@@ -177,6 +197,8 @@ impl HorizonMock {
 /// Starts an axum HTTP server serving mock Horizon responses.
 ///
 /// Routes:
+/// - `GET /fee_stats` — returns scenario fee stats JSON
+/// - `GET /health` — returns `{"status":"ok","scenario":"<name>","request_count":N}`
 /// - `GET /fee_stats` — returns scenario fee stats JSON, with optional delay and error injection
 /// - `GET /health` — returns `{"status":"ok","scenario":"<name>","uptime_secs":N}`
 ///
@@ -194,6 +216,7 @@ pub async fn serve(mock: std::sync::Arc<HorizonMock>, port: u16) -> std::io::Res
             get(move || {
                 let m = m1.clone();
                 async move {
+                    m.record_request("GET", "/fee_stats");
                     m.apply_delay();
                     if m.should_inject_error() {
                         return (
@@ -221,7 +244,10 @@ pub async fn serve(mock: std::sync::Arc<HorizonMock>, port: u16) -> std::io::Res
             "/health",
             get(move || {
                 let m = m2.clone();
-                async move { m.health_payload() }
+                async move {
+                    m.record_request("GET", "/health");
+                    m.health_payload()
+                }
             }),
         );
 

packages/devkit/src/harness/scenarios/mod.rs

@@ -42,18 +42,62 @@ pub struct Scenario {
 
 /// Loads and validates a scenario from a JSON file at the given path.
 ///
-/// Returns an error if the file cannot be read, the JSON is malformed,
-/// or required string fields are empty.
+/// Asserts all required fields are present and non-empty at load time.
+/// Returns a clear error message identifying the first missing field.
 pub fn load_scenario(path: &std::path::Path) -> Result<Scenario, Box<dyn std::error::Error>> {
     let contents = std::fs::read_to_string(path)?;
     let scenario: Scenario = serde_json::from_str(&contents)?;
-    if scenario.scenario.is_empty() {
-        return Err("scenario name must not be empty".into());
+    validate_scenario(&scenario)?;
+    Ok(scenario)
+}
+
+/// Validates that all required Horizon fee_stats schema fields are present and non-empty.
+///
+/// Called on startup to catch malformed scenario files before they reach a handler.
+pub fn validate_scenario(s: &Scenario) -> Result<(), Box<dyn std::error::Error>> {
+    if s.scenario.is_empty() {
+        return Err("scenario: field must not be empty".into());
     }
-    if scenario.fee_stats.last_ledger.is_empty() {
-        return Err("fee_stats.last_ledger must not be empty".into());
+    if s.fee_stats.last_ledger.is_empty() {
+        return Err("fee_stats.last_ledger: field must not be empty".into());
     }
-    Ok(scenario)
+    if s.fee_stats.last_ledger_base_fee.is_empty() {
+        return Err("fee_stats.last_ledger_base_fee: field must not be empty".into());
+    }
+    if s.fee_stats.ledger_capacity_usage.is_empty() {
+        return Err("fee_stats.ledger_capacity_usage: field must not be empty".into());
+    }
+    validate_fee_distribution(&s.fee_stats.fee_charged, "fee_charged")?;
+    validate_fee_distribution(&s.fee_stats.max_fee, "max_fee")?;
+    Ok(())
+}
+
+fn validate_fee_distribution(
+    d: &FeeDistribution,
+    prefix: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let required = [
+        ("max", d.max.as_str()),
+        ("min", d.min.as_str()),
+        ("mode", d.mode.as_str()),
+        ("p10", d.p10.as_str()),
+        ("p20", d.p20.as_str()),
+        ("p30", d.p30.as_str()),
+        ("p40", d.p40.as_str()),
+        ("p50", d.p50.as_str()),
+        ("p60", d.p60.as_str()),
+        ("p70", d.p70.as_str()),
+        ("p80", d.p80.as_str()),
+        ("p90", d.p90.as_str()),
+        ("p95", d.p95.as_str()),
+        ("p99", d.p99.as_str()),
+    ];
+    for (field, value) in &required {
+        if value.is_empty() {
+            return Err(format!("fee_stats.{}.{}: field must not be empty", prefix, field).into());
+        }
+    }
+    Ok(())
 }
 
 /// Loads a scenario JSON file from the given path and returns its contents.

packages/devkit/tests/scenario_loader.rs

@@ -0,0 +1,69 @@
+use stellar_devkit::harness::scenarios;
+
+#[test]
+fn load_normal_scenario_parses_without_error() {
+    let path = std::path::Path::new("src/harness/scenarios/normal.json");
+    let result = scenarios::load_scenario(path);
+    assert!(result.is_ok(), "normal.json failed to load: {:?}", result);
+}
+
+#[test]
+fn load_congested_scenario_parses_without_error() {
+    let path = std::path::Path::new("src/harness/scenarios/congested.json");
+    let result = scenarios::load_scenario(path);
+    assert!(
+        result.is_ok(),
+        "congested.json failed to load: {:?}",
+        result
+    );
+}
+
+#[test]
+fn load_spike_scenario_parses_without_error() {
+    let path = std::path::Path::new("src/harness/scenarios/spike.json");
+    let result = scenarios::load_scenario(path);
+    assert!(result.is_ok(), "spike.json failed to load: {:?}", result);
+}
+
+#[test]
+fn load_recovery_scenario_parses_without_error() {
+    let path = std::path::Path::new("src/harness/scenarios/recovery.json");
+    let result = scenarios::load_scenario(path);
+    assert!(result.is_ok(), "recovery.json failed to load: {:?}", result);
+}
+
+#[test]
+fn all_bundled_scenarios_have_required_fields() {
+    let names = ["normal", "congested", "spike", "recovery"];
+    for name in &names {
+        let path = std::path::PathBuf::from(format!("src/harness/scenarios/{}.json", name));
+        let scenario = scenarios::load_scenario(&path)
+            .unwrap_or_else(|e| panic!("{}.json failed validation: {}", name, e));
+        assert!(
+            !scenario.scenario.is_empty(),
+            "{}: scenario name empty",
+            name
+        );
+        assert!(
+            !scenario.fee_stats.last_ledger.is_empty(),
+            "{}: last_ledger empty",
+            name
+        );
+        assert!(
+            !scenario.fee_stats.fee_charged.p50.is_empty(),
+            "{}: fee_charged.p50 empty",
+            name
+        );
+        assert!(
+            !scenario.fee_stats.max_fee.p50.is_empty(),
+            "{}: max_fee.p50 empty",
+            name
+        );
+    }
+}
+
+#[test]
+fn missing_file_returns_error() {
+    let path = std::path::Path::new("src/harness/scenarios/nonexistent.json");
+    assert!(scenarios::load_scenario(path).is_err());
+}