vsdb 17.0.15

use super::*;
use std::{thread, time::Duration};

#[test]
fn dagmapraw_functions() {
    let mut i0 = DagMapRaw::new(None);
    i0.insert("k0", "v0");
    assert_eq!(i0.get("k0").unwrap().as_slice(), "v0".as_bytes());
    assert!(i0.get("k1").is_none());

    let mut i1 = DagMapRaw::new(Some(&mut i0));
    i1.insert("k1", "v1");
    assert_eq!(i1.get("k1").unwrap().as_slice(), "v1".as_bytes());
    assert_eq!(i1.get("k0").unwrap().as_slice(), "v0".as_bytes());

    let mut i2 = DagMapRaw::new(Some(&mut i1));
    i2.insert("k2", "v2");
    assert_eq!(i2.get("k2").unwrap().as_slice(), "v2".as_bytes());
    assert_eq!(i2.get("k1").unwrap().as_slice(), "v1".as_bytes());
    assert_eq!(i2.get("k0").unwrap().as_slice(), "v0".as_bytes());
    i2.insert("k2", "v2x");
    assert_eq!(i2.get("k2").unwrap().as_slice(), "v2x".as_bytes());
    assert_eq!(i2.get("k1").unwrap().as_slice(), "v1".as_bytes());
    assert_eq!(i2.get("k0").unwrap().as_slice(), "v0".as_bytes());
    i2.insert("k1", "v1x");
    assert_eq!(i2.get("k2").unwrap().as_slice(), "v2x".as_bytes());
    assert_eq!(i2.get("k1").unwrap().as_slice(), "v1x".as_bytes());
    assert_eq!(i2.get("k0").unwrap().as_slice(), "v0".as_bytes());
    i2.insert("k0", "v0x");
    assert_eq!(i2.get("k2").unwrap().as_slice(), "v2x".as_bytes());
    assert_eq!(i2.get("k1").unwrap().as_slice(), "v1x".as_bytes());
    assert_eq!(i2.get("k0").unwrap().as_slice(), "v0x".as_bytes());

    // Overlay isolation: parents never see descendant writes.
    assert!(i1.get("k2").is_none());
    assert_eq!(i1.get("k1").unwrap().as_slice(), "v1".as_bytes());
    assert_eq!(i1.get("k0").unwrap().as_slice(), "v0".as_bytes());

    assert!(i0.get("k2").is_none());
    assert!(i0.get("k1").is_none());
    assert_eq!(i0.get("k0").unwrap().as_slice(), "v0".as_bytes());

    // The original owned handles alias the same storage and observe the
    // post-prune state (`Clone` would deep-copy instead).
    let mut head = i2.prune().unwrap();
    thread::sleep(Duration::from_millis(1000));

    assert_eq!(head.get("k2").unwrap().as_slice(), "v2x".as_bytes());
    assert_eq!(head.get("k1").unwrap().as_slice(), "v1x".as_bytes());
    assert_eq!(head.get("k0").unwrap().as_slice(), "v0x".as_bytes());

    // The intermediate mainline node was merged into genesis and cleared.
    assert!(i1.is_dead());
    // The genesis handle sees the merged result (same storage as `head`).
    assert_eq!(i0.get("k2").unwrap().as_slice(), "v2x".as_bytes());

    // prune with deep stack
    for i in 10u8..=255 {
        head.insert(i.to_be_bytes(), i.to_be_bytes());
        head = DagMapRaw::new(Some(&mut head));
    }

    let mut head = head.prune().unwrap();
    thread::sleep(Duration::from_millis(1000));
    assert!(head.parent.get_value().is_none());
    assert!(head.children.iter().next().is_none());

    for i in 10u8..=255 {
        assert_eq!(
            head.get(i.to_be_bytes()).unwrap().as_slice(),
            i.to_be_bytes()
        );
    }

    for i in 0u8..=254 {
        head.remove(i.to_be_bytes());
        assert!(head.get(i.to_be_bytes()).is_none());
    }

    *(head.get_mut(255u8.to_be_bytes()).unwrap()) = 0u8.to_be_bytes().to_vec();
    assert_eq!(
        head.get(255u8.to_be_bytes()).unwrap().as_slice(),
        0u8.to_be_bytes()
    );
}

#[test]
fn test_save_and_from_meta() {
    let mut dag = DagMapRaw::new(None);
    dag.insert("k1", "v1");
    dag.insert("k2", "v2");

    let id = dag.save_meta().unwrap();
    assert_eq!(id, dag.instance_id());

    let restored = DagMapRaw::from_meta(id).unwrap();
    assert_eq!(restored.get("k1").unwrap().as_slice(), "v1".as_bytes());
    assert_eq!(restored.get("k2").unwrap().as_slice(), "v2".as_bytes());
}

