gunnar-server 3.0.0

//! **The twin's own growth holds its follower** (T21, owner ruling 2026-09-22: the twin
//! grows exactly like main and is never restarted). A held twin fetches, appends and
//! applies NOTHING, says `maintenance` on its status and on the report its primary
//! reads, still asks the primary for its head, and follows on from its own head the
//! moment it is released.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use super::super::*;
use super::{hold, is_held, release};
use crate::stores::ServerStores;

fn wait_for(what: &str, mut cond: impl FnMut() -> bool) {
    let deadline = Instant::now() + Duration::from_secs(20);
    while !cond() {
        assert!(Instant::now() < deadline, "timed out waiting for: {what}");
        std::thread::sleep(Duration::from_millis(5));
    }
}

#[test]
fn a_held_twin_applies_nothing_says_maintenance_and_follows_again_when_released() {
    let pd = tempfile::tempdir().unwrap();
    let td = tempfile::tempdir().unwrap();
    let primary = Arc::new(ServerStores::open(pd.path()).unwrap());
    primary.catalog().create_namespace("before").unwrap();
    TwinConfig {
        primary: "primary.test:50051".into(),
        joined_at_unix_ms: 0,
    }
    .save(td.path())
    .unwrap();
    let twin = Arc::new(ServerStores::open(td.path()).unwrap());

    let stop = Arc::new(AtomicBool::new(false));
    let runner = {
        let (twin, log, stop) = (
            Arc::clone(&twin),
            Arc::clone(primary.txlog()),
            Arc::clone(&stop),
        );
        std::thread::spawn(move || {
            Follower::new(twin, Box::new(log)).run(Duration::from_millis(5), &stop)
        })
    };
    wait_for("the twin level with its primary", || {
        twin.txlog().head_seq() == primary.txlog().head_seq()
    });

    // ── the grow machine holds it (BoxWorld::pause_admission on a twin) ──
    hold(twin.root());
    assert!(is_held(twin.root()));
    let held_at = twin.txlog().head_seq();
    let s = status(twin.root()).expect("a status");
    assert_eq!(s.state, State::Maintenance);
    assert_eq!(s.state.to_string(), "maintenance");
    let report = measure_report(&twin);
    assert_eq!(
        report.state, "maintenance",
        "the word the primary's heartbeat carries"
    );
    assert_eq!(
        (report.disk_total_bytes, report.disk_used_bytes),
        (0, 0),
        "an unmounted set has no size to report"
    );

    // The primary goes on writing while the twin's set is unmounted.
    primary.catalog().create_namespace("during").unwrap();
    let primary_head = primary.txlog().head_seq();
    assert!(primary_head > held_at);
    wait_for("the held twin to hear the primary's new head", || {
        status(twin.root()).is_some_and(|s| s.primary_head_seq == primary_head)
    });
    std::thread::sleep(Duration::from_millis(60));
    assert_eq!(
        twin.txlog().head_seq(),
        held_at,
        "a held twin appended nothing"
    );
    assert!(
        twin.catalog().namespace("during").unwrap().is_none(),
        "a held twin applied nothing"
    );
    let s = status(twin.root()).unwrap();
    assert_eq!(
        s.state,
        State::Maintenance,
        "still maintenance, with the lag visible: {s:?}"
    );
    assert!(!s.caught_up());

    // ── the set is back (BoxWorld::unpause_admission) ──
    release(twin.root());
    wait_for("the released twin level again", || {
        twin.txlog().head_seq() == primary_head
    });
    wait_for("the status back to following", || {
        status(twin.root()).is_some_and(|s| s.state == State::Following)
    });
    assert!(
        twin.catalog().namespace("during").unwrap().is_some(),
        "what was written during the growth is on the twin now"
    );
    assert!(status(twin.root()).unwrap().caught_up());

    stop.store(true, Ordering::SeqCst);
    runner.join().unwrap();
}

/// A twin whose follower is NOT running (never started, or between two supervisor
/// attempts) still says `maintenance` the moment it is held — the status is the
/// hold's, not the loop's — and the supervisor's `unreachable` does not overwrite it.
#[test]
fn a_held_twin_says_maintenance_even_without_a_running_follower() {
    let td = tempfile::tempdir().unwrap();
    TwinConfig {
        primary: "primary.test:50051".into(),
        joined_at_unix_ms: 0,
    }
    .save(td.path())
    .unwrap();
    let twin = Arc::new(ServerStores::open(td.path()).unwrap());
    hold(twin.root());
    stalled(
        &twin,
        1,
        "the primary did not answer",
        Duration::from_secs(5),
    );
    assert_eq!(status(twin.root()).unwrap().state, State::Maintenance);
    release(twin.root());
    assert!(!is_held(twin.root()));
}