settled-reach/server/tests/perf_bench.rs

//! Performance benchmarks: tick timing and memory usage
//!
//! Run with: cargo test --release --test perf_bench -- --ignored --nocapture
//! Output: PERF_RESULT:{json} lines for `reach dev perf` to parse.
//!
//! Tick budget target: 100ms (D-026)

use bevy_app::prelude::*;
use std::io::{BufReader, BufWriter};
use std::net::{TcpListener, TcpStream};
use std::thread;
use std::time::{Duration, Instant};

use settled_reach_server::bridge::framing::{read_framed, write_framed};
use settled_reach_server::bridge::tcp::TcpBridge;
use settled_reach_server::bridge::types::*;
use settled_reach_server::bridge::{BridgePlugin, BridgeResource};
use settled_reach_server::knowledge::registry::EntityRegistry;
use settled_reach_server::knowledge::KnowledgeGraph;
use settled_reach_server::perception::cognitive_delay::CognitiveDelay;
use settled_reach_server::perception::vision_cone::Facing;
use settled_reach_server::simulation::interaction::NearbyInteractionBuffer;
use settled_reach_server::simulation::line_pool::{LinePoolIndex, LinePoolIndexResource};
use settled_reach_server::simulation::listening::ListeningFocus;
use settled_reach_server::simulation::monologue::{
    MonologueBuffer, MonologueState, SprintAnomalyQueue,
};
use settled_reach_server::simulation::movement::{PlayerCharacter, TilePosition, WalkabilityMap};
use settled_reach_server::simulation::stance::{MovementProfile, PlayerMoveCooldown};
use settled_reach_server::simulation::SimulationPlugin;

const WARMUP_TICKS: usize = 5;
const MEASURE_TICKS: usize = 50;
const TOTAL_TICKS: usize = WARMUP_TICKS + MEASURE_TICKS;

fn read_rss_kb() -> Option<u64> {
    std::fs::read_to_string("/proc/self/status")
        .ok()
        .and_then(|s| {
            s.lines()
                .find(|l| l.starts_with("VmRSS:"))
                .and_then(|l| l.split_whitespace().nth(1))
                .and_then(|v| v.parse().ok())
        })
}