/// Postcard serde roundtrip for DagMapRaw (hand-written tuple serde, 3 fields).
#[test]
fn test_serde_roundtrip() {
    let mut dag = DagMapRaw::new(None);
    dag.insert("alpha", "A");
    dag.insert("beta", "B");

    let bytes = postcard::to_allocvec(&dag).unwrap();
    let restored: DagMapRaw = postcard::from_bytes(&bytes).unwrap();

    assert_eq!(restored.get("alpha").unwrap().as_slice(), b"A");
    assert_eq!(restored.get("beta").unwrap().as_slice(), b"B");
}

/// from_meta nonexistent.
#[test]
fn test_from_meta_nonexistent() {
    assert!(DagMapRaw::from_meta(u64::MAX).is_err());
}

/// Restore from meta, mutate, verify shared storage.
#[test]
fn test_meta_restore_then_mutate() {
    let mut dag = DagMapRaw::new(None);
    dag.insert("k1", "v1");

    let id = dag.save_meta().unwrap();
    let mut restored = DagMapRaw::from_meta(id).unwrap();

    restored.insert("k2", "v2");
    assert_eq!(dag.get("k2").unwrap().as_slice(), b"v2");
}

#[test]
fn serde_rejects_mixed_namespace_components() {
    let ns = crate::Namespace::create().unwrap();
    let data = MapxRaw::new();
    let parent: Orphan<Option<DagMapRaw>> = Orphan::new_in(&ns, None);
    let children: MapxOrdRawKey<DagMapRaw> = MapxOrdRawKey::new();
    let bytes = postcard::to_allocvec(&(data, parent, children)).unwrap();

    assert!(postcard::from_bytes::<DagMapRaw>(&bytes).is_err());
}

#[test]
#[should_panic(expected = "empty value is a tombstone")]
fn insert_empty_value_panics() {
    let mut dag = DagMapRaw::new(None);
    dag.insert("empty", []);
}

#[test]
fn destroy_unlinks_from_parent() {
    let mut parent = DagMapRaw::new(None);
    let mut child = DagMapRaw::new(Some(&mut parent));

    assert!(!parent.no_children());
    child.destroy();
    assert!(parent.no_children());
}

/// Regression: `destroy()` must clear a node's own data/parent (and all
/// owned descendants) BEFORE unregistering it from the parent's
/// registry — never the other way around (the previous ordering removed
/// the registry entry first, so a crash right after that write, but
/// before data/parent clearing landed, permanently orphaned the node's
/// still-intact storage with no registry path back to it).
///
/// This manually replicates only the "clear data + null parent" half of
/// `destroy()`'s new ordering (simulating a crash before the final
/// parent-registry unlink runs) and confirms the node remains
/// discoverable through the parent's still-intact registry entry, and is
/// recognized as reclaimable residue (`owned_or_residue`'s `None`-parent
/// arm) rather than becoming a permanently unreachable leak.
#[test]
fn destroy_retry_after_parent_null_unlinks() {
    let mut parent = DagMapRaw::new(None);
    let mut child = DagMapRaw::new(Some(&mut parent));
    child.insert("k", "v");

    // Crash after the side key is durable and the slot is nulled, before
    // the parent registry unlink.
    let linked = child.parent.get_value().unwrap();
    child
        .parent
        .set_aux(super::DESTROY_PARENT_KEY, &Some(linked));
    *child.parent.get_mut() = None;
    child.data.clear();
    assert!(!parent.no_children());

    child.destroy();
    assert!(parent.no_children());
    assert!(child.get("k").is_none());
}

#[test]
fn destroy_interrupted_before_unlink_leaves_reclaimable_residue() {
    let mut parent = DagMapRaw::new(None);
    let mut child = DagMapRaw::new(Some(&mut parent));
    child.insert("k", "v");

    assert!(!parent.no_children());

    // Simulate a crash after data/parent clearing but before the final
    // parent-registry unlink step (mirrors `destroy()`'s internal order).
    *child.parent.get_mut() = None;
    child.data.clear();

    // The registry entry must still exist (not yet removed) ...
    assert!(!parent.no_children());
    // ... and must be recognized as reclaimable residue, not foreign.
    let (_, residue) = parent.children.iter().next().unwrap();
    assert!(DagMapRaw::owned_or_residue(parent.instance_id(), &residue));

    // A subsequent prune sweep (the same registry-driven mechanism
    // `prune` itself uses) reclaims the residue, exactly as it would
    // after a real crash-then-restart.
    let empty: &[RawBytes] = &[];
    parent.prune_children_exclude(empty);
    assert!(parent.no_children());
}

/// Regression: `prune()`'s crash-safety flush barriers must be scoped to
/// the DAG's own namespace (`self.namespace().flush()`), not the whole
/// process (the previous `vsdb_flush()` calls flushed the default
/// namespace plus every other open namespace, so a flush failure in any
/// unrelated namespace could panic a `prune()` on a perfectly healthy
/// one). Exercising `prune()` end-to-end on a DAG placed in an explicit
/// non-default namespace only compiles and behaves correctly if the
/// barriers are genuinely namespace-scoped.
#[test]
fn prune_works_in_a_non_default_namespace() {
    let ns = crate::common::Namespace::create().unwrap();
    let mut head = DagMapRaw::new_in(&ns, None);
    head.insert("root", "value");

    let mut child = DagMapRaw::new_in(&ns, Some(&mut head));
    child.insert("leaf", "cv");

    let pruned = head.prune().unwrap();
    assert_eq!(pruned.get("root").unwrap().as_slice(), b"value");
    assert_eq!(child.get("root").unwrap().as_slice(), b"value");
    assert_eq!(child.get("leaf").unwrap().as_slice(), b"cv");
}

