//! Tier-3 regression gate for the 2026-08-31 shard-1 quorum-write outage
//! (REAL 3-process cluster, RF=3).
//!
//! # The incident this file exists to prevent
//!
//! A probe posted a **128-dimension** vector to `/embeddings` for slot
//! `content_vector`; the live schema declares **1536**. The origin appended the
//! blob to the WAL FIRST (`wal_blob_first` returned `Ok(Some(seq))`), validated
//! SECOND, failed, and answered the client **500**. The record was already
//! durable, so it shipped to both followers, neither could apply it, and both
//! **halted their receivers**:
//!
//! ```text
//! replicated blob batch apply failed (1 records):
//!   [op=write_item_embedding] dimension mismatch: expected 1536, got 128
//! ```
//!
//! Shard 1 froze — leader at `13540698`, followers pinned at
//! `13540694`/`13540693`, `lag` growing — and writes to the group returned
//! **503**, because the followers could not ack. One malformed HTTP request took
//! out replication for a whole shard group. Neither a restart (boot self-heal
//! re-pulls the same record) nor `POST /cluster/reseed` escaped it: the snapshot
//! is captured at the leader's applied frontier, which is itself BEHIND the
//! poison. Only forcing a leader election recovered the group.
//!
//! # What this test asserts
//!
//! 1. A dimension-mismatched `/embeddings` write is rejected **400** — a
//!    malformed vector is a caller error; the old 500 misattributed it to the
//!    server while the record was, in fact, more than failed: durable,
//!    unapplicable, and blocking.
//! 2. It appends **NOTHING**: the leader's WAL frontier (`last_seq`) does not
//!    move across the rejected writes. This is the load-bearing assertion — a
//!    fix that only corrected the status code would still poison the stream.
//! 3. Every receiver stays healthy: all three nodes converge to `lag = 0`, and a
//!    VALID write issued afterwards still replicates to both followers. A halted
//!    receiver freezes flat, so the second half is what distinguishes "alive"
//!    from "merely quiet".
//! 4. `/health` is 200 on every node.
//!
//! Node logs are discarded by the harness (`Stdio::null()` — a piped-but-undrained
//! pipe deadlocks a chatty node), so the `grep -c 'receiver halting'` check from
//! the runbook is expressed here as its observable consequence: lag returns to
//! zero and continues to track new writes.
//!
//! Run:
//! ```bash
//! cargo test -p tidal-server --features cluster-e2e --test cluster_poison_embedding -- --nocapture
//! ```

#![cfg(feature = "cluster-e2e")]
#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

mod support;

use std::time::Duration;

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

/// Region 0 = `us-east` = the initial leader in every harness topology.
const LEADER: usize = 0;
const NODES: usize = 3;

/// The harness schema declares `content_vector` at **4** dimensions
/// (`support::multiproc::write_schema`). Two floats is therefore the local
/// analogue of the live cluster's 128-into-1536 probe.
const DECLARED_DIMENSIONS: usize = 4;

/// How many times the malformed write is replayed. The incident was ONE request;
/// hammering proves the rejection cannot accumulate durable state either.
const POISON_ATTEMPTS: u64 = 5;

/// Serializes the heavy multi-process tests in THIS target.
///
/// Each test here spawns 3 OS processes. The harness's `spawn_lock` only
/// serializes the spawn itself and is released as soon as `start` returns, so
/// without this every test in the file can hold a live cluster simultaneously and
/// the resulting contention starves each other's election/convergence budgets
/// (see the identical guard in `cluster_sharding.rs`). Poison is recovered rather
/// than propagated: one failing test must not cascade into "the rest panicked on a
/// poisoned lock", which hides the original failure.
fn heavy_test_guard() -> std::sync::MutexGuard<'static, ()> {
    static LOCK: std::sync::LazyLock<std::sync::Mutex<()>> =
        std::sync::LazyLock::new(|| std::sync::Mutex::new(()));
    LOCK.lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// The leader's WAL high-water-mark, once it has stopped moving.
///
/// Returns a frontier observed IDENTICAL twice across a quiet window, so the
/// "did the rejected write append anything?" comparison is against a settled
/// value rather than a racing one. Panics if the frontier never settles, which
/// would mean the cluster is not idle and the test's premise is broken.
fn settled_leader_frontier(cluster: &MultiProcCluster) -> u64 {
    let deadline = std::time::Instant::now() + convergence_budget();
    let mut last = None;
    loop {
        let seq = cluster
            .leader_last_seq()
            .expect("the leader must report its own frontier");
        if last == Some(seq) {
            return seq;
        }
        assert!(
            std::time::Instant::now() <= deadline,
            "leader frontier never settled (last two samples {last:?} then {seq}); \
             the cluster is not idle, so an append-nothing assertion would be meaningless"
        );
        last = Some(seq);
        std::thread::sleep(Duration::from_millis(300));
    }
}

