haematite 0.7.0

//! On-disk config envelopes for the chunking-v2 migration transaction
//! (CHUNKING-POLICY.md §4.2). A child module of [`super`] so the migration's
//! four config states reuse the SAME private [`ChunkingStamp`] /
//! [`MigrationFence`] schema and the SAME atomic install discipline
//! ([`super::install_config_atomic`]) as `create`/`open`, never a second copy
//! of the layout.
//!
//! The four states, in transaction order:
//!
//! 1. **Fenced v2** ([`install_forward_fence`]) — step 2's STAMP-FIRST write:
//!    `format_version = 2` + the v2 policy block + the forward migration fence.
//!    A crash from here on reopens fenced (normal opens refuse
//!    `MigrationInProgress`; an old binary sees `format_version = 2` and refuses
//!    `FormatVersionTooNew`).
//! 2. **Flipped fence** ([`flip_to_abort`]) — step 7's durable direction flip:
//!    the fence's target becomes v1 (phase reversed) BEFORE any reverse root
//!    moves; `format_version` STAYS 2 and the v2 policy block STAYS, so an old
//!    binary keeps refusing while any v2 root still exists.
//! 3. **Stable v2** ([`finalize_v2`]) — step 6's fence removal, LAST: the exact
//!    envelope `create` writes (`format_version = 2` + policy block, no fence).
//! 4. **Stable v1** ([`finalize_v1`]) — the abort's final publication: the exact
//!    v1 envelope (`format_version = 1`, no policy block, no fence), installed
//!    only once every root and head is v1-shaped again.
//!
//! Reads go through [`read_state`], which — unlike [`super::read_config`] —
//! does NOT refuse a fenced directory: `migrate_chunking` is the one caller
//! licensed to interpret the fence and resume past it.

use std::fs;
use std::path::Path;

use super::{ChunkingStamp, DatabaseConfig, DatabaseError, MigrationFence, StoredConfig};

/// The stamped policy id for count-target (v1) chunking — the fence's source
/// label on a forward run and its target label on an abort. Distinct from
/// [`super::CHUNKING_V2_ID`] (`"bytes-v2"`), which labels the v2 side.
const POLICY_V1_LABEL: &str = "count-target-v1";

/// Fence phase strings. `phase` is advisory context for the operator/log; the
/// binding direction is carried by `target_policy` (resume ALWAYS follows the
/// persisted target, never `phase` — §4.2 step 7).
const PHASE_FORWARD: &str = "forward";
const PHASE_ABORT: &str = "abort";

/// The migration-relevant on-disk state, read WITHOUT the fence refusal that
/// [`super::read_config`] applies (only `migrate_chunking` may read past a
/// fence). All fields mirror the on-disk envelope verbatim.
pub(in crate::db) struct MigrationState {
    pub config: DatabaseConfig,
    pub format_version: Option<u32>,
    /// The stamped `(leaf_target_bytes, internal_target_bytes)` when a v2 policy
    /// block is present, else `None`.
    pub v2_targets: Option<(u64, u64)>,
    pub fence: Option<FenceView>,
}

/// A read-back view of the on-disk [`MigrationFence`]. The resume logic makes
/// exactly ONE decision from it — follow the PERSISTED target direction, never a
/// guess (§4.2 step 7) — so only that bit is surfaced.
pub(in crate::db) struct FenceView {
    /// True iff `target_policy == "bytes-v2"` (resume forward); false for the
    /// abort target `"count-target-v1"` (resume the reverse run).
    pub target_is_v2: bool,
}

/// Read `config.json` for migration: parse every migration-relevant field and
/// NEVER refuse on the fence (that refusal is [`super::read_config`]'s job for
/// normal opens; `migrate_chunking` is the sanctioned reader past the fence).
pub(in crate::db) fn read_state(data_dir: &Path) -> Result<MigrationState, DatabaseError> {
    let bytes =
        fs::read(data_dir.join(crate::db::CONFIG_FILE)).map_err(DatabaseError::ConfigRead)?;
    let stored: StoredConfig = serde_json::from_slice(&bytes)
        .map_err(|error| DatabaseError::ConfigParse(error.to_string()))?;
    let v2_targets = stored
        .chunking_policy
        .as_ref()
        .map(|stamp| (stamp.leaf_target_bytes, stamp.internal_target_bytes));
    let fence = stored.migration_fence.as_ref().map(|fence| FenceView {
        target_is_v2: fence.target_policy == super::CHUNKING_V2_ID,
    });
    let mut config = stored.config;
    // Match `Database::open`: the authoritative data_dir is the path the caller
    // named, never a possibly-stale copy inside the file — so a rewrite is
    // deterministic regardless of where the directory was moved from.
    config.data_dir = data_dir.to_path_buf();
    Ok(MigrationState {
        config,
        format_version: stored.format_version,
        v2_targets,
        fence,
    })
}