#[test]
fn deep_acyclic_chain_remains_readable_and_prunable() {
    let mut head = DagMapRaw::new(None);
    head.insert("root", "value");

    for _ in 0..1030 {
        head = DagMapRaw::new(Some(&mut head));
    }

    assert_eq!(head.get("root").unwrap().as_slice(), b"value");
    let pruned = head.prune().unwrap();
    assert_eq!(pruned.get("root").unwrap().as_slice(), b"value");
}

/// Save meta of a DagMapRaw with parent-child relationship,
/// restore, and verify the lineage is intact.
#[test]
fn test_meta_with_parent_child() {
    let mut i0 = DagMapRaw::new(None);
    i0.insert("base", "v0");

    let mut i1 = DagMapRaw::new(Some(&mut i0));
    i1.insert("child", "v1");

    let id = i1.save_meta().unwrap();
    let restored = DagMapRaw::from_meta(id).unwrap();

    // Child data
    assert_eq!(restored.get("child").unwrap().as_slice(), b"v1");
    // Inherited from parent
    assert_eq!(restored.get("base").unwrap().as_slice(), b"v0");
}

#[test]
fn test_prune_with_side_branches() {
    let mut i0 = DagMapRaw::new(None);
    i0.insert("k0", "v0");

    let mut mid = DagMapRaw::new(Some(&mut i0));

    // Side-branch child of mid
    let mut side = DagMapRaw::new(Some(&mut mid));
    side.insert("k_side", "v_side");

    // Another side-branch child of mid
    let mut side2 = DagMapRaw::new(Some(&mut mid));
    side2.insert("k_side2", "v_side2");

    // Head node on mainline
    let mut head = DagMapRaw::new(Some(&mut mid));
    head.insert("k_head", "v_head");

    // Prune
    let pruned = head.prune().unwrap();

    assert_eq!(pruned.get("k0").unwrap().as_slice(), b"v0");
    assert_eq!(pruned.get("k_head").unwrap().as_slice(), b"v_head");

    // Side branches of intermediate nodes should be destroyed
    assert!(side.get("k_side").is_none());
    assert!(side2.get("k_side2").is_none());
}

#[test]
fn destroy_sibling_preserves_other_siblings_and_parent() {
    // Destroying one child must leave the parent and the other children
    // intact — each node owns its parent slot, so the unlink is local.
    let mut p = DagMapRaw::new(None);
    p.insert("shared", "pval");

    let mut c1 = DagMapRaw::new(Some(&mut p));
    c1.insert("c1", "v1");
    let c2 = DagMapRaw::new(Some(&mut p));

    assert_eq!(c1.get("shared").unwrap().as_slice(), b"pval");
    assert_eq!(c2.get("shared").unwrap().as_slice(), b"pval");

    c1.destroy();
    assert!(c1.get("c1").is_none());

    // Parent must be untouched.
    assert_eq!(p.get("shared").unwrap().as_slice(), b"pval");
    // Surviving sibling must still reach the inherited parent data.
    assert_eq!(c2.get("shared").unwrap().as_slice(), b"pval");
}

#[test]
fn selective_prune_ids_are_publicly_discoverable() {
    let mut parent = DagMapRaw::new(None);
    let child1 = DagMapRaw::new(Some(&mut parent));
    let child2 = DagMapRaw::new(Some(&mut parent));

    let id1 = parent.child_id(&child1).unwrap();
    let id2 = parent.child_id(&child2).unwrap();
    assert_ne!(id1, id2);
    assert_eq!(parent.child_ids().len(), 2);

    parent.prune_children_include(&[id1]);
    assert!(child1.is_dead());
    assert!(!child2.is_dead());
    assert_eq!(parent.child_ids(), vec![id2]);
}

#[test]
fn destroyed_node_does_not_serve_inherited_reads() {
    // After destroy(), no handle of the node may fall through to the
    // parent and return inherited data — the unlink is persisted in the
    // node's own parent slot, so clones and restored handles see it too.
    let mut parent = DagMapRaw::new(None);
    parent.insert("k", "v");

    let mut child = DagMapRaw::new(Some(&mut parent));
    // Before destroy the child inherits the parent's value.
    assert_eq!(child.get("k").unwrap().as_slice(), b"v");

    // A shadow (aliasing handle) taken BEFORE the destroy call.
    // SAFETY: used strictly after the destroy completes; accesses are
    // sequential on one thread, satisfying SWMR.
    let stale_alias = unsafe { child.shadow() };

    child.destroy();
    // Neither the destroying handle nor the stale alias serves
    // inherited reads afterwards — the parent unlink is persisted in
    // the node's own parent slot.
    assert!(child.get("k").is_none());
    assert!(stale_alias.get("k").is_none());

    // The parent itself is untouched.
    assert_eq!(parent.get("k").unwrap().as_slice(), b"v");
}

