haematite 0.6.2

//! Database startup and lazy shard-router construction.
//!
//! LAZY SHARD MATERIALISATION: startup no longer pre-creates all `shard_count`
//! directories or spawn one actor per shard. It builds a [`ShardRouter`] over
//! the fixed `shard_count` modulus base and lets the router spawn each shard on
//! first touch. Boot cost is
//! therefore O(shards actually used), not `O(shard_count)` — a very high
//! `shard_count` is ~free until its shards are exercised.

use std::fs;
use std::sync::Arc;
use std::time::Duration;

use beamr::module::ModuleRegistry;
use beamr::scheduler::{Scheduler, SchedulerConfig};

use crate::shard::router::{ShardMode, ShardRouter};
use crate::sync::scheduler::{
    NoopSyncPullTrigger, SyncSchedulerConfig, SyncSchedulerError, SyncSchedulerHandle,
};

use super::config::{ensure_uninitialised, validate_database_config, write_config};
use super::lock::DataDirLock;
use super::{Database, DatabaseConfig, DatabaseError};

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);

/// Everything a started database owns EXCEPT the writer lock. Splitting the
/// start from the assembly lets the create path keep the lock alive across
/// its failure cleanup: removing a partial directory is only safe while the
/// exclusive lock still excludes every other writer.
struct DatabaseParts {
    config: DatabaseConfig,
    scheduler: Arc<Scheduler>,
    router: ShardRouter,
    sync_schedulers: Vec<SyncSchedulerHandle>,
}

fn assemble(parts: DatabaseParts, lock: DataDirLock) -> Database {
    Database {
        config: parts.config,
        scheduler: parts.scheduler,
        router: parts.router,
        sync_schedulers: parts.sync_schedulers,
        distribution: None,
        owner_stamps: super::owner_stamp::OwnerStamps::default(),
        lock,
        timeout: DEFAULT_TIMEOUT,
    }
}

#[derive(Clone, Copy, Debug)]
pub(super) enum StartupMode {
    Create,
    Open,
}

impl StartupMode {
    const fn shard_mode(self) -> ShardMode {
        match self {
            Self::Create => ShardMode::Create,
            Self::Open => ShardMode::Open,
        }
    }
}

/// D0 anchor remedy (canonical text in [`crate::fence::D0_ANCHOR_REMEDY`]); a
/// missing `data_dir` parent refuses with this exact remedy (FENCE-CHAIN r8
/// D0/D1, R4e).
pub(super) const D0_ANCHOR_REMEDY: &str = crate::fence::D0_ANCHOR_REMEDY;

pub(super) fn initialise_database(config: DatabaseConfig) -> Result<Database, DatabaseError> {
    // Only the database root directory is created up front; per-shard directories
    // are created lazily by the router on each shard's first touch. This is what
    // makes create() O(1) in `shard_count` instead of O(shard_count).
    // D1: a SINGLE level below a pre-existing parent; a missing parent is a typed
    // refusal carrying the D0 remedy, never a deep `create_dir_all`.
    let created_dir = !config.data_dir.exists();
    create_data_dir(&config.data_dir)?;
    // D2: the data_dir entry is fenced UNCONDITIONALLY (not only when this call
    // created it), so a create that retries after a fence failure re-fences the
    // entry rather than skipping it because the directory now exists (R4d causal
    // retry). The fence runs before the writer lock is taken; its failure returns
    // before any durable claim and leaves any residue safe for the next attempt.
    sync_new_dir_entry(&config.data_dir)?;
    // A4: the exclusive writer lock is taken BEFORE `config.json` is written, so
    // a create that loses the lock race fails loudly without clobbering the live
    // winner's config. A lock-acquire failure deliberately skips the cleanup
    // below even when this call created the directory: losing the race means a
    // concurrent winner already owns the dir it raced us into.
    // Test seam (r5 F4): deterministically reproduce the interleaving where a
    // competitor B fully initialised this directory AFTER this call created it but
    // BEFORE this call reaches `ensure_uninitialised`. A foreign completed config
    // is installed here so the refusal-transfers-ownership guard below is exercised
    // with THIS call's `created_dir` bit set.
    #[cfg(test)]
    if take_inject_foreign_config() {
        inject_foreign_completed_config(&config.data_dir)?;
    }

    let lock = DataDirLock::acquire(&config.data_dir)?;
    let data_dir = config.data_dir.clone();

    // D2 ownership-transfer (r5 F4): a refusal at `ensure_uninitialised` — this
    // call observing ANOTHER creator's completed `config.json` — irrevocably
    // transfers ownership away from this call, so NO cleanup runs regardless of
    // who originally created the directory. The interleaving this closes: A
    // creates the dir; B takes the lock first, fully initialises, closes; A takes
    // the lock later and is refused here — A's created-this-call bit must NOT
    // delete B's completed database. Cleanup below is reachable ONLY once this
    // call has PASSED `ensure_uninitialised` and owns the initialisation attempt.
    if let Err(error) = ensure_uninitialised(&config.data_dir) {
        drop(lock);
        return Err(error);
    }

    match install_and_start(config) {
        Ok(parts) => Ok(assemble(parts, lock)),
        Err(error) => {
            // Cleanup runs while the exclusive lock is STILL HELD, so a
            // concurrent create/open cannot acquire the directory between this
            // failure and its removal — and only a directory this very call
            // created is removed, never a pre-existing one. The lock is
            // released only after the removal completes.
            if created_dir {
                drop(fs::remove_dir_all(&data_dir));
            }
            drop(lock);
            Err(error)
        }
    }
}

