//! Tier-3 MULTI-PROCESS SIGTERM graceful-shutdown proof (m12p6).
//!
//! Regression test for the production bug: a `kubectl delete pod` (SIGTERM, 60s
//! grace) on a real region node did NOT run the deterministic database close, so
//! the persisted HNSW graph (`{data_dir}/vector/*.usearch`) was never written and
//! every boot paid the full multi-minute HNSW rebuild.
//!
//! Root cause: the cluster serve path's graceful HTTP drain
//! (`axum::serve(...).with_graceful_shutdown(...)`) blocks until every in-flight
//! connection closes. Sibling region nodes hold long-lived keep-alive HTTP
//! connections that do not close promptly on our SIGTERM, so the drain — and thus
//! `axum::serve(...).await` — never returned inside the 60s grace; k8s then sent
//! `SIGKILL`. That signal cannot run `Drop`/`shutdown_inner` (where
//! the WAL checkpoint marker AND the HNSW-graph checkpoint are written).
//!
//! The fix (in `main.rs` + `cluster/node.rs`):
//!
//! * `bounded_drain` caps the post-signal drain (`TIDAL_SHUTDOWN_DRAIN_MS`,
//!   default 15s) and then runs the deterministic close regardless; and
//! * `ClusterNode`/`ShardReplica::shutdown` are `&self` (the db handle is an
//!   `ArcSwapOption`), so the close runs even when a stuck connection task kept an
//!   `Arc<ClusterNode>` alive past the drain.
//!
//! # What this proves end-to-end (real OS processes, real SIGTERM)
//!
//! After a real SIGTERM to a region node that has taken writes, the node's
//! `{data_dir}/wal/checkpoint.meta` marker exists and is FRESH — the on-disk proof
//! that `TidalDb::shutdown_inner` ran on the SIGTERM path. That marker is written
//! by the SAME method, in the SAME shutdown body, immediately after
//! `checkpoint_embedding_graphs`; reaching it is exactly what the bug prevented.
//! (The HNSW `.usearch` file itself only materializes once a slot crosses the
//! dimension-aware HNSW crossover — ≥10k vectors at the harness's 4-D slot — which
//! is impractical to seed over HTTP here; the graph SAVE itself is proven by
//! `tidal/tests/m12p6_graph_persistence.rs`. This test proves the cluster SIGTERM
//! path REACHES that close, which is the regression.)
//!
//! Then a restart on the SAME data dir comes back healthy, proving the clean
//! shutdown left a recoverable state.
//!
//! ```bash
//! cargo test -p tidal-server --features cluster-e2e --test cluster_graph_persistence -- --nocapture
//! ```
#![cfg(feature = "cluster-e2e")]
#![allow(clippy::unwrap_used, clippy::missing_panics_doc)]

mod support;

use std::time::{Duration, SystemTime};

use support::multiproc::{
    ClusterOptions, MultiProcCluster, convergence_budget, seed_items_and_embeddings,
};

/// The leader index (region 0 = `us-east`).
const LEADER: usize = 0;

/// `{data_dir}/wal/checkpoint.meta` — written ONLY by `TidalDb::shutdown_inner`
/// (after writes advance the WAL seq), in the same shutdown body as the
/// HNSW-graph checkpoint. Its presence + freshness is the on-disk proof the
/// deterministic close ran on the SIGTERM path.
const WAL_CHECKPOINT_MARKER: &str = "wal/checkpoint.meta";

/// SIGTERM on a region node that has taken writes runs the deterministic
/// database close (WAL checkpoint marker written), and a restart on the same data
/// dir recovers cleanly — the graceful-shutdown path the production bug skipped.
#[test]
fn mp_sigterm_runs_deterministic_close_and_writes_checkpoint_marker() {
    // Drive the post-signal drain cap LOW on every node so the test does not wait
    // the 15s production default if a sibling keep-alive connection lingers (the
    // very condition the fix handles). 250ms is ample for loopback drain; if it
    // is exceeded, the fix still proceeds to the close — which is the point.
    let opts = (0..3).fold(ClusterOptions::new(3), |o, i| {
        o.with_env(i, "TIDAL_SHUTDOWN_DRAIN_MS", "250")
    });
    let cluster = MultiProcCluster::start_with(opts);
    cluster.wait_converged_all(convergence_budget());

    // Seed items + embeddings on the leader so the WAL seq advances past 0 (the
    // marker is only written when there is something to checkpoint).
    seed_items_and_embeddings(&cluster, LEADER, 8);
    // Let the writes settle to disk before we stop.
    std::thread::sleep(Duration::from_millis(500));

    let marker_path = cluster.data_dir(LEADER).join(WAL_CHECKPOINT_MARKER);
    let before = SystemTime::now();

    let mut cluster = cluster; // `stop_graceful` takes &mut self.
    // The REAL SIGTERM: the harness sends `kill -TERM` and waits for the process
    // to exit within its graceful budget (it does NOT hard-kill unless the budget
    // is blown — pre-fix, the drain hung and the budget WOULD blow, leaving no
    // fresh marker; post-fix the bounded drain lets the close run and exit clean).
    cluster.stop_graceful(LEADER);

    // The deterministic close must have written (or refreshed) the WAL checkpoint
    // marker. Existence alone is necessary; freshness rules out a stale marker
    // from an earlier boot.
    let meta = std::fs::metadata(&marker_path).unwrap_or_else(|e| {
        panic!(
            "WAL checkpoint marker {} missing after SIGTERM — the deterministic close did NOT run \
             on the SIGTERM path (the m12p6 regression): {e}",
            marker_path.display()
        )
    });
    let modified = meta
        .modified()
        .expect("checkpoint marker mtime unavailable");
    assert!(
        modified >= before - Duration::from_secs(2),
        "WAL checkpoint marker is stale (mtime {modified:?} predates the SIGTERM at {before:?}); \
         the close ran on an EARLIER boot, not on this SIGTERM"
    );

    // The clean shutdown must leave a recoverable state: restart on the SAME data
    // dir and require health (the harness panics if it does not converge).
    cluster.restart(LEADER, &[("TIDAL_SHUTDOWN_DRAIN_MS", "250")]);
    cluster.wait_healthy(LEADER);
}