haematite 0.7.0

//! COMMIT-COLLAPSE §11 (deferred item 3) — `merge_adopt` failure ⇒ UNRECONCILED,
//! driven through the REAL native command loop on the cooperative scheduler.
//!
//! §3/§2.2 M5: `merge_adopt` can fail AFTER it would discard the buffer, at the
//! WAL atomic replacement — either BEFORE the rename (marker RETAINED on disk) or
//! AFTER the rename but before the parent fence (marker ADOPTED on disk while the
//! in-memory root is stale). Either way the native wiring marks the cell
//! UNRECONCILED, so the shard is never fast-pathed or clean-skipped; its NEXT full
//! commit re-runs the reconciling path and converges (unreconciled cleared, clean
//! at a published root).
//!
//! The two WAL-layer fallible points (replacement rename, post-rename parent sync)
//! are exercised through the path-scoped WAL replacement failpoint. The other two
//! §3 points (index rebuild, store barrier) share the SAME native wiring — any
//! `merge_adopt` error routes through `publish_unreconciled` — and the store-barrier
//! error path is separately pinned at the actor level by
//! `expiry_mutation_tests::merge_adopt_failure_preserves_buffer_index_and_oracle`.

use std::collections::VecDeque;
use std::error::Error;
use std::path::PathBuf;
use std::sync::mpsc::{self, Receiver, TryRecvError};
use std::sync::{Arc, Mutex};

use beamr::atom::{Atom, AtomTable};
use beamr::ets::OwnedTerm;
use beamr::module::ModuleRegistry;
use beamr::native::BifRegistryImpl;
use beamr::scheduler::WasmScheduler;
use beamr::term::Term;

use super::handle::{CommandQueue, MergeAdoptReply, ShardCommand, ShardCommandKind, ShardError};
use super::native::{self, ShardNativeHandler};
use crate::db::root_advance::RootAdvanceSeam;
use crate::shard::commit_state::ShardCommitState;
use crate::tree::Hash;
use crate::wal::durable::{WalReplaceFailure, arm_replacement_failure};

type TestResult = Result<(), Box<dyn Error>>;

const MAX_TURNS_PER_REPLY: usize = 16;

fn cooperative_scheduler() -> WasmScheduler {
    let atom_table = Arc::new(AtomTable::with_common_atoms());
    let modules = Arc::new(ModuleRegistry::new());
    let bifs = Arc::new(BifRegistryImpl::new());
    WasmScheduler::new(atom_table, modules, bifs)
}

fn wake() -> OwnedTerm {
    OwnedTerm::immediate(Term::atom(Atom::OK))
}

fn enqueue_wake_and_pump<T>(
    scheduler: &mut WasmScheduler,
    pid: u64,
    commands: &CommandQueue,
    command: ShardCommand,
    reply_rx: &Receiver<T>,
) -> Result<T, Box<dyn Error>> {
    native::lock_queue(commands).push_back(command);
    scheduler.send_owned(pid, &wake())?;
    for _ in 0..MAX_TURNS_PER_REPLY {
        let _exited = scheduler.run_native_until_idle();
        match reply_rx.try_recv() {
            Ok(value) => return Ok(value),
            Err(TryRecvError::Empty) => {}
            Err(TryRecvError::Disconnected) => {
                return Err("shard reply channel disconnected before a reply".into());
            }
        }
    }
    Err("shard produced no reply within the cooperative turn budget".into())
}

struct Harness {
    scheduler: WasmScheduler,
    pid: u64,
    commands: CommandQueue,
    cell: Arc<ShardCommitState>,
    wal_path: PathBuf,
    next_id: u64,
}

impl Harness {
    fn boot() -> Result<(tempfile::TempDir, Self), Box<dyn Error>> {
        let mut scheduler = cooperative_scheduler();
        let dir = tempfile::tempdir()?;
        let store_dir = dir.path().join("store");
        let wal_path = dir.path().join("shard.wal");
        let commands: CommandQueue = Arc::new(Mutex::new(VecDeque::new()));
        let seam = RootAdvanceSeam::new();
        let cell = seam.commit_state(0);
        let factory = ShardNativeHandler::make_factory(
            store_dir,
            wal_path.clone(),
            Arc::clone(&commands),
            crate::tree::TreePolicy::V1_DEFAULT,
            seam.commit_state(0),
        );
        let pid = scheduler.spawn_native_root(factory);
        let _exited = scheduler.run_native_until_idle();
        Ok((
            dir,
            Self {
                scheduler,
                pid,
                commands,
                cell,
                wal_path,
                next_id: 1,
            },
        ))
    }

