haematite 0.7.0

use std::fmt;
use std::io;
use std::path::{Path, PathBuf};

/// Errors surfaced by the top-level database handle.
#[derive(Debug)]
pub enum DatabaseError {
    DirectoryCreate(io::Error),
    ConfigWrite(io::Error),
    ConfigRead(io::Error),
    ConfigParse(String),
    InvalidShardCount,
    ShardSpawn(String),
    SweepSpawn(String),
    ShardError(String),
    SweepError(String),
    SyncSchedulerSpawn(String),
    SyncSchedulerError(String),
    IoError(io::Error),
    /// A retired TTL cadence field was present in stored configuration with
    /// the historically invalid zero value.
    InvalidSweepInterval,
    MissingSyncTopology,
    InvalidSyncInterval,
    /// The data directory's `config.json` carries an on-disk `format_version`
    /// stamp NEWER than this binary's [`crate::db::ON_DISK_FORMAT_VERSION`]
    /// (A5 downgrade refusal): the directory was written by a newer haematite
    /// whose layout this binary cannot know, so it refuses loudly instead of
    /// misreading data. Carries both versions so the operator can see exactly
    /// which side is behind. Distinct from [`Self::ConfigParse`]: the file is
    /// well-formed, it is the binary that is too old.
    FormatVersionTooNew {
        found: u32,
        supported: u32,
    },
    SequenceConflict {
        expected: u64,
        actual: u64,
    },
    CasMismatch {
        expected: Option<u64>,
        actual: Option<u64>,
    },
    ConsistencyError(String),
    /// A live distribution-endpoint operation failed (no endpoint attached, a
    /// transport send/connect failure, or a disconnected inbound drain).
    Distribution(String),
    /// A replicated write reached peer-quorum but the proposer could not durably
    /// apply its OWN committed value locally (see [`crate::db::Database::replicate_write`]).
    ///
    /// This is reported, never swallowed: a committed write that is absent on its
    /// own writer is a correctness hazard (it reopens the heal-mid-write
    /// split-brain hole). Under single-owner-per-key (the step-3 epoch fence) the
    /// local CAS can never mismatch, so this only ever surfaces a genuine local
    /// storage/IO fault.
    LocalCommitFailed(String),
    /// An [`crate::db::Database::acquire_shard`] election lost: a strictly higher
    /// ballot was promised elsewhere on every attempt. The candidate is NOT the
    /// owner and recorded no `owner_epoch`. Carries the highest competing counter
    /// seen so a caller could retry above it later. This is a clean, safe loss —
    /// the unique-ballot / majority invariants were never relaxed.
    ElectionLost {
        highest_seen: u64,
    },
    /// An [`crate::db::Database::acquire_shard`] election could not collect a
    /// majority of promises within the timeout on any attempt (e.g. a minority of
    /// nodes was reachable). The candidate is NOT the owner — never a false win.
    ElectionTimeout {
        attempts: u32,
    },
    /// A replicated CAS write was deterministically out-voted by the cluster: a
    /// stale/deposed owner's proposal collected enough rejects that a quorum of
    /// accepts is no longer reachable, so the writer is fenced and NOTHING was
    /// applied (the typed twin of [`crate::ConsistencyError::Fenced`]). Surfaced
    /// as its own variant — distinct from a generic [`Self::ConsistencyError`]
    /// string — so a consumer (e.g. an aion shard owner) can match the fence
    /// directly and re-resolve ownership rather than parsing a Display message.
    Fenced {
        required: usize,
        possible_accepts: usize,
    },
    /// A replicated CAS write was deterministically out-voted by *value-CAS
    /// mismatches alone* — the writer is still the live owner, but enough replicas
    /// refused the precondition that a quorum of accepts became unreachable (the
    /// typed twin of [`crate::ConsistencyError::CasConflict`]). Distinct from
    /// [`Self::Fenced`] (a higher-ballot owner deposed us, requiring ownership
    /// re-resolution): a `CasConflict` caller may simply re-read and re-CAS.
    CasConflict {
        required: usize,
        possible_accepts: usize,
    },
    /// The durable `cluster/members` record (CSOT-1, task #146) could not be
    /// encoded or a stored record could not be decoded/validated. Carries the
    /// underlying [`crate::sync::ClusterMembersError`] as a string so this variant
    /// stays dependency-light and matches the existing stringified-cause style.
    ClusterMembers(String),
    /// Another live writer holds the exclusive data-dir lock (A4): a second
    /// writer process (or a second `Database` handle in this process) already
    /// owns `<data_dir>/writer.lock`, and running two writers over the same
    /// shard WALs would corrupt them. The open fails immediately — it never
    /// blocks and never touches shard state. Advisory locks self-release on
    /// process death, so this is always a LIVE writer, never a stale lock.
    /// For observation alongside a live writer use
    /// [`crate::db::ReadOnlyDatabase`], which takes no lock.
    DataDirLocked {
        lock_path: PathBuf,
    },
    /// The data-dir writer lockfile could not be opened/created or the lock
    /// syscall failed for an I/O reason distinct from contention (A4).
    LockFileIo {
        lock_path: PathBuf,
        error: io::Error,
    },
    /// `Database::create` was pointed at a directory that already holds a
    /// database (its `config.json` exists) with no live writer attached (A5):
    /// create refuses rather than clobbering the existing config. A directory
    /// written by a NEWER binary refuses as [`Self::FormatVersionTooNew`]
    /// before this check is reached. Use [`crate::db::Database::open`] on the
    /// existing directory, or point create at a fresh path.
    DataDirAlreadyInitialised {
        config_path: PathBuf,
    },
    /// A root-advance subscriber callback attempted an engine WRITE on the same
    /// [`crate::db::Database`] while the tell was being delivered (ROOT-ADVANCE-SEAM
    /// R3). The write-back WALL refused it: an inline write from a callback would
    /// recurse commit->tell->commit unbounded and would self-deadlock re-entering
    /// the shard's emission mutex. The [`Display`] carries the remedy. READS from a
    /// callback are permitted and never raise this. NOTHING was written.
    WriteDuringRootAdvanceEmission,
    /// [`crate::db::Database::scan_sequence_keys_for_shards`] was handed the same
    /// shard id more than once (COMMIT-COLLAPSE §7, §9.7 RULED). A repeated id
    /// previously scanned the shard twice and silently duplicated its output; it
    /// is now a typed refusal, which — with the existing in-range check — caps
    /// every executor batch at `shard_count` jobs by construction. Nothing was
    /// scanned. A plain (non-`#[non_exhaustive]`) variant on purpose: a wildcard
    /// arm that swallowed a future failure mode is the campaign's own
    /// silent-anything hazard, so exhaustive matching keeps every new failure a
    /// compile-time event (batched into the 0.5.0 release-noted break).
    DuplicateShardId {
        shard_id: usize,
    },
    /// The `executor_threads` sizing knob could not be honoured (COMMIT-COLLAPSE
    /// §6, §9.2): `Some(0)` is rejected loudly at validation, and an
    /// unavailable `available_parallelism()` refuses startup rather than falling
    /// back to a silent number (rule 2). Carries the remedy text.
    ExecutorThreadsInvalid(String),
    /// A fan-out batch was submitted to the bounded executor while it was
    /// shutting down (COMMIT-COLLAPSE §6): a blocked submitter is woken with this
    /// typed error as the pool drains. Nothing was dispatched.
    ExecutorShutdown,