/// Write-side envelope carrying an OPTIONAL fence and an OPTIONAL v2 policy
/// block, flattened over the caller's [`DatabaseConfig`] — the read-side twin of
/// [`StoredConfig`]. `skip_serializing_if` keeps a stable-v1 write byte-clean
/// (no `chunking_policy`/`migration_fence` keys at all), matching a real v1
/// directory exactly.
#[derive(serde::Serialize)]
struct MigrationEnvelope<'config> {
    format_version: u32,
    #[serde(skip_serializing_if = "Option::is_none")]
    chunking_policy: Option<ChunkingStamp>,
    #[serde(skip_serializing_if = "Option::is_none")]
    migration_fence: Option<MigrationFence>,
    #[serde(flatten)]
    config: &'config DatabaseConfig,
}

fn v2_stamp(leaf_target_bytes: u64, internal_target_bytes: u64) -> ChunkingStamp {
    ChunkingStamp {
        chunking: super::CHUNKING_V2_ID.to_owned(),
        leaf_target_bytes,
        internal_target_bytes,
    }
}

fn write_envelope(
    config: &DatabaseConfig,
    envelope: &MigrationEnvelope<'_>,
) -> Result<(), DatabaseError> {
    let bytes = serde_json::to_vec_pretty(envelope).map_err(|error| {
        DatabaseError::ConfigWrite(std::io::Error::new(std::io::ErrorKind::InvalidData, error))
    })?;
    super::install_config_atomic(&config.data_dir.join(crate::db::CONFIG_FILE), &bytes)
}

/// Step 2 (§4.2): STAMP FIRST into a fenced migrating state — `format_version =
/// 2`, the v2 policy block, and the FORWARD fence, in ONE atomic config rewrite.
/// The fence goes up BEFORE any shard root moves (the r1 B2 fold: stamp-LAST
/// durably published v2 roots under a v1 stamp).
pub(in crate::db) fn install_forward_fence(
    config: &DatabaseConfig,
    leaf_target_bytes: u64,
    internal_target_bytes: u64,
) -> Result<(), DatabaseError> {
    write_envelope(
        config,
        &MigrationEnvelope {
            format_version: super::ON_DISK_FORMAT_VERSION,
            chunking_policy: Some(v2_stamp(leaf_target_bytes, internal_target_bytes)),
            migration_fence: Some(MigrationFence {
                source_policy: POLICY_V1_LABEL.to_owned(),
                target_policy: super::CHUNKING_V2_ID.to_owned(),
                phase: PHASE_FORWARD.to_owned(),
            }),
            config,
        },
    )
}

/// Step 7 (§4.2): the durable DIRECTION FLIP — rewrite the fence to target v1
/// (phase reversed) BEFORE any reverse root moves. `format_version` STAYS 2 and
/// the v2 policy block STAYS present: writing 1 early would let a v1 binary in
/// (its validator accepts `Some(1)`) while v2 roots still exist. Resume after a
/// crash reads the persisted v1 target and continues the reverse run.
pub(in crate::db) fn flip_to_abort(
    config: &DatabaseConfig,
    leaf_target_bytes: u64,
    internal_target_bytes: u64,
) -> Result<(), DatabaseError> {
    write_envelope(
        config,
        &MigrationEnvelope {
            format_version: super::ON_DISK_FORMAT_VERSION,
            chunking_policy: Some(v2_stamp(leaf_target_bytes, internal_target_bytes)),
            migration_fence: Some(MigrationFence {
                source_policy: super::CHUNKING_V2_ID.to_owned(),
                target_policy: POLICY_V1_LABEL.to_owned(),
                phase: PHASE_ABORT.to_owned(),
            }),
            config,
        },
    )
}

/// Step 6 (§4.2): remove the fence LAST — the stable v2 envelope
/// (`format_version = 2` + policy block, NO fence), byte-identical to what
/// `create` writes for a v2-native database. After this the directory opens
/// normally as v2.
pub(in crate::db) fn finalize_v2(
    config: &DatabaseConfig,
    leaf_target_bytes: u64,
    internal_target_bytes: u64,
) -> Result<(), DatabaseError> {
    write_envelope(
        config,
        &MigrationEnvelope {
            format_version: super::ON_DISK_FORMAT_VERSION,
            chunking_policy: Some(v2_stamp(leaf_target_bytes, internal_target_bytes)),
            migration_fence: None,
            config,
        },
    )
}

/// The abort's final publication (§4.2 step 7): the exact stable v1 envelope —
/// `format_version = 1`, no policy block, no fence — installed ONLY after every
/// shard root and branch head is v1-shaped again. A v1 binary may now open it.
pub(in crate::db) fn finalize_v1(config: &DatabaseConfig) -> Result<(), DatabaseError> {
    write_envelope(
        config,
        &MigrationEnvelope {
            format_version: super::V1_FORMAT_VERSION,
            chunking_policy: None,
            migration_fence: None,
            config,
        },
    )
}