#[test]
fn destroy_is_visible_to_meta_restored_handles() {
    // A handle restored via from_meta AFTER the node was destroyed must
    // observe the destroyed state (cleared data + nulled parent link).
    let mut parent = DagMapRaw::new(None);
    parent.insert("k", "v");

    let mut child = DagMapRaw::new(Some(&mut parent));
    let id = child.save_meta().unwrap();

    child.destroy();

    let restored = DagMapRaw::from_meta(id).unwrap();
    assert!(restored.get("k").is_none());
    assert!(restored.is_dead());
}

#[test]
fn destroy_deep_child_chain_does_not_overflow_stack() {
    // Regression: destroy() recursed once per descendant generation and
    // overflowed the stack on deep DAGs. Build a deep child chain rooted
    // at `genesis` and destroy it iteratively.
    let mut genesis = DagMapRaw::new(None);
    genesis.insert("root", "value");

    // SAFETY: the shadow is only used as an attachment cursor while
    // building the chain; all accesses are sequential on one thread,
    // satisfying SWMR.
    let mut cur = unsafe { genesis.shadow() };
    for _ in 0..5000 {
        cur = DagMapRaw::new(Some(&mut cur));
    }

    // `genesis` owns the full descendant chain through its children map.
    // Destroying it must not overflow the stack.
    genesis.destroy();
    assert!(genesis.get("root").is_none());
}

// ---------------------------------------------------------------------------
// Crash-safety tests for prune_mainline.
//
// The prune is phased (collect → destroy branches → merge → re-parent →
// clear); a crash can stop it between any two writes.  These tests drive
// the private phase methods directly to freeze the on-disk state at each
// phase boundary and assert the canonical views (head / genesis meta /
// surviving children) stay value-exact — the contract documented on
// `prune`.
// ---------------------------------------------------------------------------

/// Chain: genesis(k0,kg) → i1(k1, kill kg) → head(k2, k1 override).
/// Returns (genesis, i1, head).
fn build_prune_fixture() -> (DagMapRaw, DagMapRaw, DagMapRaw) {
    let mut genesis = DagMapRaw::new(None);
    genesis.insert("k0", "v0");
    genesis.insert("kg", "doomed");

    let mut i1 = DagMapRaw::new(Some(&mut genesis));
    i1.insert("k1", "v1");
    i1.remove("kg"); // tombstone over a genesis key

    let mut head = DagMapRaw::new(Some(&mut i1));
    head.insert("k2", "v2");
    head.insert("k1", "v1x"); // overrides i1

    (genesis, i1, head)
}

/// Asserts the full merged (post-prune) view on a handle.
fn assert_merged_view(h: &DagMapRaw) {
    assert_eq!(h.get("k0").unwrap().as_slice(), b"v0");
    assert_eq!(h.get("k1").unwrap().as_slice(), b"v1x");
    assert_eq!(h.get("k2").unwrap().as_slice(), b"v2");
    assert!(h.get("kg").is_none());
}

#[test]
fn prune_crash_mid_merge_head_view_is_exact_and_rerun_converges() {
    let (genesis, i1, head) = build_prune_fixture();

    // Simulate a crash in the middle of phase 2: only the oldest
    // mainline node (i1) has been staged; nothing
    // cleared, children not yet re-parented.
    let linebuf = head.prune_collect_mainline().unwrap();
    assert_eq!(linebuf.len(), 2); // [i1, genesis]
    let mut staged = StagedRows::new();
    DagMapRaw::prune_fold_node(&mut staged, &linebuf[0]);

    // The head's view must be exactly the pre-prune view: overlay
    // resolution stops above the genesis for every folded key.
    assert_merged_view(&head); // (pre-prune view == merged view by construction)
    // Before publication the genesis retains its complete original view.
    assert_eq!(genesis.get("kg").as_deref(), Some(b"doomed".as_slice()));
    assert!(genesis.get("k1").is_none());
    drop(staged); // abandoned batch: no durable changes
    assert!(head.get("kg").is_none());
    // … and non-folded nodes still hold their own data.
    assert_eq!(i1.data.get("k1").unwrap().as_slice(), b"v1");

    // Recovery: the chain is fully intact, so simply re-running the
    // prune converges to the complete merged state.
    let pruned = head.prune().unwrap();
    assert_merged_view(&pruned);
    assert!(pruned.parent.get_value().is_none());
    assert!(i1.is_dead());
}