    /// The data directory is mid chunking-migration (CHUNKING-POLICY.md §4.2):
    /// its `config.json` carries the structured migration fence, so a normal
    /// [`crate::db::Database::open`] / [`crate::db::ReadOnlyDatabase::open`]
    /// REFUSES it. Serving a half-migrated shard set as a coherent database would
    /// invite exactly the cross-shard inconsistency the fence exists to prevent.
    /// Only `migrate_chunking` proceeds past the fence; the [`Display`] names the
    /// resume command. Distinct from [`Self::FormatVersionTooNew`]: this binary
    /// DOES understand the format — the directory is deliberately fenced.
    MigrationInProgress {
        resume_command: String,
    },

    /// The v2 stamped chunking policy in `config.json` is unusable: a zero target
    /// (never issued — the create/migrate stamp is always nonzero), an unknown
    /// policy id, or a format-2 directory missing its policy block entirely
    /// (CHUNKING-POLICY.md §2.3, rule 2). Distinct from [`Self::ConfigParse`] —
    /// the file is well-formed; the engine refuses to GUESS a byte-aware target
    /// rather than silently defaulting one.
    InvalidChunkingStamp(String),

    /// A `migrate_chunking` transaction (CHUNKING-POLICY.md §4.2) failed
    /// mid-run: a shard/branch rebuild, WAL commit, store open, or record
    /// advance surfaced an underlying error. The directory is left fenced (the
    /// stamp-first fence went up before any root moved), so a normal open
    /// refuses [`Self::MigrationInProgress`] and a rerun resumes idempotently.
    /// Carries the failing step's context.
    MigrationFailed(String),
}

