haematite 0.7.0

//! `migrate_chunking` — the explicit, offline, STAMP-FIRST fenced migration
//! transaction (CHUNKING-POLICY.md §4.2).
//!
//! Converts a v1 (count-target) directory to v2 (byte-aware) — or runs the
//! durable reverse — as ONE whole-directory rebuild under the A4 writer lock.
//! There is no in-place, no background, and no partial (per-shard) migration:
//! one directory, one policy, always (§4.2 non-goals).
//!
//! ## The seven steps (§4.2), in order
//!
//! 1. Acquire the A4 writer lock (offline-only, like the vacuum). A live writer
//!    ⇒ [`DatabaseError::DataDirLocked`]; this call never blocks.
//! 2. **STAMP FIRST** into a fenced migrating state: one atomic config rewrite
//!    to `format_version = 2` + the v2 policy block + the migration fence
//!    ([`config::install_forward_fence`]). The fence goes up
//!    BEFORE any root moves — the r1 B2 fold (stamp-LAST durably published v2
//!    roots under a v1 stamp). From here a crash reopens fenced: normal opens
//!    refuse [`DatabaseError::MigrationInProgress`]; an old binary sees
//!    `format_version = 2` and refuses [`DatabaseError::FormatVersionTooNew`].
//! 3. Per shard `0..shard_count`: stream every entry in key order from the
//!    current root (reads are policy-free), rebuild from empty via ONE
//!    `batch_mutate_owned` under the v2 policy (the canonicity witness — HI
//!    across commit batching), and `DurableWal::commit` the new root.
//! 4. Rebuild every branch HEAD the same way and advance its record (anchors,
//!    snapshots, and commit-log roots stay v1-shaped read-only pins).
//! 5. (v1-anchor fork crossing — a read-time rebuild, not part of this
//!    transaction; documented at the branch/fork API.)
//! 6. Remove the fence LAST ([`config::finalize_v2`]).
//! 7. **Abort is migration in reverse, with its own durable direction**: the
//!    fence's target is flipped to v1 BEFORE any reverse root moves
//!    ([`config::flip_to_abort`]); `format_version` STAYS 2 until
//!    every root and head is v1-shaped, then the exact stable v1 envelope is
//!    installed ([`config::finalize_v1`]). Resume always follows
//!    the persisted target, never a guess.
//!
//! Crash anywhere between steps 2 and 6 ⇒ the fence holds and a rerun is
//! idempotent: re-migrating an already-target-shaped shard streams its entries
//! and rebuilds to the IDENTICAL root (a no-op commit).

mod rebuild;
pub mod report;

use std::path::PathBuf;
use std::time::Instant;

use crate::tree::TreePolicy;

use super::config::{
    DEFAULT_INTERNAL_TARGET_BYTES, DEFAULT_LEAF_TARGET_BYTES, FenceView, MigrationState,
    finalize_v1, finalize_v2, flip_to_abort, install_forward_fence, read_state,
};
use super::lock::DataDirLock;
use super::{DatabaseConfig, DatabaseError};

pub use report::{MigrationDirection, MigrationOutcome, MigrationReport};

use rebuild::{BranchRebuild, ShardRebuild};

#[cfg(test)]
mod test_support;

#[cfg(test)]
#[path = "migrate_tests.rs"]
mod migrate_tests;

#[cfg(test)]
#[path = "crash_tests.rs"]
mod crash_tests;

/// What `migrate_chunking` should do with the named directory.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MigrateMode {
    /// Migrate FORWARD to v2 (or resume an in-flight forward run; or, on a dir
    /// already mid-abort, continue that abort — resume follows the persisted
    /// target). A no-op on a directory already fully v2.
    Migrate,
    /// Run the durable reverse to v1: abort an in-flight forward migration, or
    /// downgrade a completed v2 directory. A no-op on a directory already v1.
    Abort,
}