#[test]
fn prune_crash_after_merge_genesis_meta_sees_merged_state() {
    let (genesis, _i1, head) = build_prune_fixture();
    let genesis_id = genesis.save_meta().unwrap();

    // Freeze right after phase 2 (merge complete, nothing cleared,
    // children not yet re-parented).
    let mut linebuf = head.prune_collect_mainline().unwrap();
    head.prune_merge_into_genesis(&mut linebuf).unwrap();

    // The genesis — under its pre-prune instance ID — already serves the
    // complete merged view; the head still serves its exact view too.
    let restored = DagMapRaw::from_meta(genesis_id).unwrap();
    assert_merged_view(&restored);
    assert_merged_view(&head);

    // Re-run converges and keeps the genesis identity.
    let pruned = head.prune().unwrap();
    assert_eq!(pruned.instance_id(), genesis_id);
    assert_merged_view(&DagMapRaw::from_meta(genesis_id).unwrap());
}

#[test]
fn prune_retry_does_not_regress_reparented_child() {
    let (_genesis, _i1, mut head) = build_prune_fixture();
    let mut child = DagMapRaw::new(Some(&mut head));
    let child_id = head.child_id(&child).unwrap();
    let mut linebuf = head.prune_collect_mainline().unwrap();
    head.prune_merge_into_genesis(&mut linebuf).unwrap();
    let genesis = linebuf.last_mut().unwrap();
    // SAFETY: serialized immediately; all accesses are sequential.
    *child.parent.get_mut() = Some(unsafe { genesis.shadow() });
    genesis.children.insert(child_id, &child);
    assert_merged_view(&child);

    // A second interruption while retrying the oldest part of the fold
    // must not roll back values already visible through the new parent.
    let mut staged = StagedRows::new();
    DagMapRaw::prune_fold_node(&mut staged, &linebuf[0]);
    assert_merged_view(&child);
    drop(staged);
    assert_merged_view(&head.prune().unwrap());
    assert_merged_view(&child);
}

#[test]
fn prune_crash_mid_reparent_both_children_views_exact() {
    let (_genesis, _i1, mut head) = build_prune_fixture();

    let mut c1 = DagMapRaw::new(Some(&mut head));
    c1.insert("c1", "w1");
    let mut c2 = DagMapRaw::new(Some(&mut head));
    c2.insert("c2", "w2");

    // Phases 0-2, then flip ONLY c1 (simulated crash inside phase 3).
    let mut linebuf = head.prune_collect_mainline().unwrap();
    head.prune_merge_into_genesis(&mut linebuf).unwrap();
    let genesis = linebuf.last_mut().unwrap();
    // Exactly what phase 3 does for one child: re-point the parent slot
    // and re-register under the genesis with the child's REAL registry id.
    let c1_reg_id = head
        .children
        .iter()
        .find(|(_, c)| c.instance_id() == c1.instance_id())
        .unwrap()
        .0;
    // SAFETY: the shadow is written into c1's parent slot and only read
    // afterwards; all accesses are sequential on one thread (SWMR).
    *c1.parent.get_mut() = Some(unsafe { genesis.shadow() });
    genesis.children.insert(&c1_reg_id, &c1);

    // Both routes — via the merged genesis (c1) and via the intact old
    // chain (c2) — must resolve every key identically.
    for c in [&c1, &c2] {
        assert_eq!(c.get("k0").unwrap().as_slice(), b"v0");
        assert_eq!(c.get("k1").unwrap().as_slice(), b"v1x");
        assert_eq!(c.get("k2").unwrap().as_slice(), b"v2");
        assert!(c.get("kg").is_none());
    }
    assert_eq!(c1.get("c1").unwrap().as_slice(), b"w1");
    assert_eq!(c2.get("c2").unwrap().as_slice(), b"w2");

    // Recovery re-run: the already-flipped child is double-registered
    // under the genesis; phase 1 must NOT destroy it (pending_reparent
    // guard), and phase 3 re-flips it idempotently.
    let pruned = head.prune().unwrap();
    assert_merged_view(&pruned);
    assert_eq!(c1.get("c1").unwrap().as_slice(), b"w1");
    assert_eq!(c2.get("c2").unwrap().as_slice(), b"w2");
    assert_eq!(c1.get("k1").unwrap().as_slice(), b"v1x");
    assert_eq!(c2.get("k1").unwrap().as_slice(), b"v1x");
}

/// Run prune through reparent, then write the clearing marker the
/// production path fences before any clear. Tests then tear one side
/// of that clear and retry `prune`.
fn prepare_marked_clear(head: &mut DagMapRaw) {
    let mut linebuf = head.prune_collect_mainline().unwrap();
    let mainline_ids: Vec<InstanceId> = {
        let mut ids = vec![head.instance_id()];
        ids.extend(linebuf.iter().map(|n| n.instance_id()));
        ids
    };
    let pending: std::collections::HashSet<vsdb_core::common::RawBytes> =
        head.children.iter().map(|(id, _)| id).collect();
    DagMapRaw::prune_destroy_side_branches(&mut linebuf, &mainline_ids, &pending);
    head.prune_merge_into_genesis(&mut linebuf).unwrap();
    head.prune_reparent_children(linebuf.last_mut().unwrap());
    head.mark_consumed_clearing(&mut linebuf);
}

