tidaldb/tidal-server/tests/reseed_install.rs

//! m11p5 §2.2–§2.7: boot-time snapshot install end to end (B2/B3 slice).
//!
//! Drives the REAL `run_boot_install`: a fresh node with a latched
//! `reseed_required` marker discovers a leader over HTTP `/cluster/status/local`,
//! adopts the leader's term durably, fetches a snapshot from the leader's gRPC
//! `FetchSnapshot` (a fake `SnapshotSource` serving a staged tempdir), verifies
//! BLAKE3 + completeness, copies its identity files into staging, writes the
//! sentinels, and swaps the staged dir in.
//!
//! Asserts the wire contract end to end:
//! - `Installed`: the staged bytes are now canonical in the data dir, the
//!   install sentinel is present (drives the §2.6 post-open seed), the reseed
//!   marker is gone, and the node's own `election_state` survived inside staging
//!   (§2.5 vote safety);
//! - the term was adopted durably BEFORE the fetch (§2.7 persist-before-act);
//! - `needed=false`: the marker is CLEARED and nothing is swapped (the
//!   self-correcting "retry the normal boot" path, §2.2d).

#![allow(clippy::unwrap_used)] // test assertions on known-good fixtures

use std::io::Write;
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;

use tidal_net::{
    GrpcTransport, GrpcTransportConfig,
    sources::{ServingSources, SnapshotSource, SnapshotStageError, SnapshotStaging},
};
use tidal_server::cluster::reseed::{self, InstallOutcome};
use tidaldb::replication::{
    ElectionStore, HardState, ReseedMarker, ReseedMarkerStore, ReseedReason, shard::ShardId,
};

/// Bind port 0 to obtain a free, OS-assigned address.
fn free_addr() -> SocketAddr {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
    listener.local_addr().expect("local_addr")
}

/// A staged snapshot artifact: a tempdir holding files, manifest BLAKE3/size
/// computed from the bytes on disk.
struct StagedDir {
    _dir: tempfile::TempDir,
    root: std::path::PathBuf,
    manifest: Vec<(String, u64, [u8; 32])>,
}

impl StagedDir {
    fn new(files: &[(&str, &[u8])]) -> Self {
        let dir = tempfile::tempdir().expect("tempdir");
        let root = dir.path().to_path_buf();
        let mut manifest = Vec::new();
        for (rel, bytes) in files {
            let abs = root.join(rel);
            if let Some(parent) = abs.parent() {
                std::fs::create_dir_all(parent).expect("mkdir");
            }
            std::fs::write(&abs, bytes).expect("write file");
            let hash = blake3::hash(bytes);
            manifest.push(((*rel).to_string(), bytes.len() as u64, *hash.as_bytes()));
        }
        Self {
            _dir: dir,
            root,
            manifest,
        }
    }
}

/// A `SnapshotSource` serving a fixed staged dir; `needed` toggles the
/// no-snapshot-needed path.
struct FakeSnapshots {
    staged: Arc<StagedDir>,
    needed: bool,
    snapshot_seq: u64,
}

impl SnapshotSource for FakeSnapshots {
    fn stage(&self, _from_seqno: u64) -> Result<SnapshotStaging, SnapshotStageError> {
        Ok(SnapshotStaging {
            needed: self.needed,
            snapshot_seq: self.snapshot_seq,
            root: self.staged.root.clone(),
            files: if self.needed {
                self.staged.manifest.clone()
            } else {
                Vec::new()
            },
        })
    }
    fn release(&self) {}
}

/// Start a real gRPC leader serving `FetchSnapshot` from `source`.
fn leader_grpc(listen: SocketAddr, source: Arc<dyn SnapshotSource>) -> GrpcTransport {
    let sources = ServingSources::default();
    sources.set_snapshot_source(source);
    GrpcTransport::new_with_sources(
        GrpcTransportConfig {
            local_shard: ShardId(0),
            listen_addr: listen,
            insecure: true,
            ..GrpcTransportConfig::default()
        },
        sources,
    )
    .expect("leader transport")
}

