nodedb 0.4.0

// SPDX-License-Identifier: BUSL-1.1

use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicU64;
use std::sync::{Arc, Mutex};

use tracing::info;

use nodedb_types::config::tuning::WalTuning;
use nodedb_wal::segmented::{SegmentedWal, SegmentedWalConfig};
use nodedb_wal::writer::WalWriterConfig;

/// WAL manager: owns the segmented WAL and coordinates appends + sync.
///
/// The WAL is the single source of truth for durability. Every mutation
/// goes through here before being applied to any engine's in-memory state.
///
/// Thread-safety: the segmented WAL is behind a `Mutex` because multiple
/// Control Plane tasks may submit WAL appends concurrently. The mutex
/// serializes writes, which is correct — WAL appends must be ordered anyway.
/// The `sync()` call (fsync) is the expensive part and is batched via group commit.
pub struct WalManager {
    /// The segmented WAL. `Arc` so the group-commit fsync
    /// ([`WalManager::wait_durable`]) can move a handle into a `spawn_blocking`
    /// thread — the O_DIRECT `sync()` must never run inline on a Tokio worker.
    pub(super) wal: Arc<Mutex<SegmentedWal>>,
    /// The WAL directory path (for replay without holding the lock).
    pub(super) wal_dir: PathBuf,
    /// Encryption key ring (if configured). Supports dual-key reads during rotation.
    pub(super) encryption_ring: Option<nodedb_wal::crypto::KeyRing>,
    /// Dedicated audit WAL segment. When present, audit entries are written
    /// atomically alongside data writes. Append-only, never compacted.
    pub(super) audit_wal: Option<crate::wal::AuditWalSegment>,
    /// Highest LSN known to be fsync-durable. Advanced only by the
    /// group-commit leader in [`WalManager::wait_durable`] after a successful
    /// `sync()`, never past what that fsync actually made durable.
    pub(super) durable_lsn: AtomicU64,
    /// Async group-commit leader election: the first `wait_durable` caller to
    /// acquire this becomes the leader and performs the single coalesced fsync
    /// covering every concurrently-buffered write. Async so waiting for
    /// leadership never blocks a Tokio worker.
    pub(super) commit_lock: tokio::sync::Mutex<()>,
    /// Wakes `wait_durable` followers when `durable_lsn` advances (or a leader's
    /// fsync fails, so they re-attempt and observe the same error).
    pub(super) durable_notify: tokio::sync::Notify,
}

impl WalManager {
    /// Open or create a segmented WAL at the given path.
    ///
    /// The `path` argument is the WAL directory for the segmented format.
    pub fn open(path: &Path, use_direct_io: bool) -> crate::Result<Self> {
        Self::open_with_segment_size(path, use_direct_io, 0)
    }

    /// Open with explicit segment target size (bytes). 0 = default (64 MiB).
    pub fn open_with_segment_size(
        path: &Path,
        use_direct_io: bool,
        segment_target_size: u64,
    ) -> crate::Result<Self> {
        Self::open_internal(
            path,
            segment_target_size,
            WalWriterConfig {
                use_direct_io,
                ..Default::default()
            },
        )
    }

    /// Open with explicit segment target size and WAL tuning from `TuningConfig`.
    ///
    /// Uses `tuning.write_buffer_size` and `tuning.alignment` from [`WalTuning`]
    /// to configure the underlying `WalWriterConfig`, instead of the hardcoded
    /// defaults. `segment_target_size` of 0 uses the default (64 MiB).
    pub fn open_with_tuning(
        path: &Path,
        use_direct_io: bool,
        segment_target_size: u64,
        tuning: &WalTuning,
    ) -> crate::Result<Self> {
        Self::open_internal(
            path,
            segment_target_size,
            WalWriterConfig {
                write_buffer_size: tuning.write_buffer_size,
                alignment: tuning.alignment,
                use_direct_io,
                dwb_mode: None,
            },
        )
    }

    /// Shared WAL open logic: resolve segment size, open.
    pub(super) fn open_internal(
        path: &Path,
        segment_target_size: u64,
        writer_config: WalWriterConfig,
    ) -> crate::Result<Self> {
        let wal_dir = path.to_path_buf();

        let effective_target = if segment_target_size > 0 {
            segment_target_size
        } else {
            nodedb_wal::segment::DEFAULT_SEGMENT_TARGET_SIZE
        };

        let use_direct_io = writer_config.use_direct_io;
        let config = SegmentedWalConfig {
            wal_dir: wal_dir.clone(),
            segment_target_size: effective_target,
            writer_config,
        };

        let wal = SegmentedWal::open(config).map_err(crate::Error::Wal)?;

        info!(
            wal_dir = %wal_dir.display(),
            next_lsn = wal.next_lsn(),
            "WAL opened"
        );

        let audit_dir = wal_dir.join("audit.wal");
        let audit_wal = match crate::wal::AuditWalSegment::open(&audit_dir, use_direct_io) {
            Ok(aw) => {
                info!(audit_dir = %audit_dir.display(), "audit WAL opened");
                Some(aw)
            }
            Err(e) => {
                tracing::warn!(error = %e, "audit WAL failed to open (audit entries not durable)");
                None
            }
        };

        Ok(Self {
            wal: Arc::new(Mutex::new(wal)),
            wal_dir,
            encryption_ring: None,
            audit_wal,
            durable_lsn: AtomicU64::new(0),
            commit_lock: tokio::sync::Mutex::new(()),
            durable_notify: tokio::sync::Notify::new(),
        })
    }

    /// Open without O_DIRECT (for testing on tmpfs).
    pub fn open_for_testing(path: &Path) -> crate::Result<Self> {
        Self::open(path, false)
    }

    /// Get the WAL directory path.
    pub fn wal_dir(&self) -> &Path {
        &self.wal_dir
    }
}