mongreldb_log/
commit_log.rs

1//! Commit-log abstraction (spec sections 6.2 and 9.4, FND-004).
2//!
3//! [`CommitLog`] is the single authority through which commands become
4//! committed. Standalone mode has one implementation wrapping the shared WAL
5//! group commit (in `mongreldb-core`); replicated mode implements the same
6//! contract over Raft (Stage 2). The storage apply path receives only
7//! committed commands.
8
9use std::sync::atomic::{AtomicBool, Ordering};
10use std::sync::Arc;
11use std::time::Instant;
12
13use mongreldb_types::hlc::HlcTimestamp;
14use mongreldb_types::ids::TransactionId;
15
16use crate::envelope::{CommandEnvelope, EnvelopeError};
17
18/// A position in one commit log. `term` is zero for the standalone log.
19#[derive(
20    Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize,
21)]
22pub struct LogPosition {
23    /// Consensus term; zero in standalone mode.
24    pub term: u64,
25    /// Monotonic log index (the standalone epoch).
26    pub index: u64,
27}
28
29impl LogPosition {
30    /// The position before any entry.
31    pub const ZERO: Self = Self { term: 0, index: 0 };
32}
33
34/// One committed log entry returned by [`CommitLog::read_committed`].
35#[derive(Debug, Clone)]
36pub struct CommittedEntry {
37    /// Position of this entry in the log.
38    pub position: LogPosition,
39    /// Commit timestamp assigned by the log authority.
40    pub commit_ts: HlcTimestamp,
41    /// The committed command.
42    pub envelope: CommandEnvelope,
43}
44
45/// A point-in-time image of applied state at a log boundary.
46#[derive(Debug, Clone)]
47pub struct LogSnapshot {
48    /// Last log position included in the snapshot.
49    pub position: LogPosition,
50    /// Commit timestamp of `position`.
51    pub commit_ts: HlcTimestamp,
52    /// Opaque snapshot payload defined by the implementation.
53    pub data: Vec<u8>,
54}
55
56/// Durability guarantee attached to a committed command.
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
58pub enum DurabilityLevel {
59    /// Standalone shared-WAL group commit (fsync before acknowledge).
60    GroupCommit,
61    /// Replicated: leader-local disk only (optional lower guarantee).
62    LeaderDisk,
63    /// Replicated: persisted by a quorum (the replicated default, RPO 0).
64    Quorum,
65}
66
67/// Proof that a command crossed its durable commit fence (spec S1B-004).
68///
69/// Once a receipt exists, the caller is never told the write rolled back.
70#[derive(Debug, Clone)]
71pub struct CommitReceipt {
72    /// Transaction this command committed.
73    pub transaction_id: TransactionId,
74    /// Commit timestamp assigned by the log authority.
75    pub commit_ts: HlcTimestamp,
76    /// Position of the committed entry.
77    pub log_position: LogPosition,
78    /// Durability level that was satisfied.
79    pub durability: DurabilityLevel,
80}
81
82/// Deadline and cancellation for log operations.
83///
84/// This is a deliberately minimal mirror of `mongreldb_core`'s
85/// `execution::ExecutionControl`; the core type cannot move below the runtime
86/// crate in the dependency graph, so the core adapter converts (see
87/// `docs/architecture/adr/0002`).
88#[derive(Debug, Clone, Default)]
89pub struct ExecutionControl {
90    /// Optional absolute deadline; queue wait counts toward it.
91    pub deadline: Option<Instant>,
92    /// Cooperative cancellation flag.
93    pub cancellation: Option<Arc<AtomicBool>>,
94}
95
96impl ExecutionControl {
97    /// Returns an error if the operation is cancelled or past its deadline.
98    pub fn check(&self) -> Result<(), LogError> {
99        if let Some(flag) = &self.cancellation {
100            if flag.load(Ordering::Relaxed) {
101                return Err(LogError::Cancelled);
102            }
103        }
104        if let Some(deadline) = self.deadline {
105            if Instant::now() >= deadline {
106                return Err(LogError::DeadlineExceeded);
107            }
108        }
109        Ok(())
110    }
111}
112
113/// Errors produced by a [`CommitLog`].
114#[derive(Debug, thiserror::Error)]
115pub enum LogError {
116    /// The command envelope was malformed or unverifiable.
117    #[error(transparent)]
118    Envelope(#[from] EnvelopeError),
119    /// The operation was cancelled.
120    #[error("operation cancelled")]
121    Cancelled,
122    /// The operation's deadline expired.
123    #[error("deadline exceeded")]
124    DeadlineExceeded,
125    /// The log is closed and rejects new proposals.
126    #[error("commit log is closed")]
127    Closed,
128    /// The receiving replica is not the leader for the consensus group
129    /// (spec section 11.7). Retryable with the returned leader hint.
130    ///
131    /// Category mapping: this variant carries
132    /// `mongreldb_types::errors::ErrorCategory::NotLeader` semantics.
133    /// No `LogError` → `ErrorCategory` bridge exists yet (the engine
134    /// commit path does not convert `LogError` today); the bridge lands
135    /// with the Stage 2G gateway/routing wave, which consumes the hint.
136    #[error("not the leader (current leader: {leader_hint:?})")]
137    NotLeader {
138        /// The group's current leader hint when known (text form of the
139        /// leader identity, e.g. its raft node id); `None` when the group
140        /// has no known leader.
141        leader_hint: Option<String>,
142    },
143    /// The implementation does not provide this operation.
144    #[error("unsupported operation: {0}")]
145    Unsupported(&'static str),
146    /// Any other log failure.
147    #[error("commit log failure: {0}")]
148    Internal(String),
149}
150
151/// The single authority through which commands become committed.
152pub trait CommitLog: Send + Sync {
153    /// Proposes one command and waits until its durability policy is
154    /// satisfied. A returned [`CommitReceipt`] is irrevocable.
155    fn propose(
156        &self,
157        command: CommandEnvelope,
158        control: &ExecutionControl,
159    ) -> Result<CommitReceipt, LogError>;
160
161    /// Reads committed entries strictly after `after`, in log order.
162    fn read_committed(
163        &self,
164        after: LogPosition,
165        limit: usize,
166    ) -> Result<Vec<CommittedEntry>, LogError>;
167
168    /// The highest position the local state machine has applied.
169    fn applied_position(&self) -> LogPosition;
170
171    /// Captures applied state through the current applied position.
172    fn create_snapshot(&self) -> Result<LogSnapshot, LogError>;
173
174    /// Replaces applied state with a snapshot taken at a log boundary.
175    fn install_snapshot(&self, snapshot: LogSnapshot) -> Result<(), LogError>;
176}
mongreldb_log/commit_log.rs

mongreldb_log/
commit_log.rs
