melin-server-runtime 0.22.0

//! End-to-end contract test for the `ram` ack policy: acks gate on a
//! second node's *in-memory* receipt only — no disk anywhere on the ack
//! path — so every acked event must survive the death of the primary
//! via failover.
//!
//! A primary and two replicas (counter app, `--ack-policy ram`, raft +
//! `--raft-auto-promote` on all three) over real TCP and real Ed25519
//! auth. A client sends increments and collects acks — proof the
//! RAM-quorum gate opens on live replica cursors. The primary is then
//! killed; exactly one replica must auto-promote (exercising the
//! journaled policy end-to-end: the primary journaled `ram` as its
//! lineage began, the replicas hold that change, and the promotion policy
//! must accept it on the same grounds as `disk+ram`). The operator then
//! swaps the promoted node to `disk` over the admin endpoint — the documented
//! post-failover workflow, since its own gate is unsatisfiable with no
//! replicas attached — and a fresh client connection must read back the
//! full acked total: nothing the client was told about died with the
//! primary.
//!
//! Every node is also given its own startup events (see
//! [`startup_events`]), weighted so the final total says which node
//! journaled what: the genesis primary's genesis and `on_primary` before
//! the first client, no replica's own while it follows, and the promoted
//! node's `on_primary` before it serves.
//!
//! The final assertion is the reproducer that surfaced the promotion
//! peer-tip veto: without it, the election could land on the replica
//! that *lacks* the last acked event (the ack quorum is the primary
//! plus the faster replica, and the journal-tip vote filter is
//! best-effort by design), and this test failed ~40% of runs with an
//! acked increment missing after failover. It now passes because a
//! behind leader refuses to promote while a live peer advertises a
//! higher tip, and the caught-up peer campaigns to take over — see
//! `raft_promotion::auto_promotion_decision` and
//! `challenger_should_campaign`.
//!
//! Nodes are started through `melin-test-node`: on kernel TCP by default,
//! on DPDK with this crate's `dpdk` feature, under
//! `scripts/dpdk/netns-runner.sh` (see
//! `docs/internal/dpdk-testing.md`). On DPDK the winner serves
//! as a DPDK primary, and the primary is cut off the network before it is
//! stopped, as a crashed host would be: its replicas hear nothing of its
//! going, so failover completes only if they notice the silence
//! themselves (the replica end of a DPDK link's liveness deadline).

use std::io::Read;
use std::net::{SocketAddr, TcpStream};
use std::time::{Duration, Instant};

use base64::Engine;
use counter_server::{
    Counter, CounterEvent, GET_VALUE_REQUEST, KIND_RESP_ACK, KIND_RESP_VALUE, RequestDecoder,
    ResponseEncoder, increment_request,
};
use melin_client::{Connection, SigningKey};
use melin_server_runtime::StartupEvents;
use melin_server_runtime::ack_policy::AckPolicy;
use melin_server_runtime::layout::PipelineCores;
use melin_server_runtime::server::ServerConfig;
use melin_test_node::Addrs;
use melin_test_node::probe::{admin_until_ok, http_metrics, wait_for_gauge};
use melin_transport_core::test_ports::free_addr;
use melin_wire_protocol::control_codec::TAG_CHALLENGE;
use serial_test::serial;

/// Port range this file owns for `free_addr` (10000..15000);
/// `melin-raft`'s election tests own 15000..20000, `raft_failover.rs`
/// 20000..25000, `raft_smoke.rs` 25000..30000, and the notary example's
/// `round_trip.rs` 5000..10000. See `test_ports::free_addr` for the
/// scheme.
const PORT_BASE: u16 = 10_000;

/// Node `i`'s startup events: a genesis increment of `(i + 1) * GENESIS`
/// and an `on_primary` increment of `(i + 1) * ON_PRIMARY`. The weights
/// keep every contribution in its own decimal digits, apart from the
/// client's increments, so a total names the node behind each part.
fn startup_events(i: usize) -> StartupEvents<CounterEvent> {
    let weight = i as u64 + 1;
    StartupEvents {
        genesis: vec![CounterEvent::Increment {
            amount: weight * GENESIS,
        }],
        on_primary: vec![CounterEvent::Increment {
            amount: weight * ON_PRIMARY,
        }],
    }
}

