use std::collections::{BTreeSet, HashMap};

use assertables::{assume, assume_eq};
use log::info;
use rand::prelude::*;
use rand_pcg::Pcg64;
use sugars::boxed;

use anysystem::test::TestResult;
use anysystem::{Message, System};

use crate::common::*;

pub fn test_simple(config: &TestConfig) -> TestResult {
    let mut sys = build_system(config);
    let group = sys.process_names();
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)
}

pub fn test_get_members_semantics(config: &TestConfig) -> TestResult {
    let mut sys = build_system(config);
    let group = sys.process_names();
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;
    for proc in sys.process_names() {
        get_members(&mut sys, &proc)?;
    }
    Ok(true)
}

pub fn test_random_seed(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = Vec::new();
    for proc in sys.process_names() {
        let seed = match group.len() {
            0 => &proc,
            _ => group.choose(&mut rand).unwrap(),
        };
        sys.send_local_message(&proc, Message::json("JOIN", &JoinMessage { seed }));
        group.push(proc);
    }
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_process_join(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let new_proc = group.remove(rand.gen_range(0..group.len()));
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process joins the system
    sys.send_local_message(&new_proc, Message::json("JOIN", &JoinMessage { seed }));
    group.push(new_proc);
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_process_leave(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process leaves the system
    let left_proc = group.remove(rand.gen_range(0..group.len()));
    sys.send_local_message(&left_proc, Message::json("LEAVE", &LeaveMessage {}));
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_process_crash(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process crashes
    let crashed = group.remove(rand.gen_range(0..group.len()));
    crash_process(&crashed, &mut sys);
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_seed_process_crash(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0].clone();
    initialize_group(&mut sys, &group, seed)?;

    // seed process crashes
    group.remove(0);
    crash_process(seed, &mut sys);
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_process_crash_recover(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0].clone();
    initialize_group(&mut sys, &group, seed)?;

    // process crashes
    let crashed = group.remove(rand.gen_range(0..group.len()));
    crash_process(&crashed, &mut sys);
    step_until_stabilized(&mut sys, group.clone().into_iter().collect())?;

    // process recovers
    recover_process(&crashed, &mut sys, config);
    sys.send_local_message(&crashed, Message::json("JOIN", &JoinMessage { seed }));

    group.push(crashed);
    step_until_membership_stable(&mut sys, group.into_iter().collect(), "process recovery")
}

pub fn test_process_offline(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process goes offline
    let offline_proc = group.remove(rand.gen_range(0..group.len()));
    sys.network().disconnect_node(&offline_proc);
    step_until_stabilized(&mut sys, group.into_iter().collect())?;
    step_until_stabilized_at(&mut sys, BTreeSet::from([offline_proc]), "isolated process")
}

pub fn test_seed_process_offline(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0].clone();
    initialize_group(&mut sys, &group, seed)?;

    // seed process goes offline
    group.remove(0);
    sys.network().disconnect_node(seed);
    step_until_stabilized(&mut sys, group.into_iter().collect())?;
    step_until_stabilized_at(&mut sys, BTreeSet::from([seed.clone()]), "isolated process")
}

pub fn test_process_offline_recover(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process goes offline
    let offline_proc = group.remove(rand.gen_range(0..group.len()));
    sys.network().disconnect_node(&offline_proc);
    step_until_stabilized(&mut sys, group.clone().into_iter().collect())?;
    step_until_stabilized_at(
        &mut sys,
        BTreeSet::from([offline_proc.clone()]),
        "isolated process",
    )?;

    // process goes back online
    sys.network().connect_node(&offline_proc);
    group.push(offline_proc);
    step_until_membership_stable(
        &mut sys,
        group.into_iter().collect(),
        "network reconnection",
    )
}

pub fn test_network_partition(config: &TestConfig) -> TestResult {
    run_network_partition(config, false, false)
}

pub fn test_network_partition_recover(config: &TestConfig) -> TestResult {
    run_network_partition(config, false, true)
}

pub fn test_network_partition_unequal(config: &TestConfig) -> TestResult {
    run_network_partition(config, true, false)
}

pub fn test_network_partition_unequal_recover(config: &TestConfig) -> TestResult {
    run_network_partition(config, true, true)
}

fn run_network_partition(config: &TestConfig, unequal: bool, recover: bool) -> TestResult {
    if unequal {
        assume!(
            config.process_count >= 3,
            "Unequal network partition requires at least 3 processes"
        )?;
    }
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    let (group1, group2): (Vec<_>, Vec<_>) = if unequal {
        // Always a nonempty strict minority, including odd process counts.
        let (minority, majority) = group.split_at((group.len() - 1) / 2);
        (
            minority.iter().map(String::as_str).collect(),
            majority.iter().map(String::as_str).collect(),
        )
    } else {
        // Preserve the existing random partition and its seeds.
        group
            .iter()
            .map(String::as_str)
            .partition(|_| rand.gen_range(0.0..1.0) > 0.6)
    };
    info!(
        "Network partition: [{}] ({} processes) | [{}] ({} processes)",
        group1.join(", "),
        group1.len(),
        group2.join(", "),
        group2.len()
    );
    sys.network().make_partition(&group1, &group2);
    step_until_stabilized_at(
        &mut sys,
        group1.into_iter().map(String::from).collect(),
        "network partition: first subgroup",
    )?;
    step_until_stabilized_at(
        &mut sys,
        group2.into_iter().map(String::from).collect(),
        "network partition: second subgroup",
    )?;

    if recover {
        sys.network().reset();
        step_until_membership_stable(&mut sys, group.into_iter().collect(), "network recovery")?;
    }
    Ok(true)
}

pub fn test_process_cannot_receive(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process goes partially offline (cannot receive incoming messages)
    let blocked_proc = group.remove(rand.gen_range(0..group.len()));
    sys.network().drop_incoming(&blocked_proc);
    step_until_stabilized(&mut sys, group.into_iter().collect())?;
    step_until_stabilized_at(&mut sys, BTreeSet::from([blocked_proc]), "isolated process")
}

pub fn test_process_join_cannot_receive(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let new_proc = group.remove(rand.gen_range(0..group.len()));
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process tries to join the system but cannot receive messages
    // it must not be kept in the group
    sys.network().drop_incoming(&new_proc);
    sys.send_local_message(&new_proc, Message::json("JOIN", &JoinMessage { seed }));
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_process_cannot_send(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // process goes partially offline (cannot send outgoing messages)
    let blocked_proc = group.remove(rand.gen_range(0..group.len()));
    sys.network().drop_outgoing(&blocked_proc);
    step_until_stabilized(&mut sys, group.into_iter().collect())?;
    step_until_stabilized_at(&mut sys, BTreeSet::from([blocked_proc]), "isolated process")
}

pub fn test_two_processes_cannot_communicate(config: &TestConfig) -> TestResult {
    run_pairwise_link_failure(config, false)
}

pub fn test_two_processes_cannot_communicate_in_triangle(config: &TestConfig) -> TestResult {
    let config = TestConfig {
        process_count: 3,
        ..*config
    };
    run_pairwise_link_failure(&config, true)
}

fn run_pairwise_link_failure(config: &TestConfig, triangle: bool) -> TestResult {
    assume!(
        config.process_count >= 3,
        "Pairwise link failure with a connected group requires at least 3 processes"
    )?;
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    if triangle {
        group.sort();
    } else {
        group.shuffle(&mut rand);
    }
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // In a triangle, the seed remains the bridge between the other two nodes.
    let proc1 = if triangle { &group[1] } else { seed };
    let proc2 = if triangle {
        &group[2]
    } else {
        group.get(rand.gen_range(1..group.len())).unwrap()
    };
    info!(
        "Network links disabled: {proc1} <-> {proc2}; group remains connected (N={})",
        group.len()
    );
    sys.network().disable_link(proc1, proc2);
    sys.network().disable_link(proc2, proc1);
    // Preserve the original test's execution before checking convergence.
    sys.steps(1000);
    step_until_membership_stable(
        &mut sys,
        group.into_iter().collect(),
        "pairwise link failure",
    )
}

/// Once membership is correct, observe every event without advancing time in GET_MEMBERS.
/// A transient mismatch restarts the stability window, allowing eventual convergence.
fn step_until_membership_stable(
    sys: &mut System,
    group: BTreeSet<String>,
    stage: &str,
) -> TestResult {
    const STABLE_WINDOW: f64 = 30.0;
    const CONVERGENCE_LIMIT: f64 = 300.0;
    let deadline = sys.time() + CONVERGENCE_LIMIT;
    let mut stable_since = None;
    let mut checked_events = 0_u64;
    let mut search_exhausted = false;
    let mut last_mismatch = String::from("no correct membership snapshot");
    loop {
        let now = sys.time();
        // The next event may lie beyond the observed window. Its effects must
        // not invalidate the unchanged state inside the completed window.
        if stable_since
            .is_some_and(|since| now > since + STABLE_WINDOW && since + STABLE_WINDOW <= deadline)
        {
            break;
        }
        if now > deadline {
            return Err(format!(
                "Group members lists are not stable for {STABLE_WINDOW}s within {CONVERGENCE_LIMIT}s; {last_mismatch}"
            ));
        }
        if checked_events.is_multiple_of(100) {
            crate::runner::progress(format!(
                "{stage}: membership stability: N={}, simulation time={now:.1}, stable for {:.1}/{STABLE_WINDOW}s, checked events={checked_events}",
                group.len(), stable_since.map_or(0.0, |since| now - since)
            ));
        }
        let mut correct = true;
        for proc in &group {
            let data = get_members(sys, proc)?;
            let members: BTreeSet<_> = data.members.into_iter().collect();
            if members != group {
                correct = false;
                last_mismatch = format!(
                    "process {proc} at simulation time {now:.3}: expected {group:?}, got {members:?}"
                );
            }
        }
        if correct {
            let since = *stable_since.get_or_insert(now);
            if now - since >= STABLE_WINDOW {
                break;
            }
        } else {
            stable_since = None;
        }
        if !correct && !search_exhausted && now < deadline {
            // Search for the next candidate snapshot at the usual 5s cadence.
            // Only a correct candidate starts event-by-event observation. This
            // avoids filling AnySystem's retained trace with failed queries.
            search_exhausted = !sys.step_for_duration(5.0_f64.min(deadline - now));
            continue;
        }
        if !correct {
            return Err(format!(
                "Group members lists are not stable for {STABLE_WINDOW}s within {CONVERGENCE_LIMIT}s; {last_mismatch}"
            ));
        }
        if !sys.step() {
            // With no pending events, a correct membership cannot change.
            break;
        }
        checked_events += 1;
    }
    info!(
        "Membership remained stable for {STABLE_WINDOW}s ({stage}; checked after {checked_events} simulation events)"
    );
    Ok(true)
}

pub fn test_slow_network(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // slow down network for a while
    sys.network().set_delays(0.1, 1.0);
    sys.steps(200);
    sys.network().set_delays(0.01, 0.1);
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_flaky_network(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // make network unreliable for a while
    sys.network().set_drop_rate(0.5);
    sys.steps(1000);
    sys.network().set_drop_rate(0.0);
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_flaky_network_on_start(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];

    // make network unreliable from the start
    sys.network().set_drop_rate(0.2);
    for proc in &group {
        sys.send_local_message(proc, Message::json("JOIN", &JoinMessage { seed }));
    }
    sys.steps(1000);
    sys.network().set_drop_rate(0.0);
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_flaky_network_and_crash(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    // make network unreliable for a while + crash process
    sys.network().set_drop_rate(0.5);
    let crashed = group.remove(rand.gen_range(0..group.len()));
    crash_process(&crashed, &mut sys);
    sys.steps(1000);
    sys.network().set_drop_rate(0.0);
    step_until_stabilized(&mut sys, group.into_iter().collect())
}

pub fn test_chaos_monkey(config: &TestConfig) -> TestResult {
    let mut rand = Pcg64::seed_from_u64(config.seed);
    let mut sys = build_system(config);
    let mut group = sys.process_names();
    group.shuffle(&mut rand);
    let seed = &group[0];
    initialize_group(&mut sys, &group, seed)?;

    info!(
        "CHAOS MONKEY run seed: {}; join order: [{}]; seed process: {seed}",
        config.seed,
        group.join(", ")
    );
    // Keep at least two participants; pairwise link failures must not isolate
    // a participant still expected in the control group.
    let failure_count = 6.min(group.len().saturating_sub(2));
    for step in 1..=failure_count {
        let p = rand.gen_range(0.0..1.0);
        // do some nasty things
        match p {
            p if p < 0.25 => {
                // crash process
                let victim = group.remove(rand.gen_range(0..group.len()));
                info!(
                    "- Step {step} at {:.2}: process {victim} crashes",
                    sys.time()
                );
                crash_process(&victim, &mut sys);
            }
            p if p < 0.5 => {
                // disconnect process
                let victim = group.remove(rand.gen_range(0..group.len()));
                info!(
                    "- Step {step} at {:.2}: process {victim} disconnects",
                    sys.time()
                );
                sys.network().disconnect_node(&victim);
            }
            p if p < 0.75 => {
                // partially disconnect process (cannot receive)
                let victim = group.remove(rand.gen_range(0..group.len()));
                info!(
                    "- Step {step} at {:.2}: process {victim} cannot receive messages",
                    sys.time()
                );
                sys.network().drop_incoming(&victim);
            }
            _ => {
                // two processes cannot communicate with each other
                let proc1 = group.get(rand.gen_range(0..group.len())).unwrap();
                let mut proc2 = group.get(rand.gen_range(0..group.len())).unwrap();
                while proc1 == proc2 {
                    proc2 = group.get(rand.gen_range(0..group.len())).unwrap();
                }
                info!(
                    "- Step {step} at {:.2}: processes {proc1} and {proc2} cannot communicate",
                    sys.time()
                );
                sys.network().disable_link(proc1, proc2);
                sys.network().disable_link(proc2, proc1);
            }
        }
        info!(
            "Control group after step {step}: [{}] ({} processes)",
            group.join(", "),
            group.len()
        );
        step_until_stabilized(&mut sys, group.clone().into_iter().collect())?;
    }
    Ok(true)
}

pub fn test_scalability_normal(config: &TestConfig) -> TestResult {
    let sys_sizes = [
        config.process_count,
        config.process_count * 2,
        config.process_count * 5,
        config.process_count * 10,
    ];
    let mut measurements = Vec::new();
    for size in sys_sizes {
        let mut run_config = *config;
        run_config.process_count = size;
        let mut rand = Pcg64::seed_from_u64(config.seed);
        let mut sys = build_system(&run_config);
        let mut group = sys.process_names();
        group.shuffle(&mut rand);
        let seed = &group[0];
        initialize_group(&mut sys, &group, seed)?;

        let init_time = sys.time();
        let init_net_traffic = sys.network().traffic();
        let init_msg_count = sys.network().network_message_count();
        let mut init_loads = HashMap::new();
        for proc in sys.process_names() {
            init_loads.insert(
                proc.clone(),
                sys.sent_message_count(&proc) + sys.received_message_count(&proc),
            );
        }

        crate::runner::progress(format!(
            "steady-state measurement: N={size}, simulation time={:.1}",
            sys.time()
        ));
        sys.step_for_duration(10.0);

        let mut loads = Vec::new();
        for proc in sys.process_names() {
            let load = sys.sent_message_count(&proc) + sys.received_message_count(&proc);
            loads.push(load - init_loads.get(&proc).unwrap());
        }
        let min_load = *loads.iter().min().unwrap();
        let max_load = *loads.iter().max().unwrap();
        let duration = sys.time() - init_time;
        let traffic = sys.network().traffic();
        let message_count = sys.network().network_message_count();
        measurements.push((
            duration,
            (traffic - init_net_traffic) as f64 / duration,
            (message_count - init_msg_count) as f64 / duration,
            max_load as f64 / duration,
            max_load as f64 / min_load as f64,
        ));
    }
    let mut scaling_ok = true;
    let mut load_ratio_ok = true;
    for i in 0..sys_sizes.len() {
        let (time, traffic, message_count, max_load, load_ratio) = measurements[i];
        println!(
            "- N = {}: time - {:.2}, traffic/s - {:.2}, messages/s - {:.2}, max load - {:.2}, max/min load - {:.2}",
            sys_sizes[i], time, traffic, message_count, max_load, load_ratio
        );
        if load_ratio > 5.0 {
            load_ratio_ok = false;
        }
        if i > 0 {
            let size_ratio = sys_sizes[i] as f64 / sys_sizes[i - 1] as f64;
            let traffic_ratio = traffic / measurements[i - 1].1;
            let messages_ratio = message_count / measurements[i - 1].2;
            if traffic_ratio > 2.0 * size_ratio || messages_ratio > 2.0 * size_ratio {
                scaling_ok = false;
            }
        }
    }
    assume!(scaling_ok, "Bad network load scaling")?;
    assume!(load_ratio_ok, "Bad max/min process load")?;
    Ok(true)
}

pub fn test_scalability_crash(config: &TestConfig) -> TestResult {
    let sys_sizes = [
        config.process_count,
        config.process_count * 2,
        config.process_count * 5,
        config.process_count * 10,
    ];
    let mut measurements = Vec::new();
    for size in sys_sizes {
        let mut run_config = *config;
        run_config.process_count = size;
        let mut rand = Pcg64::seed_from_u64(config.seed);
        let mut sys = build_system(&run_config);
        let mut group = sys.process_names();
        group.shuffle(&mut rand);
        let seed = &group[0];
        initialize_group(&mut sys, &group, seed)?;

        let init_time = sys.time();
        let init_net_traffic = sys.network().traffic();
        let init_msg_count = sys.network().network_message_count();
        let mut init_loads = HashMap::new();
        for proc in sys.process_names() {
            init_loads.insert(
                proc.clone(),
                sys.sent_message_count(&proc) + sys.received_message_count(&proc),
            );
        }

        let crashed = group.remove(rand.gen_range(0..group.len()));
        crash_process(&crashed, &mut sys);
        step_until_stabilized_at(
            &mut sys,
            group.clone().into_iter().collect(),
            "stabilization after crash",
        )?;

        let mut loads = Vec::new();
        for proc in sys.process_names() {
            if proc != crashed {
                let load = sys.sent_message_count(&proc) + sys.received_message_count(&proc);
                loads.push(load - init_loads.get(&proc).unwrap());
            }
        }
        let min_load = *loads.iter().min().unwrap();
        let max_load = *loads.iter().max().unwrap();
        let duration = sys.time() - init_time;
        let traffic = sys.network().traffic();
        let message_count = sys.network().network_message_count();
        measurements.push((
            duration,
            (traffic - init_net_traffic) as f64 / duration,
            (message_count - init_msg_count) as f64 / duration,
            max_load as f64 / duration,
            max_load as f64 / min_load as f64,
        ));
    }
    let mut scaling_ok = true;
    let mut load_ratio_ok = true;
    for i in 0..sys_sizes.len() {
        let (time, traffic, message_count, max_load, load_ratio) = measurements[i];
        println!(
            "- N = {}: time - {:.2}, traffic/s - {:.2}, messages/s - {:.2}, max load - {:.2}, max/min load - {:.2}",
            sys_sizes[i], time, traffic, message_count, max_load, load_ratio
        );
        if load_ratio > 5.0 {
            load_ratio_ok = false;
        }
        if i > 0 {
            let size_ratio = sys_sizes[i] as f64 / sys_sizes[i - 1] as f64;
            let traffic_ratio = traffic / measurements[i - 1].1;
            let messages_ratio = message_count / measurements[i - 1].2;
            if traffic_ratio > 2.0 * size_ratio || messages_ratio > 2.0 * size_ratio {
                scaling_ok = false;
            }
        }
    }
    assume!(scaling_ok, "Bad network load scaling")?;
    assume!(load_ratio_ok, "Bad max/min process load")?;
    Ok(true)
}

fn initialize_group(sys: &mut System, group: &Vec<String>, seed: &str) -> TestResult {
    for proc in group {
        sys.send_local_message(proc, Message::json("JOIN", &JoinMessage { seed }));
    }
    step_until_stabilized_at(
        sys,
        group.clone().into_iter().collect(),
        "initial group assembly",
    )
}

fn crash_process(name: &str, sys: &mut System) {
    // we just crash the node on which the process is running
    sys.crash_node(name);
}

fn recover_process(name: &str, sys: &mut System, config: &TestConfig) {
    sys.recover_node(name);
    let clock_skew = sys.gen_range(0.0..10.0);
    // A reboot may change the local clock. Apply it before on_start is called.
    sys.set_node_clock_skew(name, clock_skew);
    info!(
        "Process {name} recovers at simulation time {:.6} with clock skew {clock_skew:.6}",
        sys.time()
    );
    let process = config.process_factory.build((name,), config.seed);
    sys.add_process(name, boxed!(process), name);
}

fn get_members(sys: &mut System, proc: &str) -> Result<MembersMessage, String> {
    let sent_before = sys.sent_message_count(proc);
    sys.send_local_message(proc, Message::json("GET_MEMBERS", &GetMembersMessage {}));
    assume_eq!(
        sys.sent_message_count(proc),
        sent_before,
        format!("GET_MEMBERS must not send network messages: process {proc}")
    )?;
    let messages = sys
        .step_until_local_message_max_steps(proc, 0)
        .map_err(|_| {
            format!("GET_MEMBERS must return immediately without simulation steps: process {proc}")
        })?;
    assume_eq!(messages.len(), 1, "expected exactly one MEMBERS message")?;
    assume_eq!(messages[0].tip, "MEMBERS", "Wrong message type")?;
    serde_json::from_str(&messages[0].data)
        .map_err(|error| format!("Invalid MEMBERS message from {proc}: {error}"))
}

fn step_until_stabilized(sys: &mut System, group: BTreeSet<String>) -> TestResult {
    step_until_stabilized_at(sys, group, "membership stabilization")
}

fn step_until_stabilized_at(sys: &mut System, group: BTreeSet<String>, stage: &str) -> TestResult {
    let max_time = sys.time() + 300.; // timeout is 5 minutes
    let mut stabilized = BTreeSet::new();
    let mut memberlists = HashMap::new();

    while stabilized.len() < group.len() && sys.time() < max_time {
        crate::runner::progress(format!("{stage}: N={}, correct members lists={}/{}, simulation time={:.1}, network messages={}",
            group.len(), stabilized.len(), group.len(), sys.time(), sys.network().network_message_count()));
        let cont = sys.step_for_duration(5.);
        stabilized.clear();
        // Query processes in a stable order without advancing simulation time.
        for proc in group.iter() {
            let data = get_members(sys, proc)?;
            let members: BTreeSet<String> = data.members.clone().into_iter().collect();
            if members.eq(&group) {
                stabilized.insert(proc.clone());
            }
            memberlists.insert(proc.clone(), data.members);
        }
        if !cont {
            break;
        }
    }

    if stabilized != group && group.len() <= 10 {
        println!("Members lists:");
        for proc in sys.process_names() {
            if group.contains(&proc) {
                let members = memberlists.get_mut(&proc).unwrap();
                members.sort();
                println!("- [{}] {}", proc, members.join(", "));
            }
        }
        let mut expected = group.clone().into_iter().collect::<Vec<_>>();
        expected.sort();
        println!("Expected group: {}", expected.join(", "));
    }
    assume_eq!(stabilized, group, "Group members lists are not stabilized")?;
    Ok(true)
}