#[test]
fn prune_retry_completes_partial_clearing_markers() {
    // Cover a process crash after only the head marker, as well as
    // non-contiguous marker sets left by independently recovered shards.
    for marker_mask in 0..4 {
        let (genesis, i1, mut previous_head) = build_prune_fixture();
        let mut head = DagMapRaw::new(Some(&mut previous_head));
        let head_id = head.save_meta().unwrap();
        let mut survivor = DagMapRaw::new(Some(&mut head));
        survivor.insert("survivor", "value");

        let mut linebuf = head.prune_collect_mainline().unwrap();
        head.prune_merge_into_genesis(&mut linebuf).unwrap();
        head.namespace().flush();
        head.prune_reparent_children(linebuf.last_mut().unwrap());
        head.namespace().flush();

        // SAFETY: only serialized into recovery markers; all accesses
        // to the fixture and restored aliases are sequential.
        let marker = Some(unsafe { genesis.shadow() });
        head.parent.set_aux(PRUNE_CLEARING_KEY, &marker);
        for (bit, node) in linebuf[..2].iter_mut().enumerate() {
            if marker_mask & (1 << bit) != 0 {
                node.parent.set_aux(PRUNE_CLEARING_KEY, &marker);
            }
        }
        head.namespace().flush();

        // Retry must consume every intermediate, including an unmarked
        // ancestor that otherwise hides the rest of the marked chain.
        let pruned = DagMapRaw::from_meta(head_id).unwrap().prune().unwrap();
        assert_eq!(pruned.instance_id(), genesis.instance_id());
        assert_merged_view(&pruned);
        for consumed in [&head, &previous_head, &i1] {
            assert!(consumed.is_dead(), "partial marker set {marker_mask}");
            assert!(consumed.no_children());
            assert!(consumed.get("k1").is_none());
        }
        assert_merged_view(&survivor);
        assert_eq!(survivor.get("survivor").as_deref(), Some(&b"value"[..]));
        assert_eq!(pruned.children.iter().count(), 1);

        // A subsequent retry still resolves to the same surviving root.
        let retried = DagMapRaw::from_meta(head_id).unwrap().prune().unwrap();
        assert_eq!(retried.instance_id(), genesis.instance_id());
        assert_merged_view(&retried);
        assert_merged_view(&survivor);
        assert_eq!(retried.children.iter().count(), 1);
    }
}

thread_local! {
    static CLEAR_STEPS_LEFT: std::cell::Cell<Option<usize>> = const { std::cell::Cell::new(None) };
}

pub(super) fn after_clear_step() {
    CLEAR_STEPS_LEFT.with(|steps| {
        if let Some(left) = steps.get() {
            steps.set(left.checked_sub(1));
            assert!(left > 0, "interrupted consumed-node cleanup");
        }
    });
}

#[test]
fn prune_cleanup_survives_repeated_interruptions() {
    use std::panic::{AssertUnwindSafe, catch_unwind};

    for already_marked in [false, true] {
        for cut in 0..8 {
            let mut ancestors = vec![DagMapRaw::new(None)];
            for i in 1..5u8 {
                let mut node = DagMapRaw::new(ancestors.last_mut());
                node.insert([i], [i]);
                ancestors.push(node);
            }
            let mut head = ancestors.pop().unwrap();
            let mut survivor = DagMapRaw::new(Some(&mut head));
            survivor.insert("survivor", "value");
            if already_marked {
                prepare_marked_clear(&mut head);
            }
            let head_id = head.save_meta().unwrap();
            head.namespace().flush();

            // Interrupt normal cleanup or its retry after each data or
            // registry clear, then restore the consumed head.
            CLEAR_STEPS_LEFT.with(|steps| steps.set(Some(cut)));
            let interrupted = catch_unwind(AssertUnwindSafe(|| head.prune()));
            CLEAR_STEPS_LEFT.with(|steps| steps.set(None));
            assert!(interrupted.is_err(), "cleanup cut {cut} was not reached");
            ancestors[0].namespace().flush();

            if cut < 4 {
                // No discovery edge may disappear during the data pass.
                for node in &ancestors {
                    assert!(!node.no_children());
                }
            } else {
                // Once registry removal starts, even a different shard's
                // early unlink cannot strand any user data.
                for node in &ancestors[1..] {
                    assert!(node.data.iter().next().is_none());
                    assert!(node.parent.get_value().is_none());
                }
            }

            let recovered = DagMapRaw::from_meta(head_id).unwrap().prune().unwrap();
            assert_eq!(recovered.instance_id(), ancestors[0].instance_id());
            for i in 1..5u8 {
                assert_eq!(recovered.get([i]), Some(vec![i]));
                assert_eq!(survivor.get([i]), Some(vec![i]));
            }
            assert_eq!(
                survivor.get("survivor").as_deref(),
                Some(b"value".as_slice())
            );
            for node in &ancestors[1..] {
                assert!(node.is_dead(), "cut {cut} stranded an intermediate node");
                assert!(node.no_children());
            }
            let head = DagMapRaw::from_meta(head_id).unwrap();
            assert!(head.is_dead());
            assert!(head.no_children());
            assert_eq!(recovered.children.iter().count(), 1);
        }
    }
}