const GENESIS: u64 = 1_000_000;
const ON_PRIMARY: u64 = 1_000;

// ---------------------------------------------------------------------------
// Client helpers, on `melin-client`
// ---------------------------------------------------------------------------

/// Connect and authenticate a client, retrying until the node serves:
/// the kernel backlog accepts the TCP SYN before the accept loop runs,
/// so a successful `connect` alone proves nothing, and `connect_by`
/// retries what time can fix.
fn connect_authenticated(addr: SocketAddr, key: &SigningKey) -> Connection {
    let mut conn = Connection::connect_by(addr, key, Instant::now() + Duration::from_secs(30))
        .unwrap_or_else(|e| panic!("timed out waiting for a serving node at {addr}: {e}"));
    // Generous post-auth timeout: acks under `ram` wait on a live replica
    // round-trip, and CI machines schedule these three servers on shared
    // cores.
    conn.set_read_timeout(Duration::from_secs(30))
        .expect("set timeout");
    conn
}

/// Whether a client connect to `addr` is answered with the auth
/// Challenge — the definitive "this node serves as primary" signal (a
/// replica's client listener never accepts). Only the tag is looked at:
/// any challenge counts, whatever protocol it speaks.
fn serves_clients(addr: SocketAddr, patience: Duration) -> bool {
    let Ok(mut stream) = TcpStream::connect_timeout(&addr, Duration::from_millis(300)) else {
        return false;
    };
    if stream.set_read_timeout(Some(patience)).is_err() {
        return false;
    }
    let mut head = [0u8; 5];
    stream.read_exact(&mut head).is_ok() && head[4] == TAG_CHALLENGE
}

// ---------------------------------------------------------------------------
// Cluster scaffolding
// ---------------------------------------------------------------------------

struct NodeSetup {
    key: SigningKey,
    /// Client and replication addresses.
    addrs: Addrs,
    raft_addr: SocketAddr,
    health_addr: SocketAddr,
    admin_addr: SocketAddr,
}

/// Deadline diagnostics: the servers install no tracing subscriber
/// under test, so the panic message is the only record of what each
/// node believed when a wait timed out.
fn cluster_summary(nodes: &[NodeSetup]) -> String {
    nodes
        .iter()
        .enumerate()
        .map(|(i, n)| {
            let gauges = match http_metrics(n.health_addr) {
                None => "health endpoint unreachable".to_owned(),
                Some(body) => body
                    .lines()
                    .filter(|l| {
                        l.starts_with("melin_raft_") || l.starts_with("melin_replicas_connected")
                    })
                    .collect::<Vec<_>>()
                    .join(" "),
            };
            format!("node {}: {gauges}", i + 1)
        })
        .collect::<Vec<_>>()
        .join("\n")
}