impl fmt::Display for DatabaseError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::DirectoryCreate(error) => {
                write!(formatter, "failed to create database directory: {error}")
            }
            Self::ConfigWrite(error) => {
                write!(formatter, "failed to write database config: {error}")
            }
            Self::ConfigRead(error) => write!(formatter, "failed to read database config: {error}"),
            Self::ConfigParse(message) => {
                write!(formatter, "failed to parse database config: {message}")
            }
            Self::InvalidShardCount => write!(formatter, "database shard_count must be at least 1"),
            Self::ShardSpawn(message) => {
                write!(formatter, "failed to spawn shard actor: {message}")
            }
            Self::SweepSpawn(message) => {
                write!(formatter, "failed to spawn sweep actor: {message}")
            }
            Self::ShardError(message) => write!(formatter, "shard operation failed: {message}"),
            Self::SweepError(message) => write!(formatter, "sweep operation failed: {message}"),
            Self::SyncSchedulerSpawn(message) => {
                write!(formatter, "failed to spawn sync scheduler: {message}")
            }
            Self::SyncSchedulerError(message) => {
                write!(formatter, "sync scheduler failed: {message}")
            }
            Self::IoError(error) => write!(formatter, "database I/O error: {error}"),
            Self::InvalidSweepInterval => {
                write!(formatter, "legacy TTL cadence must be greater than zero")
            }
            Self::MissingSyncTopology => {
                write!(formatter, "distributed database requires sync topology")
            }
            Self::InvalidSyncInterval => {
                write!(formatter, "sync_interval must be greater than zero")
            }
            Self::FormatVersionTooNew { found, supported } => {
                fmt_format_version_too_new(formatter, *found, *supported)
            }
            Self::SequenceConflict { expected, actual } => write!(
                formatter,
                "sequence conflict on append: expected {expected}, actual {actual}"
            ),
            Self::CasMismatch { expected, actual } => write!(
                formatter,
                "cas mismatch: expected {expected:?}, actual {actual:?}"
            ),
            Self::ConsistencyError(message) => {
                write!(formatter, "consistency requirement failed: {message}")
            }
            Self::Distribution(message) => {
                write!(formatter, "distribution endpoint error: {message}")
            }
            Self::LocalCommitFailed(message) => fmt_local_commit_failed(formatter, message),
            Self::ElectionLost { highest_seen } => fmt_election_lost(formatter, *highest_seen),
            Self::ElectionTimeout { attempts } => fmt_election_timeout(formatter, *attempts),
            Self::Fenced {
                required,
                possible_accepts,
            } => fmt_fenced(formatter, *required, *possible_accepts),
            Self::CasConflict {
                required,
                possible_accepts,
            } => fmt_cas_conflict(formatter, *required, *possible_accepts),
            Self::ClusterMembers(message) => {
                write!(formatter, "cluster/members record error: {message}")
            }
            Self::DataDirLocked { lock_path } => fmt_data_dir_locked(formatter, lock_path),
            Self::LockFileIo { lock_path, error } => write!(
                formatter,
                "failed to acquire the data-dir writer lock at {}: {error}",
                lock_path.display()
            ),
            Self::DataDirAlreadyInitialised { config_path } => {
                fmt_data_dir_already_initialised(formatter, config_path)
            }
            Self::WriteDuringRootAdvanceEmission => {
                formatter.write_str(super::root_advance::WRITE_DURING_EMISSION_REMEDY)
            }
            Self::MigrationInProgress { resume_command } => {
                fmt_migration_in_progress(formatter, resume_command)
            }
            Self::InvalidChunkingStamp(message) => {
                write!(
                    formatter,
                    "invalid v2 chunking stamp in database config: {message}"
                )
            }
            Self::DuplicateShardId { shard_id } => fmt_duplicate_shard_id(formatter, *shard_id),
            Self::ExecutorThreadsInvalid(message) => fmt_executor_threads(formatter, message),
            Self::ExecutorShutdown => write!(
                formatter,
                "the database executor is shutting down; the fan-out batch was not dispatched"
            ),
            Self::MigrationFailed(message) => fmt_migration_failed(formatter, message),
        }
    }
}