#[test]
fn prune_retry_after_parent_null_returns_genesis() {
    let (genesis, i1, mut head) = build_prune_fixture();
    let genesis_id = genesis.instance_id();
    let mut child = DagMapRaw::new(Some(&mut head));
    child.insert("c", "cv");
    prepare_marked_clear(&mut head);

    // Parent-null landed; children and data did not.
    *head.parent.get_mut() = None;

    let pruned = head.prune().unwrap();
    assert_eq!(pruned.instance_id(), genesis_id);
    assert_merged_view(&pruned);
    assert_eq!(child.get("c").unwrap().as_slice(), b"cv");
    assert_eq!(child.get("k2").unwrap().as_slice(), b"v2");
    assert!(i1.is_dead());
}

#[test]
fn prune_retry_after_data_clear_does_not_regress_genesis() {
    let (genesis, i1, mut head) = build_prune_fixture();
    let genesis_id = genesis.instance_id();
    let mut child = DagMapRaw::new(Some(&mut head));
    child.insert("c", "cv");
    prepare_marked_clear(&mut head);

    // Inverse tear: data and children clears landed, parent-null did not.
    head.children.clear();
    head.data.clear();

    let pruned = head.prune().unwrap();
    assert_eq!(pruned.instance_id(), genesis_id);
    assert_merged_view(&pruned);
    assert_merged_view(&child);
    assert_eq!(child.get("c").unwrap().as_slice(), b"cv");
    assert!(i1.is_dead());
}

#[test]
fn prune_interrupted_before_clear_residue_is_reclaimed_by_next_prune() {
    let (genesis, i1, mut head) = build_prune_fixture();
    head.insert("k_head_only", "hv");

    let mut child = DagMapRaw::new(Some(&mut head));
    child.insert("c", "cv");

    // Crash right after phase 3: merge + re-parent done, clears never ran.
    let mut linebuf = head.prune_collect_mainline().unwrap();
    let mainline_ids: Vec<InstanceId> = {
        let mut ids = vec![head.instance_id()];
        ids.extend(linebuf.iter().map(|n| n.instance_id()));
        ids
    };
    let pending: std::collections::HashSet<vsdb_core::common::RawBytes> =
        head.children.iter().map(|(id, _)| id).collect();
    DagMapRaw::prune_destroy_side_branches(&mut linebuf, &mainline_ids, &pending);
    head.prune_merge_into_genesis(&mut linebuf).unwrap();
    head.prune_reparent_children(linebuf.last_mut().unwrap());

    // Canonical recovery point: the surviving child (now under genesis).
    assert_eq!(child.get("c").unwrap().as_slice(), b"cv");
    assert_eq!(child.get("k_head_only").unwrap().as_slice(), b"hv");

    // The old head/intermediate still hold residue data (leak, not
    // corruption).  The NEXT prune — from the surviving child — must
    // reclaim them: they are side branches of the new mainline now.
    assert!(head.data.iter().next().is_some());
    let pruned2 = child.prune().unwrap();
    assert_eq!(pruned2.instance_id(), genesis.instance_id());

    assert!(head.is_dead());
    assert!(i1.is_dead());
    assert_eq!(pruned2.get("c").unwrap().as_slice(), b"cv");
    assert_eq!(pruned2.get("k_head_only").unwrap().as_slice(), b"hv");
    assert_eq!(pruned2.get("k1").unwrap().as_slice(), b"v1x");
}

#[test]
fn destroy_does_not_follow_stale_foreign_registry_entries() {
    // The children registry is an index, not ownership: destroying a node
    // must not clear a child whose own parent slot points elsewhere
    // (a stale double-registration left by an interrupted prune).
    let mut real_parent = DagMapRaw::new(None);
    real_parent.insert("p", "pv");
    let mut child = DagMapRaw::new(Some(&mut real_parent));
    child.insert("c", "cv");

    let mut stranger = DagMapRaw::new(None);
    stranger.insert("s", "sv");
    // Stale registration of `child` under `stranger` WITHOUT re-pointing
    // the child's parent slot.
    stranger.children.insert([0xffu8; 16], &child);

    stranger.destroy();

    // The stranger died; the foreign child survived untouched.
    assert!(stranger.is_dead());
    assert_eq!(child.get("c").unwrap().as_slice(), b"cv");
    assert_eq!(child.get("p").unwrap().as_slice(), b"pv");
    assert!(!real_parent.no_children());
}

