smarm/tests/cluster_two_node.rs

//! RFC 010 c4 — two-node harness smoke tests.
//!
//! Roadmap: "spawn two, handshake-less connect, both exit clean." The
//! listener node binds port 0 and announces its concrete address; the
//! dialer connects raw (no Hello — c5 doesn't exist yet), pushes one
//! Heartbeat through the real framed codec, and closes. Assertions are on
//! protocol-visible lines only. Flake budget: see tests/common/mod.rs.
#![cfg(feature = "cluster")]

mod common;

use common::{maybe_child, spawn_node};
use smarm::cluster::envelope::Frame;
use smarm::cluster::transport::tcp::TcpTransport;
use smarm::cluster::transport::{FramedConn, Transport};

const ROLES: &[(&str, fn())] = &[
    ("listener", role_listener),
    ("dialer", role_dialer),
    ("hang", role_hang),
    ("fail", role_fail),
];

fn role_listener() {
    smarm::run(|| {
        let mut l = TcpTransport.listen("127.0.0.1:0").unwrap();
        println!("LISTENING {}", l.local_addr());
        let mut fc = FramedConn::new(l.accept().unwrap());
        match fc.recv() {
            Ok(Some(Frame::Heartbeat)) => println!("RECV heartbeat"),
            other => {
                println!("RECV unexpected: {other:?}");
                std::process::exit(3);
            }
        }
        match fc.recv() {
            Ok(None) => println!("PEER-CLOSED clean"),
            other => {
                println!("PEER-CLOSED unexpected: {other:?}");
                std::process::exit(3);
            }
        }
    });
    println!("EXIT ok");
}

fn role_dialer() {
    let addr = std::env::var("SMARM_PEER_ADDR").expect("SMARM_PEER_ADDR not set");
    smarm::run(move || {
        let mut fc = FramedConn::new(TcpTransport.dial(&addr).unwrap());
        fc.send(&Frame::Heartbeat).unwrap();
        fc.close();
        println!("SENT heartbeat");
    });
    println!("EXIT ok");
}

fn role_hang() {
    println!("HANGING");
    loop {
        std::thread::sleep(std::time::Duration::from_secs(3600));
    }
}

fn role_fail() {
    std::process::exit(7);
}

/// The roadmap smoke test: two real processes, raw transport connect, one
/// frame across, clean close observed on both sides, both exit 0.
#[test]
fn two_nodes_connect_and_exit_clean() {
    maybe_child(ROLES);
    let mut listener = spawn_node("listener", &[]);
    let addr = listener.wait_listening();
    let mut dialer = spawn_node("dialer", &[("SMARM_PEER_ADDR", &addr)]);
    dialer.wait_line("SENT heartbeat", |l| l == "SENT heartbeat");
    listener.wait_line("RECV heartbeat", |l| l == "RECV heartbeat");
    listener.wait_line("clean peer close", |l| l == "PEER-CLOSED clean");
    dialer.wait_exit_ok();
    listener.wait_exit_ok();
}

/// Reap guarantee: dropping a Node kills a hung child — no orphan survives
/// a panicking test.
#[test]
fn drop_reaps_hung_node() {
    maybe_child(ROLES);
    let mut node = spawn_node("hang", &[]);
    node.wait_line("HANGING", |l| l == "HANGING");
    let pid = node.pid().expect("live child has a pid") as libc::pid_t;
    drop(node);
    // After Drop's kill+wait the pid is fully reaped: signalling it fails
    // with ESRCH (pid-reuse in this instant is not a realistic race).
    let rc = unsafe { libc::kill(pid, 0) };
    assert_eq!(rc, -1, "process still signallable after Drop");
    let errno = std::io::Error::last_os_error().raw_os_error();
    assert_eq!(errno, Some(libc::ESRCH), "expected ESRCH, got {errno:?}");
}

/// Nonzero child exits surface as statuses, not hangs or panics.
#[test]
fn nonzero_exit_is_reported() {
    maybe_child(ROLES);
    let mut node = spawn_node("fail", &[]);
    let status = node.wait_exit();
    assert_eq!(status.code(), Some(7));
}