net-mesh 0.28.0

//! Property tests (plan test strategy): for *any* interleaving of K
//! gangs over J islands, the two invariants hold —
//!
//! 1. **No two `Active` claims share a GPU.** Islands are disjoint GPU
//!    sets and the island is the `ResourceId`, so this reduces to "no
//!    island is held by two gangs"; we assert it on the literal GPU
//!    ids to match the plan's wording.
//! 2. **All-or-none.** Every gang either fully holds its island set
//!    (all `Active`, by it) or holds nothing.
//!
//! The winning subset depends on thread interleaving, but the
//! invariants are interleaving-independent — which is exactly what a
//! property test pins. Gang island-sets are generated by a seeded LCG
//! (deterministic, dependency-free) and run concurrently; several
//! seeds exercise different contention shapes.

use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::Duration;

use crate::adapter::net::behavior::fold::{
    Fold, IslandId, NodeId, ReservationFold, ReservationQuery, ReservationState, UnitId,
};
use crate::adapter::net::behavior::gang::{
    acquire_gang, activate_island, Claimant, GangClaim, GangOutcome,
};
use crate::adapter::net::current_timestamp_micros;
use crate::adapter::net::identity::EntityKeypair;

/// Tiny deterministic PRNG (PCG-style output of an LCG). Dependency-
/// free so the property test is reproducible from a seed.
fn next(state: &mut u64) -> u64 {
    *state = state
        .wrapping_mul(6_364_136_223_846_793_005)
        .wrapping_add(1_442_695_040_888_963_407);
    // Return the high bits (better distributed than the low bits).
    *state >> 33
}

/// Island `i` owns the disjoint unit range `[i*4, i*4+4)`.
///
/// Computed in `u64` with a checked narrowing cast: a bare
/// `island as UnitId` (u64→u32) before the multiply could let two
/// distinct island ids alias the same unit range via truncation, which
/// would silently weaken the disjoint-unit invariant this helper
/// encodes. The test's ids stay small, so the cast never trips.
fn units_of(island: IslandId) -> [UnitId; 4] {
    let base = UnitId::try_from(island * 4).expect("island id too large for test unit range");
    [base, base + 1, base + 2, base + 3]
}

fn holder_of(fold: &Fold<ReservationFold>, island: IslandId) -> Option<(NodeId, bool)> {
    fold.query(ReservationQuery::State(island))
        .first()
        .and_then(|(_, s)| match s {
            ReservationState::Free => None,
            ReservationState::Reserved { holder, .. } => Some((*holder, false)),
            ReservationState::Active { holder, .. } => Some((*holder, true)),
        })
}

fn run_one_interleaving(seed: u64, gangs: usize, islands: u64) {
    let fold = Arc::new(Fold::<ReservationFold>::with_sweep_interval(Duration::ZERO));
    let mut rng = seed;

    // Generate each gang's desired island set (1..=2 distinct islands).
    let gang_sets: Vec<Vec<IslandId>> = (0..gangs)
        .map(|_| {
            let size = 1 + (next(&mut rng) % 2) as usize;
            let mut set: HashSet<IslandId> = HashSet::new();
            while set.len() < size {
                set.insert(next(&mut rng) % islands);
            }
            set.into_iter().collect()
        })
        .collect();

    let deadline = current_timestamp_micros() + 300_000;
    let handles: Vec<_> = gang_sets
        .iter()
        .cloned()
        .map(|want| {
            let fold = fold.clone();
            std::thread::spawn(move || {
                let kp = EntityKeypair::generate();
                let node = kp.entity_id().node_id();
                let mut claimant = Claimant::new(&fold, &kp, node);
                let claim = GangClaim {
                    job: 1,
                    islands: want.clone(),
                    deadline_us: deadline,
                };
                let outcome = acquire_gang(
                    &mut claimant,
                    &claim,
                    1_000_000,
                    current_timestamp_micros,
                    |_| std::thread::sleep(Duration::from_micros(150)),
                )
                .expect("acquire");
                if let GangOutcome::Held(held) = outcome {
                    // Drive each held island Reserved → Active (the
                    // gang "starts compute"). Held all → all activate.
                    for &island in &held {
                        activate_island(&fold, &kp, node, claimant.next_gen(), island, 1).unwrap();
                    }
                    // Hold (no release) so the final snapshot is stable.
                    Some((node, held))
                } else {
                    None
                }
            })
        })
        .collect();

    let results: Vec<Option<(NodeId, Vec<IslandId>)>> =
        handles.into_iter().map(|h| h.join().unwrap()).collect();

    // --- Invariant 2: all-or-none, per gang ---
    for (gang_idx, result) in results.iter().enumerate() {
        match result {
            Some((node, held)) => {
                // The reported set must equal the gang's desired set
                // (full hold), and every island Active by this node.
                let mut want = gang_sets[gang_idx].clone();
                want.sort_unstable();
                want.dedup();
                let mut got = held.clone();
                got.sort_unstable();
                assert_eq!(got, want, "a Held gang must hold its FULL island set");
                for &island in held {
                    assert_eq!(
                        holder_of(&fold, island),
                        Some((*node, true)),
                        "every island of a Held gang must be Active by it",
                    );
                }
            }
            None => { /* held nothing — checked via the holder map below */ }
        }
    }

    // --- Invariant 1: no two Active claims share a GPU ---
    let mut unit_owner: HashMap<UnitId, NodeId> = HashMap::new();
    let mut active_islands = 0usize;
    for island in 0..islands {
        if let Some((holder, true)) = holder_of(&fold, island) {
            active_islands += 1;
            for unit in units_of(island) {
                assert!(
                    unit_owner.insert(unit, holder).is_none(),
                    "unit {unit} claimed by two Active gangs (seed {seed})",
                );
            }
        }
    }

    // Sanity: the set of Active islands is exactly the union of the
    // winners' held sets (no Active island is unaccounted for, i.e. no
    // gang holds an island it didn't report).
    let reported: usize = results
        .iter()
        .filter_map(|r| r.as_ref().map(|(_, h)| h.len()))
        .sum();
    assert_eq!(
        active_islands, reported,
        "every Active island belongs to exactly one reporting gang (seed {seed})",
    );
}

#[test]
fn k_gangs_over_j_islands_preserve_disjoint_active_and_all_or_none() {
    // A handful of seeds → different contention shapes; the invariants
    // hold under every thread interleaving of each.
    for seed in [0x1111_2222, 0xDEAD_BEEF, 0x0BAD_F00D, 0x5EED_5EED] {
        run_one_interleaving(seed, /* gangs */ 6, /* islands */ 5);
    }
}