/// Options for one `migrate_chunking` run. No config-file surface and no silent
/// defaults: the target directory and the direction are explicit per call.
#[derive(Debug, Clone)]
pub struct MigrateOptions {
    pub data_dir: PathBuf,
    pub mode: MigrateMode,
}

impl MigrateOptions {
    /// Forward migration of the named directory.
    #[must_use]
    pub const fn migrate(data_dir: PathBuf) -> Self {
        Self {
            data_dir,
            mode: MigrateMode::Migrate,
        }
    }

    /// Reverse (abort/downgrade) run on the named directory.
    #[must_use]
    pub const fn abort(data_dir: PathBuf) -> Self {
        Self {
            data_dir,
            mode: MigrateMode::Abort,
        }
    }
}

/// Run the migration transaction (§4.2).
///
/// Acquires the A4 writer lock for the whole run and constructs NO `Database`
/// (the offline discipline the vacuum established): a running database over the
/// same shards would corrupt the WALs.
///
/// Idempotent and crash-resumable: a directory left fenced by an earlier
/// crashed run resumes in its persisted direction and converges to the identical
/// roots; a directory already in the requested target format is a no-op.
pub fn migrate_chunking(options: &MigrateOptions) -> Result<MigrationReport, DatabaseError> {
    // `crash_after` is `None` in production, so `run` never stops early; the
    // `None` arm is an unreachable internal invariant surfaced as a typed error
    // rather than a panic (the crate forbids `expect`/`panic`).
    run(options, None)?.map_or_else(
        || {
            Err(DatabaseError::MigrationFailed(
                "internal invariant: the migration stopped early with no crash injection"
                    .to_owned(),
            ))
        },
        Ok,
    )
}

/// The two byte targets a forward/abort run stamps: the directory's existing v2
/// stamp when one is present (resume), else the ratified defaults (§2.3).
fn v2_targets(state: &MigrationState) -> (u64, u64) {
    state
        .v2_targets
        .unwrap_or((DEFAULT_LEAF_TARGET_BYTES, DEFAULT_INTERNAL_TARGET_BYTES))
}

/// Resolve what to DO from the on-disk state and the requested mode. The fence's
/// persisted target is authoritative for a resume; `mode` decides only when
/// there is no fence, or requests a flip of an in-flight forward run.
enum Plan {
    /// Already in the requested target format — nothing to do.
    NoOp(MigrationDirection),
    /// Run forward to v2 (start if `install_fence`, else resume).
    Forward { install_fence: bool },
    /// Run the reverse to v1 (flip the direction first if `flip`, else resume).
    Abort { flip: bool },
}

fn plan(mode: MigrateMode, state: &MigrationState) -> Plan {
    match (&state.fence, mode) {
        // Mid-transaction: resume ALWAYS follows the persisted target...
        (
            Some(FenceView {
                target_is_v2: true, ..
            }),
            MigrateMode::Migrate,
        ) => Plan::Forward {
            install_fence: false,
        },
        // ...except an explicit abort of an in-flight FORWARD run flips it.
        (
            Some(FenceView {
                target_is_v2: true, ..
            }),
            MigrateMode::Abort,
        ) => Plan::Abort { flip: true },
        // A persisted ABORT target is followed by both a resume and a plain migrate.
        (
            Some(FenceView {
                target_is_v2: false,
                ..
            }),
            _,
        ) => Plan::Abort { flip: false },
        // Unfenced: the format stamp decides whether there is anything to do.
        (None, MigrateMode::Migrate) => {
            if is_v2(state) {
                Plan::NoOp(MigrationDirection::Forward)
            } else {
                Plan::Forward {
                    install_fence: true,
                }
            }
        }
        (None, MigrateMode::Abort) => {
            if is_v2(state) {
                Plan::Abort { flip: true }
            } else {
                Plan::NoOp(MigrationDirection::Abort)
            }
        }
    }
}

