ozzy_journal/
progress.rs

1//! Generation-scoped accounting for complete journal writes and syncs.
2//!
3//! This does not perform I/O or decide commit. The adapter reports a write only
4//! after a complete sealed group, and a sync only after the required barrier.
5//! Tickets identify local actions, not hashes, quorum evidence, or disk proofs.
6
7/// One active journal writer incarnation, supplied by the owning runtime.
8///
9/// Never reuse a value across journals, reopen, or selected-lineage replacement.
10/// It is independent of the persisted segment/manifest generation and VSR view.
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub struct JournalGeneration(pub u128);
13
14/// Contiguous group-operation prefix. Zero denotes the empty journal.
15#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord)]
16pub struct OpNumber(pub u64);
17
18/// One admitted, indivisible local write group. Fields cannot be forged externally.
19#[derive(Debug, Clone, Copy, PartialEq, Eq)]
20pub struct WriteTicket {
21    generation: JournalGeneration,
22    first: OpNumber,
23    through: OpNumber,
24}
25
26impl WriteTicket {
27    /// Writer incarnation that admitted this group.
28    pub const fn generation(self) -> JournalGeneration {
29        self.generation
30    }
31    /// First operation in this group, inclusive.
32    pub const fn first(self) -> OpNumber {
33        self.first
34    }
35    /// Last operation in this group, inclusive.
36    pub const fn through(self) -> OpNumber {
37        self.through
38    }
39}
40
41/// Prefix captured when a synchronization request is issued.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub struct SyncTicket {
44    generation: JournalGeneration,
45    through: OpNumber,
46}
47
48impl SyncTicket {
49    /// Writer incarnation that issued this synchronization.
50    pub const fn generation(self) -> JournalGeneration {
51        self.generation
52    }
53    /// Complete written prefix this synchronization may acknowledge.
54    pub const fn through(self) -> OpNumber {
55        self.through
56    }
57}
58
59/// Read-only local progress. None of these positions implies quorum commit.
60#[derive(Debug, Clone, Copy, PartialEq, Eq)]
61pub struct JournalSnapshot {
62    /// Current writer incarnation.
63    pub generation: JournalGeneration,
64    /// Last admitted operation.
65    pub accepted: OpNumber,
66    /// Last operation in a completely written group.
67    pub written: OpNumber,
68    /// Last operation covered by successful sync evidence.
69    pub durable: OpNumber,
70    /// Uncertain I/O outcome blocks subsequent actions until recovery.
71    pub faulted: bool,
72}
73
74/// Single-writer local completion state. No allocation, runtime, or I/O.
75#[derive(Debug)]
76pub struct JournalProgress {
77    state: JournalSnapshot,
78}
79
80impl JournalProgress {
81    /// Start from an independently verified and synchronized recovered prefix.
82    ///
83    /// The caller must complete storage recovery before supplying this prefix.
84    /// Bytes visible only in a surviving OS page cache are not stable evidence.
85    /// Reopening requires a fresh generation, including after an I/O error.
86    pub const fn recover(generation: JournalGeneration, through: OpNumber) -> Self {
87        Self {
88            state: JournalSnapshot {
89                generation,
90                accepted: through,
91                written: through,
92                durable: through,
93                faulted: false,
94            },
95        }
96    }
97
98    /// Inspect progress without obtaining mutation authority.
99    pub const fn snapshot(&self) -> JournalSnapshot {
100        self.state
101    }
102
103    /// Reserve a contiguous group after external payload/capacity validation.
104    /// Errors leave all state unchanged. Zero and counter wrap are rejected.
105    pub fn admit(&mut self, operations: u64) -> Result<WriteTicket, ProgressError> {
106        self.require_active(self.state.generation)?;
107        if operations == 0 {
108            return Err(ProgressError::EmptyGroup);
109        }
110        let through = self
111            .state
112            .accepted
113            .0
114            .checked_add(operations)
115            .ok_or(ProgressError::OpExhausted)?;
116        let ticket = WriteTicket {
117            generation: self.state.generation,
118            first: OpNumber(self.state.accepted.0 + 1),
119            through: OpNumber(through),
120        };
121        self.state.accepted = ticket.through;
122        Ok(ticket)
123    }
124
125    /// Report a complete sealed write, in journal order. Duplicates are harmless.
126    /// Partial writes must not call this method. They remain the adapter's work.
127    pub fn complete_write(&mut self, ticket: WriteTicket) -> Result<(), ProgressError> {
128        self.require_active(ticket.generation)?;
129        if ticket.through > self.state.accepted {
130            return Err(ProgressError::BeyondAccepted);
131        }
132        if ticket.through <= self.state.written {
133            return Ok(());
134        }
135        if Some(ticket.first.0) != self.state.written.0.checked_add(1) {
136            return Err(ProgressError::WriteGap);
137        }
138        self.state.written = ticket.through;
139        Ok(())
140    }
141
142    /// Capture the complete written prefix before submitting a sync action.
143    pub fn begin_sync(&self) -> Result<SyncTicket, ProgressError> {
144        self.require_active(self.state.generation)?;
145        Ok(SyncTicket {
146            generation: self.state.generation,
147            through: self.state.written,
148        })
149    }
150
151    /// Report successful sync for the captured prefix, not newer written data.
152    /// Duplicate/older completions cannot lower the durable prefix.
153    pub fn complete_sync(&mut self, ticket: SyncTicket) -> Result<(), ProgressError> {
154        self.require_active(ticket.generation)?;
155        if ticket.through > self.state.written {
156            return Err(ProgressError::BeyondWritten);
157        }
158        self.state.durable = self.state.durable.max(ticket.through);
159        Ok(())
160    }
161
162    /// Report one complete write whose backend operation also synchronized the
163    /// exact group before returning. This supports transactional local adapters
164    /// that do not expose separate write and sync callbacks.
165    pub fn complete_durable_write(&mut self, ticket: WriteTicket) -> Result<(), ProgressError> {
166        self.complete_write(ticket)?;
167        self.state.durable = self.state.durable.max(ticket.through);
168        Ok(())
169    }
170
171    /// An uncertain write/sync faults this generation. Past evidence is retained.
172    /// A stale error from a replaced generation cannot fault its replacement.
173    pub fn fail(&mut self, generation: JournalGeneration) -> Result<(), ProgressError> {
174        if generation != self.state.generation {
175            return Err(ProgressError::StaleGeneration);
176        }
177        self.fault();
178        Ok(())
179    }
180
181    /// Fault the current generation after its owner found conflicting history.
182    pub fn fault(&mut self) {
183        self.state.faulted = true;
184    }
185
186    fn require_active(&self, generation: JournalGeneration) -> Result<(), ProgressError> {
187        if generation != self.state.generation {
188            return Err(ProgressError::StaleGeneration);
189        }
190        if self.state.faulted {
191            return Err(ProgressError::Faulted);
192        }
193        Ok(())
194    }
195}
196
197/// Invalid or stale journal transition. No error advances any prefix.
198#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
199pub enum ProgressError {
200    /// The writer has an uncertain I/O outcome and needs recovery.
201    #[error("journal generation is faulted")]
202    Faulted,
203    /// Completion belongs to another writer incarnation.
204    #[error("stale journal generation")]
205    StaleGeneration,
206    /// A write group must contain at least one operation.
207    #[error("empty journal group")]
208    EmptyGroup,
209    /// Operation numbering cannot wrap.
210    #[error("journal operation number exhausted")]
211    OpExhausted,
212    /// Completion claims operations that were never admitted.
213    #[error("write completion exceeds admitted prefix")]
214    BeyondAccepted,
215    /// An earlier complete write has not been reported.
216    #[error("write completion leaves a journal gap")]
217    WriteGap,
218    /// Sync claims data not known to have been completely written.
219    #[error("sync completion exceeds written prefix")]
220    BeyondWritten,
221}
222
223#[cfg(kani)]
224mod proofs;
ozzy_journal/progress.rs

ozzy_journal/
progress.rs