/// The owning portion of create that runs under the writer lock AFTER
/// `ensure_uninitialised` has confirmed this call owns the initialisation
/// attempt: install the stamped config, start the shards. A failure here is
/// cleanup-eligible (this call owns the directory).
fn install_and_start(config: DatabaseConfig) -> Result<DatabaseParts, DatabaseError> {
    write_config(&config)?;
    start_database(config, StartupMode::Create)
}

/// A4: open path — acquire the exclusive writer lock, then start the shards.
///
/// FENCE-CHAIN r8 R1d (F3) / R4g: the open path fences the EXISTING `data_dir`
/// entry WHILE HOLDING the lock, then starts actors. A fence failure returns
/// [`DatabaseError::DirectoryCreate`] and NEVER removes the existing directory
/// (the open path created nothing to remove; the dropped lock releases cleanly).
/// One `data_dir`-entry fsync per `Database::open`, part of the signed per-open
/// bound.
pub(super) fn open_database(config: DatabaseConfig) -> Result<Database, DatabaseError> {
    let lock = DataDirLock::acquire(&config.data_dir)?;
    sync_new_dir_entry(&config.data_dir)?;
    start_database(config, StartupMode::Open).map(|parts| assemble(parts, lock))
}

/// Create `data_dir` a single level below its pre-existing parent (D1), accepting
/// an existing directory (D2) and a concurrent creator's win. A missing parent is
/// a typed [`DatabaseError::DirectoryCreate`] refusal preserving `NotFound` and
/// carrying the D0 remedy. In test builds the creation is journalled so a CUT can
/// rewind an unfenced `data_dir` entry (the L5 crash pin).
fn create_data_dir(data_dir: &std::path::Path) -> Result<(), DatabaseError> {
    match fs::metadata(data_dir) {
        Ok(metadata) if metadata.is_dir() => Ok(()),
        Ok(_metadata) => Err(DatabaseError::DirectoryCreate(std::io::Error::other(
            format!(
                "data directory path is not a directory: {}",
                data_dir.display()
            ),
        ))),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            #[cfg(test)]
            let reservation = crate::fence::journal::reserve_create_dir(data_dir);
            match fs::create_dir(data_dir) {
                Ok(()) => {
                    #[cfg(test)]
                    reservation.commit();
                    Ok(())
                }
                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {
                    #[cfg(test)]
                    reservation.cancel();
                    Ok(())
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                    #[cfg(test)]
                    reservation.cancel();
                    Err(DatabaseError::DirectoryCreate(std::io::Error::new(
                        std::io::ErrorKind::NotFound,
                        format!(
                            "data directory parent does not exist: {} — {D0_ANCHOR_REMEDY}",
                            data_dir.display()
                        ),
                    )))
                }
                Err(error) => {
                    #[cfg(test)]
                    reservation.cancel();
                    Err(DatabaseError::DirectoryCreate(error))
                }
            }
        }
        Err(error) => Err(DatabaseError::DirectoryCreate(error)),
    }
}