fn fmt_migration_failed(formatter: &mut fmt::Formatter<'_>, message: &str) -> fmt::Result {
    write!(
        formatter,
        "chunking migration failed (the directory is left fenced — rerun \
         `haem migrate-chunking` to resume): {message}"
    )
}

/// Message bodies split out of `Display::fmt` to keep that exhaustive match
/// within the function-length lint as the error surface grows.
fn fmt_local_commit_failed(formatter: &mut fmt::Formatter<'_>, message: &str) -> fmt::Result {
    write!(
        formatter,
        "replicated write reached quorum but local durable commit failed: {message}"
    )
}

fn fmt_election_lost(formatter: &mut fmt::Formatter<'_>, highest_seen: u64) -> fmt::Result {
    write!(
        formatter,
        "shard election lost: a higher ballot (counter {highest_seen}) was promised elsewhere"
    )
}

fn fmt_election_timeout(formatter: &mut fmt::Formatter<'_>, attempts: u32) -> fmt::Result {
    write!(
        formatter,
        "shard election timed out without a majority after {attempts} attempts"
    )
}

/// Message body for the deterministic CAS fence, split out of `Display::fmt` to
/// keep that exhaustive match within the function-length lint.
fn fmt_fenced(
    formatter: &mut fmt::Formatter<'_>,
    required: usize,
    possible_accepts: usize,
) -> fmt::Result {
    write!(
        formatter,
        "fenced by CAS rejects: required {required} accepts, only {possible_accepts} still possible"
    )
}

/// Message body for the value-CAS loss, split out of `Display::fmt`.
fn fmt_cas_conflict(
    formatter: &mut fmt::Formatter<'_>,
    required: usize,
    possible_accepts: usize,
) -> fmt::Result {
    write!(
        formatter,
        "lost CAS by value mismatch: required {required} accepts, only {possible_accepts} still possible"
    )
}

fn fmt_migration_in_progress(
    formatter: &mut fmt::Formatter<'_>,
    resume_command: &str,
) -> fmt::Result {
    write!(
        formatter,
        "database is mid chunking-migration (a migration fence is present in config.json); \
         normal open is refused — resume with `{resume_command}` or restore the pre-migration \
         backup the CLI told you to take"
    )
}

/// Message body for the duplicate-shard-id scan refusal (COMMIT-COLLAPSE §7),
/// split out of `Display::fmt`.
/// Message body for the executor-threads config refusal, split out of
/// `Display::fmt` to keep that exhaustive match within the function-length lint.
fn fmt_executor_threads(formatter: &mut fmt::Formatter<'_>, message: &str) -> fmt::Result {
    write!(
        formatter,
        "invalid executor_threads configuration: {message}"
    )
}

fn fmt_duplicate_shard_id(formatter: &mut fmt::Formatter<'_>, shard_id: usize) -> fmt::Result {
    write!(
        formatter,
        "scan_sequence_keys_for_shards was given shard id {shard_id} more than once; duplicate \
         shard ids are refused (they would scan a shard twice and duplicate its output) — pass \
         each shard id at most once"
    )
}

/// Long-form message bodies for the A4/A5 data-dir refusals, split out of
/// `Display::fmt` to keep that exhaustive match within the function-length
/// lint as the error surface grows.
fn fmt_format_version_too_new(
    formatter: &mut fmt::Formatter<'_>,
    found: u32,
    supported: u32,
) -> fmt::Result {
    write!(
        formatter,
        "database on-disk format version {found} is newer than the newest format version \
         this binary supports ({supported}); refusing to open — use a haematite build \
         that understands format version {found}"
    )
}

fn fmt_data_dir_locked(formatter: &mut fmt::Formatter<'_>, lock_path: &Path) -> fmt::Result {
    write!(
        formatter,
        "data dir is locked by another live writer (writer lock held at {}); \
         a second writer would corrupt the shard WALs — use ReadOnlyDatabase to observe",
        lock_path.display()
    )
}

fn fmt_data_dir_already_initialised(
    formatter: &mut fmt::Formatter<'_>,
    config_path: &Path,
) -> fmt::Result {
    write!(
        formatter,
        "data dir already holds a database ({} exists); refusing to clobber it — \
         open the existing database instead of creating over it",
        config_path.display()
    )
}