/// Start a tiny HTTP server on a dedicated thread that answers
/// `GET /cluster/status/local` with `{is_leader, term}`. Returns the address +
/// a runtime handle kept alive for the test.
fn leader_http_status(is_leader: bool, term: u64) -> (SocketAddr, std::thread::JoinHandle<()>) {
    use axum::{Json, Router, routing::get};

    let addr = free_addr();
    let handle = std::thread::Builder::new()
        .name("test-status-http".into())
        .spawn(move || {
            let rt = tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .unwrap();
            rt.block_on(async move {
                let app = Router::new().route(
                    "/cluster/status/local",
                    get(move || async move {
                        Json(serde_json::json!({
                            "is_leader": is_leader,
                            "term": term,
                            "region": "leader",
                        }))
                    }),
                );
                let listener = tokio::net::TcpListener::bind(addr).await.unwrap();
                axum::serve(listener, app).await.unwrap();
            });
        })
        .unwrap();
    // Give the server a moment to bind.
    std::thread::sleep(std::time::Duration::from_millis(150));
    (addr, handle)
}

/// Write a 2-region topology naming the leader's HTTP + gRPC addresses.
fn write_topology(
    dir: &Path,
    leader_http: SocketAddr,
    leader_grpc: SocketAddr,
) -> std::path::PathBuf {
    let path = dir.join("topology.yaml");
    let mut f = std::fs::File::create(&path).unwrap();
    // `joiner` is this node; `leader` is the discovered leader. The joiner's
    // addresses are placeholders (it never dials itself during discovery).
    writeln!(f, "regions:").unwrap();
    writeln!(f, "  - name: joiner").unwrap();
    writeln!(f, "    grpc_addr: 127.0.0.1:1").unwrap();
    writeln!(f, "    http_addr: 127.0.0.1:2").unwrap();
    writeln!(f, "  - name: leader").unwrap();
    writeln!(f, "    grpc_addr: {leader_grpc}").unwrap();
    writeln!(f, "    http_addr: {leader_http}").unwrap();
    writeln!(f, "leader: leader").unwrap();
    path
}

/// A `data_dir` that is a SUBDIRECTORY of a tempdir (§2.3: the swap renames the
/// data dir, so it must not be a mount root).
fn data_dir_in(tmp: &Path) -> std::path::PathBuf {
    let d = tmp.join("db");
    std::fs::create_dir_all(&d).unwrap();
    d
}

#[test]
fn boot_install_discovers_fetches_and_swaps() {
    let _ = tracing_subscriber::fmt::try_init();
    let file_a: &[u8] = b"snapshot ledger bytes";
    let file_b: Vec<u8> = (0u8..=255).cycle().take(2_500_000).collect(); // multi-chunk
    let staged = Arc::new(StagedDir::new(&[
        ("ledger/000001.sst", file_a),
        ("wal/seg-0001.seg", &file_b),
    ]));

    let grpc_addr = free_addr();
    let _grpc = leader_grpc(
        grpc_addr,
        Arc::new(FakeSnapshots {
            staged: Arc::clone(&staged),
            needed: true,
            snapshot_seq: 4096,
        }),
    );
    let (http_addr, _http) = leader_http_status(true, 7);

    let tmp = tempfile::tempdir().unwrap();
    let data_dir = data_dir_in(tmp.path());
    let topo_path = write_topology(tmp.path(), http_addr, grpc_addr);
    let topology = tidal_server::cluster::load_topology(Some(&topo_path)).unwrap();

    // The node's own identity: a durable election_state at term 3 that MUST
    // survive the swap inside staging (§2.5), and a WAL dir (so the node looks
    // like it has run before).
    std::fs::create_dir_all(data_dir.join("wal")).unwrap();
    let estore = ElectionStore::new(&data_dir);
    estore
        .persist(HardState {
            current_term: 3,
            voted_for: None,
        })
        .unwrap();

    // Latch a snapshot-required marker recorded at from_seqno 100.
    let marker_store = ReseedMarkerStore::new(&data_dir);
    marker_store
        .persist(ReseedMarker {
            reason: ReseedReason::SnapshotRequired,
            from_seqno: 100,
        })
        .unwrap();

    let outcome = reseed::run_boot_install_with(
        &topology,
        "joiner",
        &data_dir,
        ShardId(0),
        std::time::Duration::from_secs(8),
    )
    .unwrap();
    assert_eq!(outcome, InstallOutcome::Installed, "the snapshot installs");

    // The staged bytes are now canonical in the data dir.
    assert_eq!(
        std::fs::read(data_dir.join("ledger/000001.sst")).unwrap(),
        file_a
    );
    assert_eq!(
        std::fs::read(data_dir.join("wal/seg-0001.seg")).unwrap(),
        file_b
    );

    // The install sentinel is present and records the artifact seq AND the
    // discovered leader's region (drives §2.6 + the post-install catch-up pull).
    // `leader` is region index 1 in the topology (`joiner` is 0).
    let sentinel = reseed::read_install_sentinel(&data_dir).unwrap();
    assert_eq!(
        sentinel,
        Some(reseed::InstallSentinel {
            snapshot_seq: 4096,
            leader_region: Some(1),
        }),
        "install sentinel carries the artifact seq + discovered leader region"
    );

    // The reseed marker is GONE (it lived in the old dir, discarded by the swap).
    assert_eq!(
        marker_store.load().unwrap(),
        None,
        "marker consumed by the swap"
    );

    // §2.5: the node's own election_state survived INSIDE staging — term 3, and
    // the discovered term 7 was adopted durably (persist-before-act, §2.7),
    // overwriting it (7 > 3).
    let recovered = match estore.load(true).unwrap() {
        tidaldb::replication::BootState::Rejoin(h) => h,
        other => panic!("expected Rejoin, got {other:?}"),
    };
    assert_eq!(
        recovered.current_term, 7,
        "the discovered leader term was adopted durably before the fetch (§2.7)"
    );

    // No staging or discard dirs leak after a successful swap.
    let dirs = reseed::SwapDirs::derive(&data_dir).unwrap();
    assert!(!dirs.staging.exists(), "staging consumed");
    assert!(!dirs.discard.exists(), "discard cleaned up");
}