/// A directory is v2 when its stamp is `format_version = 2` with a policy block.
fn is_v2(state: &MigrationState) -> bool {
    state.format_version == Some(super::config::ON_DISK_FORMAT_VERSION)
        && state.v2_targets.is_some()
}

/// `crash_after` (test-only) names the step boundary AFTER which the run stops,
/// leaving the on-disk state exactly as a real crash there would — the lock then
/// drops as the process would on death. `None` in production: the run always
/// completes.
fn run(
    options: &MigrateOptions,
    crash_after: Option<&str>,
) -> Result<Option<MigrationReport>, DatabaseError> {
    let started = Instant::now();
    // Step 1: the A4 writer lock, held for the whole run. A live writer refuses
    // here (DataDirLocked); the offline contract is the vacuum's, verbatim.
    let _lock = DataDirLock::acquire(&options.data_dir)?;
    let state = read_state(&options.data_dir)?;

    match plan(options.mode, &state) {
        Plan::NoOp(direction) => Ok(Some(no_op_report(options, &state, direction, started))),
        Plan::Forward { install_fence } => {
            run_forward(options, &state, install_fence, crash_after, started)
        }
        Plan::Abort { flip } => run_abort(options, &state, flip, crash_after, started),
    }
}

/// A completed-run report for a directory already in the requested format.
fn no_op_report(
    options: &MigrateOptions,
    state: &MigrationState,
    direction: MigrationDirection,
    started: Instant,
) -> MigrationReport {
    base_report(
        options,
        state,
        direction,
        MigrationOutcome::AlreadyInTargetFormat,
        &Tally::default(),
        started,
    )
}

/// Running tallies collected across the shard/branch loops.
#[derive(Default)]
struct Tally {
    shards_rebuilt: usize,
    shards_unchanged: usize,
    shards_empty: usize,
    branch_heads_rebuilt: usize,
    branch_heads_unchanged: usize,
}

fn base_report(
    options: &MigrateOptions,
    state: &MigrationState,
    direction: MigrationDirection,
    outcome: MigrationOutcome,
    tally: &Tally,
    started: Instant,
) -> MigrationReport {
    let (source_policy, target_policy) = match direction {
        MigrationDirection::Forward => ("count-target-v1", "bytes-v2"),
        MigrationDirection::Abort => ("bytes-v2", "count-target-v1"),
    };
    MigrationReport {
        data_dir: options.data_dir.clone(),
        outcome,
        direction,
        source_policy: source_policy.to_owned(),
        target_policy: target_policy.to_owned(),
        shard_count: state.config.shard_count,
        shards_rebuilt: tally.shards_rebuilt,
        shards_unchanged: tally.shards_unchanged,
        shards_empty: tally.shards_empty,
        branch_heads_rebuilt: tally.branch_heads_rebuilt,
        branch_heads_unchanged: tally.branch_heads_unchanged,
        duration_ms: started.elapsed().as_millis(),
        vacuum_before_migrate: report::VACUUM_BEFORE_MIGRATE,
        manifest_bracket: report::MANIFEST_BRACKET,
    }
}

/// Rebuild every shard then every branch head under `policy`, tallying and
/// honouring the crash oracle. Returns `Ok(None)` if a crash boundary was hit
/// (the caller returns early, leaving the directory fenced).
fn rebuild_all(
    config: &DatabaseConfig,
    policy: TreePolicy,
    crash_after: Option<&str>,
    tally: &mut Tally,
    shard_label: impl Fn(usize) -> String,
    all_shards_label: &str,
) -> Result<Option<()>, DatabaseError> {
    let data_dir = &config.data_dir;
    for shard_id in 0..config.shard_count {
        match rebuild::rebuild_shard(data_dir, shard_id, policy)? {
            ShardRebuild::Rebuilt => tally.shards_rebuilt += 1,
            ShardRebuild::Unchanged => tally.shards_unchanged += 1,
            ShardRebuild::Empty => tally.shards_empty += 1,
        }
        if crash_after == Some(shard_label(shard_id).as_str()) {
            return Ok(None);
        }
    }
    if crash_after == Some(all_shards_label) {
        return Ok(None);
    }
    for (index, name) in rebuild::branch_names(data_dir)?.into_iter().enumerate() {
        match rebuild::rebuild_one_branch(data_dir, &name, policy)? {
            BranchRebuild::Rebuilt => tally.branch_heads_rebuilt += 1,
            BranchRebuild::Unchanged => tally.branch_heads_unchanged += 1,
        }
        if crash_after == Some(format!("after_branch_head_{index}").as_str()) {
            return Ok(None);
        }
    }
    Ok(Some(()))
}