/// Post a malformed embedding to the leader and return `(status, body)`.
fn post_malformed(cluster: &MultiProcCluster, entity_id: u64) -> (u16, String) {
    let resp = cluster.post(
        LEADER,
        "/embeddings",
        &serde_json::json!({ "entity_id": entity_id, "values": [0.1, 0.2] }),
    );
    let status = resp.status().as_u16();
    let body = resp.text().unwrap_or_default();
    (status, body)
}

#[test]
fn mp_malformed_embedding_is_rejected_400_and_never_enters_the_replication_stream() {
    let _heavy = heavy_test_guard();
    let cluster = MultiProcCluster::start(NODES);
    let followers: Vec<usize> = (0..NODES).filter(|&i| i != LEADER).collect();

    // Steady state: valid 4-dim embeddings replicated to every node.
    seed_items_and_embeddings(&cluster, LEADER, 4);
    cluster.wait_converged_all(convergence_budget());
    let frontier_before = settled_leader_frontier(&cluster);
    println!("[poison] converged at leader frontier {frontier_before}");

    // ── 1. The malformed write is a CALLER error: 400, not 500 ──
    for attempt in 0..POISON_ATTEMPTS {
        let entity_id = 1000 + attempt;
        let (status, body) = post_malformed(&cluster, entity_id);
        assert_eq!(
            status, 400,
            "a 2-float vector against a {DECLARED_DIMENSIONS}-dim slot is a caller \
             error and must be 400 (the incident returned 500, misattributing it to \
             the server); body: {body}"
        );
        assert!(
            body.contains("dimension mismatch"),
            "the rejection must name the real cause so a caller can fix its model: {body}"
        );
    }

    // ── 2. It appended NOTHING ──
    //
    // The whole defect was ordering: append, then validate. If the record still
    // enters the log, it still ships, and the followers still halt — the status
    // code is cosmetic next to this.
    let frontier_after = settled_leader_frontier(&cluster);
    assert_eq!(
        frontier_after, frontier_before,
        "{POISON_ATTEMPTS} rejected embeddings must append NOTHING to the WAL; the \
         leader frontier moved {frontier_before} -> {frontier_after}, so the poison \
         is durable and will ship to every follower"
    );

    // ── 3. Every receiver is still healthy ──
    //
    // First: nothing halted, so the group is still converged at lag 0.
    cluster.wait_converged_all(convergence_budget());

    // Then the half that a halted receiver cannot fake — a follower whose
    // receiver died reports lag 0 forever while applying nothing, so prove the
    // streams still MOVE by shipping a valid write through them.
    let resp = cluster.post(
        LEADER,
        "/embeddings",
        &serde_json::json!({ "entity_id": 2000, "values": [1.0, 2.0, 3.0, 4.0] }),
    );
    assert_eq!(
        resp.status().as_u16(),
        204,
        "a correctly-sized embedding must still be accepted after the rejections"
    );
    let frontier_live = settled_leader_frontier(&cluster);
    assert!(
        frontier_live > frontier_after,
        "the valid write must advance the frontier the rejected ones left alone \
         ({frontier_after} -> {frontier_live})"
    );
    cluster.wait_converged_all(convergence_budget());
    for &idx in &followers {
        let status = cluster
            .local_status(idx)
            .expect("a live follower must serve /cluster/status/local");
        assert_eq!(
            status["lag_events"].as_u64(),
            Some(0),
            "follower {} must be at lag 0 after the post-rejection write; a halted \
             receiver freezes instead of tracking: {status}",
            cluster.region_name(idx)
        );
        assert!(
            status["applied_events"].as_u64().unwrap_or(0) >= frontier_live,
            "follower {} must have APPLIED up to the new frontier {frontier_live}, not \
             merely report zero lag: {status}",
            cluster.region_name(idx)
        );
    }

    // ── 4. And every node still reports healthy ──
    for idx in 0..NODES {
        let resp = cluster.get(idx, "/health");
        assert_eq!(
            resp.status().as_u16(),
            200,
            "node {} must be healthy after the rejected writes",
            cluster.region_name(idx)
        );
    }
    println!("[poison] all {NODES} nodes healthy, lag 0, frontier {frontier_live}");
}