#[test]
#[serial]
fn acked_events_survive_primary_death_under_ram_policy() {
    // Opt-in diagnostics: with RUST_LOG set, capture the nodes' tracing
    // output (all three run in this process) so a failing run records
    // the control-plane election dialogue, not just the panic-time
    // metrics snapshot. No-op when RUST_LOG is unset.
    melin_test_node::logs::capture_if_requested();
    let tmp = tempfile::tempdir().expect("tempdir");

    let nodes: Vec<NodeSetup> = (0..3)
        .map(|i| NodeSetup {
            key: SigningKey::from_bytes(&[0x61 + i as u8; 32]),
            addrs: melin_test_node::addrs(i, || free_addr(PORT_BASE)),
            raft_addr: free_addr(PORT_BASE),
            health_addr: free_addr(PORT_BASE),
            admin_addr: free_addr(PORT_BASE),
        })
        .collect();
    // One key for both client writes and admin: the operator role covers
    // each.
    let client_key = SigningKey::from_bytes(&[0x11; 32]);

    let b64 = |k: &SigningKey| {
        base64::engine::general_purpose::STANDARD.encode(k.verifying_key().to_bytes())
    };
    let auth_path = tmp.path().join("authorized_keys");
    let mut table: String = nodes
        .iter()
        .enumerate()
        .map(|(i, n)| format!("replication {} node-{}\n", b64(&n.key), i + 1))
        .collect();
    table.push_str(&format!("operator {} client\n", b64(&client_key)));
    std::fs::write(&auth_path, table).unwrap();

    // Identical peer list on every node, self included.
    let peers: Vec<String> = nodes
        .iter()
        .enumerate()
        .map(|(i, n)| format!("{}@{}#{}", i + 1, n.raft_addr, b64(&n.key)))
        .collect();

    let make_config = |i: usize| -> ServerConfig {
        let key_path = tmp.path().join(format!("key-{i}"));
        std::fs::write(&key_path, nodes[i].key.to_bytes()).unwrap();
        ServerConfig {
            journal: tmp.path().join(format!("node-{i}.journal")),
            authorized_keys: auth_path.clone(),
            ack_policy: AckPolicy::Ram,
            // Unpinned, and therefore yielding: three nodes run in this
            // process alongside the rest of the suite, and the default
            // layout would stack every node's same-role thread on one core
            // while a spinner would starve whatever shares its core, the
            // test's own client included.
            cores: PipelineCores::unpinned(),
            tick_interval_ms: 0,
            snapshot_interval_ms: 0,
            health_bind: Some(nodes[i].health_addr),
            admin_bind: Some(nodes[i].admin_addr),
            replication_key: Some(key_path),
            raft_bind: Some(nodes[i].raft_addr),
            raft_node_id: Some(i as u64 + 1),
            raft_peer: peers.clone(),
            raft_dir: Some(tmp.path().join(format!("node-{i}.raft"))),
            raft_auto_promote: true,
            ..melin_test_node::config()
        }
    };

    let replication_addr = nodes[0].addrs.replication();
    let start = |i: usize, config: ServerConfig| {
        melin_test_node::start_at::<Counter>(
            &nodes[i].addrs,
            config,
            startup_events(i),
            (),
            RequestDecoder,
            ResponseEncoder,
        )
    };

    // --- Primary (node 0): `ram` ack policy, replication bind. ---
    let primary = {
        let mut config = make_config(0);
        config.replication_bind = Some(replication_addr);
        start(0, config)
    };

    // --- Replicas (nodes 1, 2). ---
    let replicas: Vec<_> = (1..3)
        .map(|i| {
            let mut config = make_config(i);
            config.replica_of = Some(replication_addr);
            start(i, config)
        })
        .collect();

    // --- Phase 1: cluster forms (both replicas attached, a raft leader
    // elected). Client traffic waits for this so the RAM-quorum gate is
    // satisfiable when the first increment arrives.
    let deadline = Instant::now() + Duration::from_secs(60);
    let leader_at_formation = loop {
        if primary.is_finished() {
            panic!("primary exited early: {:?}", primary.join());
        }
        let connected = http_metrics(nodes[0].health_addr)
            .is_some_and(|m| m.contains("melin_replicas_connected 2\n"));
        let leader = nodes.iter().position(|n| {
            http_metrics(n.health_addr).is_some_and(|m| m.contains("melin_raft_is_leader 1\n"))
        });
        if connected && leader.is_some() {
            break leader;
        }
        assert!(
            Instant::now() < deadline,
            "cluster never formed: replicas_connected_2={connected} leader={leader:?}\n{}",
            cluster_summary(&nodes)
        );
        std::thread::sleep(Duration::from_millis(200));
    };

    // --- Phase 2: acked client traffic. Each response returned only
    // after a replica confirmed in-memory receipt — the `ram` gate
    // live, on real cursors. The first ack already carries node 0's
    // genesis and `on_primary`, journaled before it served; the client
    // adds 1 + 2 + 4 = 7.
    let acked_total = GENESIS + ON_PRIMARY + 7;
    {
        let mut conn = connect_authenticated(primary.addr(), &client_key);
        let mut expected_total = GENESIS + ON_PRIMARY;
        for amount in [1u64, 2, 4] {
            expected_total += amount;
            let responses = conn
                .request(&increment_request(amount))
                .expect("the increment is answered");
            assert_eq!(responses.len(), 1);
            assert_eq!(responses[0][0], KIND_RESP_ACK, "increment must be acked");
            let value = u64::from_le_bytes(responses[0][1..9].try_into().unwrap());
            assert_eq!(value, expected_total, "ack carries the running total");
        }
    }

    // --- Phase 3: kill the primary. Every acked event now exists only
    // on the surviving nodes (their RAM, and their journals as their
    // own disk syncs trail through). On DPDK it is cut off first, so
    // that its stop announces nothing: no RST reaches the replicas, and
    // each must drop its link on the liveness deadline before the
    // promotion below may proceed (auto-promotion refuses while the
    // primary link is up). On kernel TCP a node has no link of its own
    // to cut; its stop closes the links, as a crash would. ---
    #[cfg(feature = "dpdk")]
    primary.cut_off();
    primary.stop();

    // --- Phase 4: exactly one replica auto-promotes. The promotion
    // policy sees the ack policy the replica's lineage holds — the `ram`
    // the dead primary journaled — and must accept it (an ack always
    // waited for a second node, so the election recency filter proves the
    // winner holds every acked event). ---
    let deadline = Instant::now() + Duration::from_secs(60);
    let winner = loop {
        assert!(
            Instant::now() < deadline,
            "no replica promoted within the deadline\n{}",
            cluster_summary(&nodes)
        );
        let serving: Vec<usize> = (1..3)
            .filter(|&i| serves_clients(nodes[i].addrs.client(), Duration::from_millis(500)))
            .collect();
        match serving.len() {
            0 => std::thread::sleep(Duration::from_millis(200)),
            1 => break serving[0],
            _ => panic!("both replicas promoted — split brain: {serving:?}"),
        }
    };
    let loser = if winner == 1 { 2 } else { 1 };
    assert!(
        !serves_clients(nodes[loser].addrs.client(), Duration::from_secs(2)),
        "the losing replica must not serve clients"
    );
    // The lineage's `ram` came over with the stream: the promoted node's
    // journaled policy is the one its replica phase received.
    wait_for_gauge(
        nodes[winner].health_addr,
        "melin_ack_policy_journaled",
        AckPolicy::Ram.as_u8().into(),
    );

    // --- Phase 5: the documented post-failover workflow. The promoted
    // node still runs `ram`, whose gate is structurally
    // unsatisfiable with no replicas attached (fail-closed), so the
    // operator swaps it to `disk` over the admin endpoint. Retried:
    // the admin listener is up from boot, but promotion may still be
    // settling when the first command lands.
    admin_until_ok(nodes[winner].admin_addr, &client_key, "ACK-POLICY disk");
    // Journaled, and in force: the promoted node runs with replication
    // disabled, so nothing could witness the change.
    wait_for_gauge(
        nodes[winner].health_addr,
        "melin_ack_policy_in_force",
        AckPolicy::Disk.as_u8().into(),
    );

    // --- Phase 6: the acked total survived. A fresh client reads the
    // counter from the new primary; every event the old primary acked
    // under the `ram` ack policy must be in it — and on top of it the
    // winner's own `on_primary`, journaled on promotion before it served,
    // while neither replica's genesis ever was. ---
    {
        let mut conn = connect_authenticated(nodes[winner].addrs.client(), &client_key);
        let responses = conn
            .request(&GET_VALUE_REQUEST)
            .expect("the query is answered");
        assert_eq!(responses.len(), 1);
        assert_eq!(responses[0][0], KIND_RESP_VALUE);
        let value = u64::from_le_bytes(responses[0][1..9].try_into().unwrap());
        let seqs: Vec<String> = (1..3)
            .map(|i| {
                let m = http_metrics(nodes[i].health_addr).unwrap_or_default();
                let line = m
                    .lines()
                    .filter(|l| {
                        l.starts_with("melin_journal_sequence")
                            || l.starts_with("melin_events_processed")
                    })
                    .collect::<Vec<_>>()
                    .join(" ");
                format!("node {}: {line}", i + 1)
            })
            .collect();
        assert_eq!(
            value,
            acked_total + (winner as u64 + 1) * ON_PRIMARY,
            "the promoted node must hold every event acked under `ram`, plus its own \
             on_primary events (winner=node {}, leader at formation=node {:?}, {})",
            winner + 1,
            leader_at_formation.map(|i| i + 1),
            seqs.join("; ")
        );
    }

    // Teardown.
    for replica in replicas {
        replica.stop();
    }
}