#[test]
fn boot_install_needed_false_clears_marker_and_does_not_swap() {
    let _ = tracing_subscriber::fmt::try_init();
    // A leader whose live WAL still covers the frontier → needed=false.
    let staged = Arc::new(StagedDir::new(&[("unused", b"x")]));
    let grpc_addr = free_addr();
    let _grpc = leader_grpc(
        grpc_addr,
        Arc::new(FakeSnapshots {
            staged,
            needed: false,
            snapshot_seq: 0,
        }),
    );
    let (http_addr, _http) = leader_http_status(true, 2);

    let tmp = tempfile::tempdir().unwrap();
    let data_dir = data_dir_in(tmp.path());
    let topo_path = write_topology(tmp.path(), http_addr, grpc_addr);
    let topology = tidal_server::cluster::load_topology(Some(&topo_path)).unwrap();

    std::fs::write(data_dir.join("existing"), b"keep me").unwrap();
    let marker_store = ReseedMarkerStore::new(&data_dir);
    marker_store
        .persist(ReseedMarker {
            reason: ReseedReason::SnapshotRequired,
            from_seqno: 50,
        })
        .unwrap();

    let outcome = reseed::run_boot_install_with(
        &topology,
        "joiner",
        &data_dir,
        ShardId(0),
        std::time::Duration::from_secs(8),
    )
    .unwrap();
    assert_eq!(
        outcome,
        InstallOutcome::NotNeeded,
        "needed=false clears the marker, no swap"
    );
    assert_eq!(
        marker_store.load().unwrap(),
        None,
        "marker cleared (self-correcting)"
    );
    assert!(
        data_dir.join("existing").exists(),
        "the existing data dir is untouched (no swap)"
    );
    assert!(
        reseed::read_install_sentinel(&data_dir).unwrap().is_none(),
        "no install sentinel on the needed=false path"
    );
}

/// A marker boot whose leader is unreachable FALLS BACK (degraded), never blocks
/// boot forever — the review-blocker fix (§2.2). The marker stays latched.
#[test]
fn boot_install_falls_back_when_no_leader_reachable() {
    let _ = tracing_subscriber::fmt::try_init();
    // A status server that reports is_leader=FALSE (no leader to be found), and
    // an unbound gRPC port. Discovery never succeeds → fall back.
    let (http_addr, _http) = leader_http_status(false, 0);
    let grpc_addr = free_addr(); // nothing listening here

    let tmp = tempfile::tempdir().unwrap();
    let data_dir = data_dir_in(tmp.path());
    let topo_path = write_topology(tmp.path(), http_addr, grpc_addr);
    let topology = tidal_server::cluster::load_topology(Some(&topo_path)).unwrap();

    let marker_store = ReseedMarkerStore::new(&data_dir);
    marker_store
        .persist(ReseedMarker {
            reason: ReseedReason::SnapshotRequired,
            from_seqno: 10,
        })
        .unwrap();

    let outcome = reseed::run_boot_install_with(
        &topology,
        "joiner",
        &data_dir,
        ShardId(0),
        std::time::Duration::from_millis(600),
    )
    .unwrap();
    assert_eq!(
        outcome,
        InstallOutcome::FellBack,
        "an unreachable leader falls back to opening the existing dir (never blocks boot)"
    );
    assert!(
        marker_store.load().unwrap().is_some(),
        "the marker stays latched on fallback (the reseed retries next boot)"
    );
}