impl std::error::Error for DatabaseError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::DirectoryCreate(error)
            | Self::ConfigWrite(error)
            | Self::ConfigRead(error)
            | Self::IoError(error)
            | Self::LockFileIo { error, .. } => Some(error),
            Self::ConfigParse(_)
            | Self::InvalidShardCount
            | Self::ShardSpawn(_)
            | Self::SweepSpawn(_)
            | Self::ShardError(_)
            | Self::SweepError(_)
            | Self::SyncSchedulerSpawn(_)
            | Self::SyncSchedulerError(_)
            | Self::InvalidSweepInterval
            | Self::MissingSyncTopology
            | Self::InvalidSyncInterval
            | Self::FormatVersionTooNew { .. }
            | Self::SequenceConflict { .. }
            | Self::CasMismatch { .. }
            | Self::ConsistencyError(_)
            | Self::Distribution(_)
            | Self::LocalCommitFailed(_)
            | Self::ElectionLost { .. }
            | Self::ElectionTimeout { .. }
            | Self::Fenced { .. }
            | Self::CasConflict { .. }
            | Self::ClusterMembers(_)
            | Self::DataDirLocked { .. }
            | Self::DataDirAlreadyInitialised { .. }
            | Self::WriteDuringRootAdvanceEmission
            | Self::DuplicateShardId { .. }
            | Self::ExecutorThreadsInvalid(_)
            | Self::ExecutorShutdown
            | Self::MigrationInProgress { .. }
            | Self::InvalidChunkingStamp(_)
            | Self::MigrationFailed(_) => None,
        }
    }
}

impl From<crate::sync::ClusterMembersError> for DatabaseError {
    fn from(error: crate::sync::ClusterMembersError) -> Self {
        Self::ClusterMembers(error.to_string())
    }
}

impl From<io::Error> for DatabaseError {
    fn from(error: io::Error) -> Self {
        Self::IoError(error)
    }
}

impl From<crate::sync::ConsistencyError> for DatabaseError {
    /// Preserve the deterministic CAS fence as the typed [`Self::Fenced`] so
    /// consumers can match it; every other consistency failure keeps its existing
    /// stringified [`Self::ConsistencyError`] form (behaviour unchanged).
    fn from(error: crate::sync::ConsistencyError) -> Self {
        match error {
            crate::sync::ConsistencyError::Fenced {
                required,
                possible_accepts,
            } => Self::Fenced {
                required,
                possible_accepts,
            },
            crate::sync::ConsistencyError::CasConflict {
                required,
                possible_accepts,
            } => Self::CasConflict {
                required,
                possible_accepts,
            },
            other => Self::ConsistencyError(other.to_string()),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::DatabaseError;
    use crate::sync::ConsistencyError;

    #[test]
    fn consistency_fence_maps_to_typed_fenced() {
        let mapped = DatabaseError::from(ConsistencyError::Fenced {
            required: 3,
            possible_accepts: 1,
        });
        assert!(
            matches!(
                mapped,
                DatabaseError::Fenced {
                    required: 3,
                    possible_accepts: 1
                }
            ),
            "the deterministic CAS fence must survive as the typed DatabaseError::Fenced"
        );
    }

    #[test]
    fn consistency_cas_conflict_maps_to_typed_cas_conflict() {
        // The value-CAS loss must survive as the typed DatabaseError::CasConflict
        // (preserving required/possible_accepts), distinct from the typed Fenced and
        // from the stringified fallback.
        let mapped = DatabaseError::from(ConsistencyError::CasConflict {
            required: 3,
            possible_accepts: 1,
        });
        assert!(
            matches!(
                mapped,
                DatabaseError::CasConflict {
                    required: 3,
                    possible_accepts: 1
                }
            ),
            "the value-CAS loss must survive as the typed DatabaseError::CasConflict"
        );
    }

    #[test]
    fn other_consistency_failures_stay_stringified() {
        // A non-fence consistency failure must NOT be misclassified as a fence; it
        // keeps its existing stringified ConsistencyError form (behaviour unchanged).
        for error in [
            ConsistencyError::QuorumUnavailable {
                required: 2,
                possible: 1,
            },
            ConsistencyError::TransportUnavailable,
            ConsistencyError::AckFailed,
        ] {
            let display = error.to_string();
            let mapped = DatabaseError::from(error);
            assert!(
                matches!(mapped, DatabaseError::ConsistencyError(ref message) if *message == display),
                "non-fence consistency failures must remain DatabaseError::ConsistencyError"
            );
        }
    }
}