gunnar-server 2.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()));
}