clt_database/mvcc/persistent_storage/

mod.rs

use crate::io::FileSyncType;
use crate::storage::encryption::EncryptionContext;
use crate::storage::sqlite3_ondisk::DatabaseHeader;
use crate::sync::atomic::{AtomicI64, AtomicU64, Ordering};
use crate::sync::Arc;
use crate::sync::RwLock;
use crate::turso_assert;
use std::fmt::Debug;

pub mod logical_log;
use crate::mvcc::database::{LogRecord, RowVersion};
use crate::mvcc::persistent_storage::logical_log::{
    serialize_header_entry, serialize_op_entry, LogicalLog, OnSerializationComplete,
    DEFAULT_LOG_CHECKPOINT_THRESHOLD,
};
use crate::{CheckpointResult, Completion, File, LimboError, Result};

pub trait DurableStorage: Send + Sync + Debug {
    /// Append one row-version op to `log_record`'s payload buffer, in the
    /// on-disk wire format used by the logical log. Updates `op_count`.
    fn serialize_row_version(
        &self,
        log_record: &mut LogRecord,
        row_version: &RowVersion,
        portable_extension: Option<&[u8]>,
    ) -> Result<()>;

    /// Append a `DatabaseHeader` op to `log_record`'s payload buffer.
    fn serialize_database_header(
        &self,
        log_record: &mut LogRecord,
        header: &DatabaseHeader,
    ) -> Result<()>;

    /// Write a transaction to the logical log without advancing the writer offset.
    ///
    /// If `on_serialization_complete` is provided, it is called with shared
    /// ownership of the framed bytes and the running CRC after framing but
    /// before the disk write. The callback runs while the internal write lock
    /// is held, so it should be fast.
    fn log_tx(
        &self,
        m: LogRecord,
        on_serialization_complete: OnSerializationComplete<'_>,
    ) -> Result<(Completion, u64)>;

    /// If `m` needs a logical-log header upgrade before it can be appended,
    /// start that write and return its completion. Callers must wait for this
    /// completion and then call `log_tx`.
    fn upgrade_header_for_log_tx(&self, m: &LogRecord) -> Result<Option<Completion>>;

    fn sync(&self, sync_type: FileSyncType) -> Result<Completion>;

    /// Called after a logical-log write completed successfully, before the
    /// transaction is made visible by advancing the logical-log offset.
    ///
    /// Implementations may return a completion for any additional durability
    /// work that must finish before commit publication.
    fn on_log_write_complete(&self) -> Result<Completion> {
        Ok(Completion::new_yield())
    }

    /// Persist the current logical-log header to durable storage.
    ///
    /// This is used by MVCC recovery/checkpoint flows. Keeping this in the trait avoids
    /// reaching into concrete storage internals.
    fn update_header(&self) -> Result<Completion>;

    /// Truncate the logical log, discarding frames at or below
    /// `checkpointed_through_ts` (the checkpoint's published boundary). Frames
    /// above the boundary (uncheckpointed concurrent commits) are preserved.
    fn truncate(&self, checkpointed_through_ts: u64) -> Result<Completion>;

    /// Reset the logical log to a fresh header-only file.
    ///
    /// Used after an external database restore so future MVCC recovery starts
    /// from the restored image instead of replaying stale local log frames.
    fn reset_to_fresh_header(&self) -> Result<Completion>;
    fn get_logical_log_file(&self) -> Arc<dyn File>;
    fn logical_log_offset(&self) -> u64;
    fn should_checkpoint(&self) -> bool;
    /// Set the checkpoint threshold in bytes of logical-log data written.
    /// A negative value disables automatic checkpointing.
    fn set_checkpoint_threshold(&self, threshold: i64);
    fn checkpoint_threshold(&self) -> i64;
    fn advance_logical_log_offset_after_success(&self, bytes: u64) -> Result<()>;
    fn discard_pending_log_write(&self) -> Result<()> {
        Ok(())
    }
    fn restore_logical_log_state_after_recovery(&self, offset: u64, running_crc: u32);

    /// Set the in-memory log header from a previously-read on-disk header.
    ///
    /// Called during recovery to seed the CRC state from the header's salt.
    fn set_header(&self, header: logical_log::LogHeader);

    /// Called when a checkpoint begins, before any rows are written to the B-tree.
    fn on_checkpoint_start(&self) -> Result<()> {
        Ok(())
    }

    /// Called after the checkpoint has fully completed: rows are flushed, WAL is
    /// truncated, and the logical log is reset.
    fn on_checkpoint_end(&self, _result: Result<&CheckpointResult>) -> Result<()> {
        Ok(())
    }

    fn encryption_ctx(&self) -> Option<EncryptionContext> {
        None
    }
}

pub struct Storage {
    pub logical_log: RwLock<LogicalLog>,
    /// Shadowed from LogicalLog::offset for lock-free should_checkpoint() reads.
    log_offset: AtomicU64,
    checkpoint_threshold: AtomicI64,
}