#[test]
fn is_dead_recognizes_tombstoned_entries() {
    // `remove()` writes an empty-value tombstone rather than physically
    // deleting the entry (see `insert`'s doc comment on the tombstone
    // convention), so `is_dead()` must not be fooled by a backing store
    // that still has iterable (but empty) entries.
    let mut node = DagMapRaw::new(None);
    assert!(node.is_dead());

    node.insert("only", "v");
    assert!(!node.is_dead());
    assert_eq!(node.get("only").unwrap().as_slice(), b"v");

    node.remove("only");
    assert!(node.get("only").is_none());
    // The tombstone left `node.data` non-empty (one empty-value entry),
    // but the node has no live data, parent, or children.
    assert!(node.is_dead());

    // Multiple keys, only some removed: still not dead.
    node.insert("a", "va");
    node.insert("b", "vb");
    node.remove("a");
    assert!(!node.is_dead());
    node.remove("b");
    assert!(node.is_dead());
}

#[test]
fn get_mut_tombstone_edit_during_panic_does_not_abort() {
    use std::panic::{AssertUnwindSafe, catch_unwind};

    let mut d = DagMapRaw::new(None);
    d.insert("k", "v");
    // Before the fix, the guard's tombstone assert fired while unwinding:
    // a second panic, i.e. a process abort.
    let r = catch_unwind(AssertUnwindSafe(|| {
        let mut g = d.get_mut("k").unwrap();
        g.clear();
        panic!("interrupted edit");
    }));
    assert!(r.is_err());
    assert_eq!(d.get("k").unwrap().as_slice(), b"v");
}
#[test]
fn clone_prune_preserves_original_graph() {
    let mut root = DagMapRaw::new(None);
    root.insert(b"inherited", b"root");
    let mut head = DagMapRaw::new(Some(&mut root));
    head.insert(b"head", b"value");
    let mut child = DagMapRaw::new(Some(&mut head));
    child.insert(b"child", b"value");
    let child_id = head.child_id(&child).unwrap();

    let copy = head.clone();
    assert_ne!(copy.instance_id(), head.instance_id());
    assert_eq!(copy.namespace().id(), head.namespace().id());
    assert_eq!(copy.get(b"inherited"), Some(b"root".to_vec()));
    assert_eq!(copy.child_ids(), vec![child_id.clone()]);
    let copied_child = copy.children.get(&child_id).unwrap();
    assert_ne!(copied_child.instance_id(), child.instance_id());

    let mut copied_root = copy.prune().unwrap();
    assert_ne!(copied_root.instance_id(), root.instance_id());
    assert_eq!(copied_child.get(b"head"), Some(b"value".to_vec()));
    assert_eq!(copied_child.get(b"child"), Some(b"value".to_vec()));
    assert_eq!(head.get(b"head"), Some(b"value".to_vec()));
    assert_eq!(child.get(b"child"), Some(b"value".to_vec()));
    assert!(root.child_id(&head).is_some());

    copied_root.destroy();
    assert!(copied_child.is_dead());
    assert_eq!(child.get(b"child"), Some(b"value".to_vec()));
    assert_eq!(child.get(b"inherited"), Some(b"root".to_vec()));
}

#[test]
fn clone_remaps_prune_recovery_markers() {
    let mut root = DagMapRaw::new(None);
    let mut head = DagMapRaw::new(Some(&mut root));
    head.insert(b"key", b"value");
    let bytes = postcard::to_allocvec(&head).unwrap();
    let original_root = head.prune().unwrap();
    let consumed: DagMapRaw = postcard::from_bytes(&bytes).unwrap();

    let copied_root = consumed.clone().prune().unwrap();
    assert_ne!(copied_root.instance_id(), original_root.instance_id());
    assert_eq!(copied_root.get(b"key"), Some(b"value".to_vec()));
    assert_eq!(original_root.get(b"key"), Some(b"value".to_vec()));
}

#[test]
fn prune_retry_visits_repeated_marked_nodes_once() {
    let mut genesis = DagMapRaw::new(None);
    genesis.insert(b"base", b"base-value");
    let mut head = DagMapRaw::new(Some(&mut genesis));
    head.insert(b"head", b"head-value");
    let survivor = DagMapRaw::new(Some(&mut head));
    let bytes = postcard::to_allocvec(&head).unwrap();
    let expected = head.prune().unwrap().instance_id();

    // Component deserialization can retain a registry alias. A repeated
    // marked entry must not prevent an already-completed prune from retrying.
    let consumed: DagMapRaw = postcard::from_bytes(&bytes).unwrap();
    let (_, _, mut children): (
        MapxRaw,
        Orphan<Option<DagMapRaw>>,
        MapxOrdRawKey<DagMapRaw>,
    ) = postcard::from_bytes(&bytes).unwrap();
    children.insert(b"repeat-a", &consumed);
    children.insert(b"repeat-b", &consumed);

    let recovered = consumed.prune().unwrap();
    assert_eq!(recovered.instance_id(), expected);
    assert_eq!(recovered.get(b"base").unwrap(), b"base-value");
    assert_eq!(recovered.get(b"head").unwrap(), b"head-value");
    assert_eq!(survivor.get(b"head").unwrap(), b"head-value");
    assert!(children.iter().next().is_none());
}