    fn put(&mut self, key: &[u8], value: &[u8]) -> TestResult {
        let (tx, rx) = mpsc::sync_channel(1);
        let id = self.next_id;
        self.next_id += 1;
        let result: Result<(), ShardError> = enqueue_wake_and_pump(
            &mut self.scheduler,
            self.pid,
            &self.commands,
            ShardCommand {
                id,
                kind: ShardCommandKind::Put {
                    key: key.to_vec(),
                    value: value.to_vec(),
                    ttl: None,
                    reply: tx,
                },
            },
            &rx,
        )?;
        result?;
        Ok(())
    }

    fn commit(&mut self) -> Result<Hash, Box<dyn Error>> {
        let (tx, rx) = mpsc::sync_channel(1);
        let id = self.next_id;
        self.next_id += 1;
        let result: Result<(Hash, _), ShardError> = enqueue_wake_and_pump(
            &mut self.scheduler,
            self.pid,
            &self.commands,
            ShardCommand {
                id,
                kind: ShardCommandKind::Commit { reply: tx },
            },
            &rx,
        )?;
        Ok(result?.0)
    }

    /// Issue an empty-promiser `merge_adopt` (re-adopts the local committed root,
    /// so it reaches the WAL replacement) and return whether it SUCCEEDED — the only
    /// property the item-3 test asserts (the cell carries the reconciliation state).
    fn merge_adopt_local(&mut self) -> Result<bool, Box<dyn Error>> {
        let (tx, rx): (MergeAdoptReply, _) = mpsc::sync_channel(1);
        let id = self.next_id;
        self.next_id += 1;
        native::lock_queue(&self.commands).push_back(ShardCommand {
            id,
            kind: ShardCommandKind::MergeAdopt {
                promisers: Vec::new(),
                reply: tx,
            },
        });
        self.scheduler.send_owned(self.pid, &wake())?;
        for _ in 0..MAX_TURNS_PER_REPLY {
            let _exited = self.scheduler.run_native_until_idle();
            match rx.try_recv() {
                Ok(value) => return Ok(value.is_ok()),
                Err(TryRecvError::Empty) => {}
                Err(TryRecvError::Disconnected) => {
                    return Err("merge_adopt reply disconnected".into());
                }
            }
        }
        Err("merge_adopt produced no reply".into())
    }
}

/// Drive the full item-3 contract for one WAL failure point: a `merge_adopt` that
/// fails at `failure` marks the cell UNRECONCILED (classified dirty, fast path
/// refused), and the NEXT commit re-reconciles and converges to clean.
fn merge_adopt_failure_marks_unreconciled_then_converges(failure: WalReplaceFailure) -> TestResult {
    let (_dir, mut harness) = Harness::boot()?;
    harness.put(b"seed", b"v")?;
    let baseline = harness.commit()?; // local committed root => clean
    assert!(
        !harness.cell.classify().dirty,
        "clean baseline before the merge_adopt failure"
    );

    // Arm the WAL replacement to fail, then run a merge_adopt that reaches it.
    arm_replacement_failure(&harness.wal_path, failure);
    let succeeded = harness.merge_adopt_local()?;
    assert!(
        !succeeded,
        "the injected WAL replacement failure surfaces as a merge_adopt error"
    );

    // The cell is UNRECONCILED — classified dirty, and NOT fast-pathable.
    let snapshot = harness.cell.snapshot();
    assert!(
        snapshot.unreconciled,
        "a merge_adopt failure after the discard marks the cell UNRECONCILED (§3)"
    );
    assert!(
        harness.cell.classify().dirty,
        "an unreconciled shard classifies DIRTY (never clean-skipped)"
    );
    assert!(
        harness.cell.is_unreconciled(),
        "the fast path's unreconciled guard is set — the next commit runs the full path"
    );

    // The NEXT full commit re-reconciles and converges.
    let converged = harness.commit()?;
    let after = harness.cell.snapshot();
    assert!(
        !after.unreconciled && after.committed_gen == after.dirty_gen && after.root.is_some(),
        "the next full commit clears unreconciled and republishes a clean snapshot"
    );
    assert_eq!(
        converged, baseline,
        "convergence returns the reconciled root (the seed data is intact)"
    );
    Ok(())
}

/// Marker-RETAINED branch (fail before the rename): disk keeps the old WAL, the
/// cell is unreconciled, and the next commit converges.
#[test]
fn merge_adopt_before_persist_is_unreconciled_then_converges() -> TestResult {
    merge_adopt_failure_marks_unreconciled_then_converges(WalReplaceFailure::BeforePersist)
}

/// Marker-ADOPTED branch (fail after the rename, before the parent fence): disk
/// carries the adopted marker while the in-memory root is stale — the exact §3
/// divergence — and the next commit still converges from the unreconciled state.
#[test]
fn merge_adopt_after_persist_is_unreconciled_then_converges() -> TestResult {
    merge_adopt_failure_marks_unreconciled_then_converges(
        WalReplaceFailure::AfterPersistBeforeFence,
    )
}