impl Storage {
    pub fn new(
        file: Arc<dyn File>,
        io: Arc<dyn crate::IO>,
        encryption_ctx: Option<EncryptionContext>,
    ) -> Self {
        Self {
            logical_log: RwLock::new(LogicalLog::new(file, io, encryption_ctx)),
            log_offset: AtomicU64::new(0),
            checkpoint_threshold: AtomicI64::new(DEFAULT_LOG_CHECKPOINT_THRESHOLD),
        }
    }

    /// Update the shadow offset to stay in sync with LogicalLog::offset.
    /// Called after any operation that mutates the canonical offset under the write lock.
    #[inline(always)]
    fn shadow_offset_store(&self, value: u64) {
        self.log_offset.store(value, Ordering::Relaxed);
    }

    #[inline(always)]
    fn shadow_offset_advance(&self, bytes: u64) {
        self.log_offset.fetch_add(bytes, Ordering::Relaxed);
    }
}

impl DurableStorage for Storage {
    fn serialize_row_version(
        &self,
        log_record: &mut LogRecord,
        row_version: &RowVersion,
        portable_extension: Option<&[u8]>,
    ) -> Result<()> {
        serialize_op_entry(&mut log_record.buf, row_version, portable_extension)?;
        log_record.op_count = log_record.op_count.checked_add(1).ok_or_else(|| {
            LimboError::InternalError("logical log op_count exceeds u32".to_string())
        })?;
        Ok(())
    }

    fn serialize_database_header(
        &self,
        log_record: &mut LogRecord,
        header: &DatabaseHeader,
    ) -> Result<()> {
        turso_assert!(
            !log_record.has_header,
            "DatabaseHeader op appended more than once to a single LogRecord"
        );
        serialize_header_entry(&mut log_record.buf, header);
        log_record.has_header = true;
        log_record.op_count = log_record.op_count.checked_add(1).ok_or_else(|| {
            LimboError::InternalError("logical log op_count exceeds u32".to_string())
        })?;
        Ok(())
    }

    fn log_tx(
        &self,
        m: LogRecord,
        on_serialization_complete: OnSerializationComplete<'_>,
    ) -> Result<(Completion, u64)> {
        self.logical_log
            .write()
            .log_tx_deferred_offset(m, on_serialization_complete)
    }

    fn upgrade_header_for_log_tx(&self, m: &LogRecord) -> Result<Option<Completion>> {
        self.logical_log.write().upgrade_header_for_log_tx(m)
    }

    fn sync(&self, sync_type: FileSyncType) -> Result<Completion> {
        self.logical_log.write().sync(sync_type)
    }

    fn update_header(&self) -> Result<Completion> {
        self.logical_log.write().update_header()
    }

    fn truncate(&self, checkpointed_through_ts: u64) -> Result<Completion> {
        let mut log = self.logical_log.write();
        let c = log.truncate(checkpointed_through_ts)?;
        // Shadow the log's actual offset: 0 if it truncated, unchanged if it
        // skipped (uncheckpointed frames remain), so should_checkpoint() stays
        // accurate.
        let new_offset = log.offset;
        drop(log);
        self.shadow_offset_store(new_offset);
        Ok(c)
    }

    fn reset_to_fresh_header(&self) -> Result<Completion> {
        let c = self.logical_log.write().reset_to_fresh_header()?;
        self.shadow_offset_store(0);
        Ok(c)
    }

    fn get_logical_log_file(&self) -> Arc<dyn File> {
        self.logical_log.read().file.clone()
    }

    fn logical_log_offset(&self) -> u64 {
        self.log_offset.load(Ordering::Relaxed)
    }

    fn encryption_ctx(&self) -> Option<EncryptionContext> {
        self.logical_log.read().encryption_ctx().cloned()
    }

    /// Lock-free: reads shadowed atomics only.
    fn should_checkpoint(&self) -> bool {
        let threshold = self.checkpoint_threshold.load(Ordering::Relaxed);
        if threshold < 0 {
            return false;
        }
        self.log_offset.load(Ordering::Relaxed) >= threshold as u64
    }

    fn set_checkpoint_threshold(&self, threshold: i64) {
        self.checkpoint_threshold
            .store(threshold, Ordering::Relaxed);
    }

    fn checkpoint_threshold(&self) -> i64 {
        self.checkpoint_threshold.load(Ordering::Relaxed)
    }

    fn advance_logical_log_offset_after_success(&self, bytes: u64) -> Result<()> {
        self.logical_log.write().advance_offset_after_success(bytes);
        self.shadow_offset_advance(bytes);
        Ok(())
    }

    fn restore_logical_log_state_after_recovery(&self, offset: u64, running_crc: u32) {
        let mut log = self.logical_log.write();
        log.offset = offset;
        log.running_crc = running_crc;
        self.shadow_offset_store(offset);
    }

    fn set_header(&self, header: logical_log::LogHeader) {
        self.logical_log.write().set_header(header);
    }
}

impl Debug for Storage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "LogicalLog {{ logical_log }}")
    }
}