/// Steps 2–3–4–6: the forward run.
fn run_forward(
    options: &MigrateOptions,
    state: &MigrationState,
    install_fence: bool,
    crash_after: Option<&str>,
    started: Instant,
) -> Result<Option<MigrationReport>, DatabaseError> {
    let config = &state.config;
    let (leaf, internal) = v2_targets(state);
    // Step 2: STAMP FIRST into the fenced state (skip when resuming — the fence
    // is already up, and re-installing the identical envelope is a needless
    // rewrite).
    if install_fence {
        install_forward_fence(config, leaf, internal)?;
        if crash_after == Some("after_forward_fence") {
            return Ok(None);
        }
    }
    let policy = TreePolicy::v2(leaf, internal);
    let mut tally = Tally::default();
    if rebuild_all(
        config,
        policy,
        crash_after,
        &mut tally,
        |shard_id| format!("after_shard_{shard_id}"),
        "after_all_shards",
    )?
    .is_none()
    {
        return Ok(None);
    }
    if crash_after == Some("before_fence_removal") {
        return Ok(None);
    }
    // Step 6: remove the fence LAST.
    finalize_v2(config, leaf, internal)?;
    Ok(Some(base_report(
        options,
        state,
        MigrationDirection::Forward,
        MigrationOutcome::MigratedToV2,
        &tally,
        started,
    )))
}

/// Step 7: the durable reverse run. `flip` installs/rewrites the fence to target
/// v1 (format STAYS 2) before any reverse root moves; on a resume it is already
/// v1-targeted. The final v1 envelope is installed ONLY after every root and
/// head is v1-shaped.
fn run_abort(
    options: &MigrateOptions,
    state: &MigrationState,
    flip: bool,
    crash_after: Option<&str>,
    started: Instant,
) -> Result<Option<MigrationReport>, DatabaseError> {
    let config = &state.config;
    let (leaf, internal) = v2_targets(state);
    if flip {
        flip_to_abort(config, leaf, internal)?;
        if crash_after == Some("after_abort_flip") {
            return Ok(None);
        }
    }
    let mut tally = Tally::default();
    if rebuild_all(
        config,
        TreePolicy::V1_DEFAULT,
        crash_after,
        &mut tally,
        |shard_id| format!("after_abort_shard_{shard_id}"),
        "after_all_abort_shards",
    )?
    .is_none()
    {
        return Ok(None);
    }
    if crash_after == Some("before_v1_finalize") {
        return Ok(None);
    }
    // The exact stable v1 envelope, installed last — format flips 2 → 1 only now
    // that no v2 root exists (§4.2 step 7).
    finalize_v1(config)?;
    Ok(Some(base_report(
        options,
        state,
        MigrationDirection::Abort,
        MigrationOutcome::AbortedToV1,
        &tally,
        started,
    )))
}

/// Test-only entry point that drives the run with a crash injected AFTER the
/// named step boundary, returning `Ok(None)` when the crash fired. Private:
/// reachable only from the child test modules.
#[cfg(test)]
fn migrate_with_crash(
    options: &MigrateOptions,
    crash_after: &str,
) -> Result<Option<MigrationReport>, DatabaseError> {
    run(options, Some(crash_after))
}