/// Full plugin stack tick benchmark with real content.
///
/// Boots the server with production content, runs WARMUP_TICKS to stabilize,
/// then measures MEASURE_TICKS of app.update() wall-clock time. Reports entity
/// counts from observer snapshots and process RSS.
///
/// Protocol contract: BridgePlugin uses non-blocking receive (WouldBlock →
/// empty input vec), so the server always advances even if the client hasn't
/// sent input yet. The server sends a snapshot each tick; the client blocks
/// on read until one arrives, then responds with (empty) input. No deadlock
/// possible — see TcpBridge::receive_inputs and send_snapshot in bridge/tcp.rs.
#[test]
#[ignore]
fn perf_tick_timing() {
    let listener = TcpListener::bind("127.0.0.1:0").expect("bind listener");
    let server_addr = listener.local_addr().expect("get local addr");

    // Server thread: full plugin stack, timed ticks
    let server_handle = thread::spawn(move || -> Vec<Duration> {
        let bridge = TcpBridge::accept_on(listener).expect("accept connection");

        let mut app = App::new();
        app.add_plugins(SimulationPlugin { seed: 0 });
        app.add_plugins(BridgePlugin);
        app.add_plugins(settled_reach_server::knowledge::KnowledgePlugin);
        app.add_plugins(settled_reach_server::npc::NpcPlugin);
        app.insert_resource(LinePoolIndexResource(LinePoolIndex::default()));
        app.insert_resource(BridgeResource::new(bridge));
        app.insert_resource(WalkabilityMap::new(32, 32, 1));

        let profile = MovementProfile::smuggler();
        let mut registry = EntityRegistry::new(0);
        let player = app
            .world_mut()
            .spawn((
                PlayerCharacter,
                TilePosition::new(16, 16, 0),
                Facing::default(),
                KnowledgeGraph::new(),
                NearbyInteractionBuffer::default(),
                MonologueState::default(),
                MonologueBuffer::default(),
                SprintAnomalyQueue::default(),
                CognitiveDelay::default(),
                ListeningFocus::new(TilePosition::new(16, 16, 0)),
                profile,
                profile.initial_stance(),
                PlayerMoveCooldown::default(),
            ))
            .id();
        registry.register(player);
        app.insert_resource(registry);

        let mut timings = Vec::with_capacity(TOTAL_TICKS);
        for _ in 0..TOTAL_TICKS {
            let start = Instant::now();
            app.update();
            timings.push(start.elapsed());
        }
        timings
    });

    // Client: pump protocol — read snapshots, send empty inputs
    let stream = TcpStream::connect(server_addr).expect("client connect");
    stream
        .set_read_timeout(Some(Duration::from_secs(30)))
        .expect("set read timeout");
    let mut reader = BufReader::new(stream.try_clone().expect("clone for reader"));
    let mut writer = BufWriter::new(stream);

    let mut entity_counts: Vec<usize> = Vec::with_capacity(TOTAL_TICKS);
    for tick in 0..TOTAL_TICKS {
        // Server may close connection after its last tick — handle gracefully
        let payload = match read_framed(&mut reader) {
            Ok(Some(p)) => p,
            Ok(None) | Err(_) => break,
        };

        let snapshot: ObserverSnapshot = rmp_serde::from_slice(&payload)
            .unwrap_or_else(|e| panic!("deserialization error at tick {}: {}", tick, e));

        entity_counts.push(snapshot.entities.len());

        let empty: Vec<PlayerInput> = vec![];
        let input_payload = rmp_serde::to_vec(&empty).expect("serialize empty input");
        if write_framed(&mut writer, &input_payload).is_err() {
            break;
        }
    }

    // Must have received enough measured snapshots for meaningful results.
    // Require all warmup ticks plus at least half the measurement window.
    let min_snapshots = WARMUP_TICKS + MEASURE_TICKS / 2;
    assert!(
        entity_counts.len() >= min_snapshots,
        "Only received {} snapshots, need at least {} ({} warmup + {} measured)",
        entity_counts.len(),
        min_snapshots,
        WARMUP_TICKS,
        MEASURE_TICKS / 2
    );

    drop(reader);
    drop(writer);

    let timings = server_handle
        .join()
        .expect("server thread panicked during tick benchmark");

    // Analyze measured ticks (skip warmup)
    let measured_us: Vec<u64> = timings
        .iter()
        .skip(WARMUP_TICKS)
        .map(|d| d.as_micros() as u64)
        .collect();

    let min = *measured_us.iter().min().unwrap();
    let max = *measured_us.iter().max().unwrap();
    let sum: u64 = measured_us.iter().sum();
    let mean = sum / measured_us.len() as u64;

    let mut sorted = measured_us.clone();
    sorted.sort();
    // Nearest-rank p95: index = floor(0.95 * (N-1)) for 0-based indexing.
    let p95_idx = ((sorted.len() - 1) as f64 * 0.95).floor() as usize;
    let p95 = sorted[p95_idx.min(sorted.len() - 1)];

    let entity_counts_measured: Vec<usize> =
        entity_counts.iter().skip(WARMUP_TICKS).copied().collect();
    let avg_entities =
        entity_counts_measured.iter().sum::<usize>() / entity_counts_measured.len().max(1);
    let max_entities = entity_counts_measured.iter().max().copied().unwrap_or(0);

    let rss_kb = read_rss_kb();

    let result = serde_json::json!({
        "tick_timing": {
            "warmup_ticks": WARMUP_TICKS,
            "measured_ticks": measured_us.len(),
            "min_us": min,
            "max_us": max,
            "mean_us": mean,
            "p95_us": p95,
            "all_us": measured_us,
        },
        "entities": {
            "avg_per_snapshot": avg_entities,
            "max_per_snapshot": max_entities,
        },
        "memory": {
            "rss_kb": rss_kb,
        },
    });

    println!("PERF_RESULT:{}", serde_json::to_string(&result).unwrap());
}