sim/tests/cluster.rs

//! Runs the real binary as a gateway with two shard processes and checks that the UI sees one
//! world spanning both shards.

use std::io::{Read, Write};
use std::net::{TcpListener, TcpStream};
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};

struct SimProcess(Child);

impl Drop for SimProcess {
    fn drop(&mut self) {
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}

impl SimProcess {
    fn wait_for_failure(&mut self) -> String {
        let deadline = Instant::now() + Duration::from_secs(30);
        let status = loop {
            if let Some(status) = self.0.try_wait().unwrap() {
                break status;
            }
            assert!(
                Instant::now() < deadline,
                "the shard stayed alive after its simulation thread panicked"
            );
            std::thread::sleep(Duration::from_millis(50));
        };
        let mut stderr = String::new();
        self.0
            .stderr
            .take()
            .unwrap()
            .read_to_string(&mut stderr)
            .unwrap();
        assert_eq!(status.code(), Some(1), "{stderr}");
        assert!(
            stderr.contains("shard simulation thread panicked; terminating shard"),
            "{stderr}"
        );
        stderr
    }
}

fn shard_command() -> Command {
    let mut command = Command::new(env!("CARGO_BIN_EXE_sim"));
    command
        .args([
            "shard",
            "--index",
            "0",
            "--count",
            "1",
            "--database",
            "sqlite::memory:",
        ])
        .env("SIM_REMOTING_BIND", "127.0.0.1:0")
        .env_remove("SIM_REMOTING_ADVERTISE")
        .env(
            "SIM_CONFIG",
            concat!(env!("CARGO_MANIFEST_DIR"), "/config.rn"),
        )
        .env("RUST_LOG", "sim=info,picard=off")
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    command
}

#[test]
fn shard_exits_when_simulation_startup_panics() {
    let mut shard = SimProcess(
        shard_command()
            .env("SIM_REMOTING_BIND", "invalid-address")
            .spawn()
            .unwrap(),
    );
    let stderr = shard.wait_for_failure();
    assert!(
        stderr.contains("failed to start shard remoting"),
        "{stderr}"
    );
}

#[cfg(feature = "console")]
#[test]
fn shard_exits_when_actor_replies_time_out() {
    use std::io::{BufRead, BufReader};
    use tungstenite::Message;

    let mut shard = SimProcess(shard_command().spawn().unwrap());
    let stdout = shard.0.stdout.take().unwrap();
    let (listening, address) = std::sync::mpsc::channel();
    let reader = std::thread::spawn(move || {
        let mut lines = BufReader::new(stdout).lines();
        for line in lines.by_ref() {
            let line = line.unwrap();
            if let Some(address) = line.strip_prefix("SHARD_LISTENING ") {
                listening.send(address.to_owned()).unwrap();
                break;
            }
        }
        for line in lines {
            line.unwrap();
        }
    });
    let address = address.recv_timeout(Duration::from_secs(30)).unwrap();
    let (mut socket, _) = tungstenite::connect(format!("ws://{address}/control")).unwrap();
    if let tungstenite::stream::MaybeTlsStream::Plain(stream) = socket.get_ref() {
        stream
            .set_read_timeout(Some(Duration::from_secs(30)))
            .unwrap();
    }

    socket
        .send(Message::Text(
            serde_json::json!({
                "Console": serde_json::json!({
                    "id": 1,
                    "op": "pause",
                    "path": "/user/world-inbox"
                }).to_string()
            })
            .to_string()
            .into(),
        ))
        .unwrap();
    let response = socket.read().unwrap().into_text().unwrap();
    let reply: serde_json::Value = serde_json::from_str(&response).unwrap();
    assert!(
        reply.get("Failed").is_none(),
        "unexpected shard failure: {}",
        shard.wait_for_failure()
    );
    let console: serde_json::Value = serde_json::from_str(
        reply["Console"]
            .as_str()
            .unwrap_or_else(|| panic!("expected a console reply: {reply}")),
    )
    .unwrap();
    assert!(console.get("error").is_none(), "{console}");

    socket
        .send(Message::Text(
            serde_json::json!({"View": {"terrain": false}})
                .to_string()
                .into(),
        ))
        .unwrap();
    let stderr = shard.wait_for_failure();
    reader.join().unwrap();
    assert!(
        stderr.contains("Picard entity actors did not answer request")
            && stderr.contains("timed out waiting on channel"),
        "{stderr}"
    );
}

fn get(address: &str, path: &str) -> Option<serde_json::Value> {
    let mut stream = TcpStream::connect(address).ok()?;
    stream.set_read_timeout(Some(Duration::from_secs(5))).ok()?;
    write!(
        stream,
        "GET {path} HTTP/1.1\r\nHost: {address}\r\nConnection: close\r\n\r\n"
    )
    .ok()?;
    let mut response = String::new();
    stream.read_to_string(&mut response).ok()?;
    let (_, body) = response.split_once("\r\n\r\n")?;
    serde_json::from_str(body).ok()
}

#[test]
fn gateway_presents_the_shards_as_one_world() {
    let address = {
        let probe = TcpListener::bind("127.0.0.1:0").unwrap();
        probe.local_addr().unwrap().to_string()
    };
    let directory = std::env::temp_dir().join(format!("sim-cluster-test-{}", std::process::id()));
    std::fs::create_dir_all(&directory).unwrap();
    let database = format!(
        "sqlite://{}/shard-{{shard}}.db?mode=rwc",
        directory.display()
    );
    let _gateway = SimProcess(
        Command::new(env!("CARGO_BIN_EXE_sim"))
            .arg("5")
            .env("SIM_SHARDS", "2")
            .env("SIM_ADDR", &address)
            .env("SHARD_DATABASE_URL", &database)
            .env(
                "SIM_CONFIG",
                concat!(env!("CARGO_MANIFEST_DIR"), "/config.rn"),
            )
            .env("RUST_LOG", "warn")
            .stdout(Stdio::null())
            .spawn()
            .unwrap(),
    );

    let deadline = Instant::now() + Duration::from_secs(120);
    let mut first_tick = None;
    loop {
        assert!(Instant::now() < deadline, "the cluster never got going");
        std::thread::sleep(Duration::from_millis(500));
        let Some(state) = get(&address, "/api/state") else {
            continue;
        };
        let Some(tick) = state["tick"].as_u64() else {
            continue;
        };
        let start = *first_tick.get_or_insert(tick);
        let shards = state["shards"].as_array().unwrap();
        assert_eq!(shards.len(), 2);
        let border = shards[1][0].as_u64().unwrap();
        let people = state["people"].as_array().unwrap();
        let west = people.iter().any(|p| p[1].as_u64() < Some(border));
        let east = people.iter().any(|p| p[1].as_u64() >= Some(border));
        if tick > start + 5 && west && east {
            assert_eq!(state["homes"].as_array().unwrap().len(), 2);
            break;
        }
    }
    let terrain = get(&address, "/api/terrain").unwrap();
    assert!(terrain["w"].as_u64().unwrap() > 0);
    let _ = std::fs::remove_dir_all(&directory);
}