fn start_database(
    config: DatabaseConfig,
    mode: StartupMode,
) -> Result<DatabaseParts, DatabaseError> {
    validate_database_config(&config)?;
    let scheduler = create_scheduler()?;
    let router = build_router(&scheduler, &config, mode)?;
    let sync_schedulers = match spawn_sync_schedulers(&scheduler, &config, &router) {
        Ok(handles) => handles,
        Err(error) => {
            router.shutdown_all(DEFAULT_TIMEOUT);
            return Err(error);
        }
    };
    Ok(DatabaseParts {
        config,
        scheduler,
        router,
        sync_schedulers,
    })
}

/// Fence the directory entry for `data_dir` itself: `config.json` and its content
/// are fsynced, but their dirents live inside `data_dir`, whose OWN entry in its
/// parent is what makes the database reachable after power loss. The entry fence
/// is delegated to the one fence-chain primitive ([`crate::fence::sync_dir_entry`]),
/// which carries the Unix-only platform gap and hosts the crash battery's hooks;
/// only the immediate parent is synced — pre-existing ancestors are the operator's
/// D0 precondition.
fn sync_new_dir_entry(data_dir: &std::path::Path) -> Result<(), DatabaseError> {
    let Some(parent) = data_dir.parent() else {
        return Ok(());
    };
    crate::fence::sync_dir_entry(parent).map_err(DatabaseError::DirectoryCreate)
}

// Test seam for the r5 F4 ownership-transfer pin: arm the NEXT create so a
// foreign competitor's completed config appears after this call creates the dir
// but before it reaches `ensure_uninitialised`.
#[cfg(test)]
thread_local! {
    static INJECT_FOREIGN_CONFIG: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}

#[cfg(test)]
pub(super) fn inject_foreign_config_next_create() {
    INJECT_FOREIGN_CONFIG.with(|flag| flag.set(true));
}

#[cfg(test)]
fn take_inject_foreign_config() -> bool {
    INJECT_FOREIGN_CONFIG.with(std::cell::Cell::take)
}

/// The competitor's completed config: a DISTINCT `shard_count` so the pin can
/// prove the winner's database survived byte-identically (never deleted, never
/// rewritten by the refused creator).
#[cfg(test)]
pub(super) const FOREIGN_SHARD_COUNT: usize = 9;

#[cfg(test)]
fn inject_foreign_completed_config(data_dir: &std::path::Path) -> Result<(), DatabaseError> {
    write_config(&DatabaseConfig {
        data_dir: data_dir.to_path_buf(),
        shard_count: FOREIGN_SHARD_COUNT,
        distributed: None,
    })
}

fn create_scheduler() -> Result<Arc<Scheduler>, DatabaseError> {
    Scheduler::new(SchedulerConfig::default(), Arc::new(ModuleRegistry::new()))
        .map(Arc::new)
        .map_err(DatabaseError::ShardSpawn)
}

fn build_router(
    scheduler: &Arc<Scheduler>,
    config: &DatabaseConfig,
    mode: StartupMode,
) -> Result<ShardRouter, DatabaseError> {
    ShardRouter::new(
        Arc::clone(scheduler),
        &config.data_dir,
        config.shard_count,
        mode.shard_mode(),
    )
    .ok_or(DatabaseError::InvalidShardCount)
}

fn spawn_sync_schedulers(
    scheduler: &Arc<Scheduler>,
    config: &DatabaseConfig,
    router: &ShardRouter,
) -> Result<Vec<SyncSchedulerHandle>, DatabaseError> {
    let Some(distributed) = &config.distributed else {
        return Ok(Vec::new());
    };
    let topology = distributed
        .topology
        .clone()
        .ok_or(DatabaseError::MissingSyncTopology)?;
    let scheduler_config = SyncSchedulerConfig::new(
        distributed.local_node.clone(),
        distributed.nodes.clone(),
        topology,
        config.shard_count,
        Duration::from_millis(distributed.sync_interval),
    );
    // LAZY: the scheduler syncs ONLY materialised shards (via the router's
    // membership view) — an un-materialised shard holds no data, so pulling it
    // would be pure waste.
    SyncSchedulerHandle::spawn_with_shard_source(
        Arc::clone(scheduler),
        scheduler_config,
        Arc::new(NoopSyncPullTrigger),
        Arc::new(router.membership()),
        DEFAULT_TIMEOUT,
    )
    .map(|handle| vec![handle])
    .map_err(map_sync_scheduler_error)
}

fn map_sync_scheduler_error(error: SyncSchedulerError) -> DatabaseError {
    match error {
        SyncSchedulerError::Spawn(message) => DatabaseError::SyncSchedulerSpawn(message),
        other => DatabaseError::SyncSchedulerError(other.to_string()),
    }
}