ozzy_replication/
normal.rs

1mod admission;
2mod commit;
3mod installation;
4
5use std::collections::VecDeque;
6
7use ozzy_journal::progress::{JournalProgress, JournalSnapshot};
8use ozzy_proto::NodeId;
9
10use crate::{
11    Configuration, JournalGeneration, OpNumber, PipelineLimits, Prefix, PreparedOperation,
12    ReplicationError, Scope, SyncTicket, VoterIndex, Voters, WriteTicket,
13};
14
15/// Whether the normal protocol may emit votes, commit, or apply.
16#[derive(Debug, Clone, Copy, PartialEq, Eq)]
17pub enum Status {
18    /// Exact installed view permits normal operation.
19    Normal,
20    /// Old-view decisions stopped; outstanding disk work may still settle.
21    Fenced,
22    /// Conflicting lineage or uncertain storage requires recovery/quarantine.
23    Faulted,
24}
25
26/// Result of atomically validating a complete prepare group.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum Admission {
29    /// Every operation was already accepted with identical metadata.
30    Duplicate,
31    /// Persist only this newly admitted suffix of the submitted operation slice.
32    Write {
33        /// Generation-scoped exact write range, not durability evidence.
34        ticket: WriteTicket,
35        /// Index of the first new operation in the supplied slice.
36        first_new: usize,
37    },
38}
39
40/// Read-only normal state, suitable for adapter scheduling and invariant checks.
41#[derive(Debug, Clone, Copy, PartialEq, Eq)]
42pub struct ReplicaSnapshot {
43    /// Current group/configuration/view identity.
44    pub scope: Scope,
45    /// Whether normal protocol actions are still allowed.
46    pub status: Status,
47    /// Independent local write and synchronization evidence.
48    pub journal: JournalSnapshot,
49    /// Highest validated, contiguous canonical operation.
50    pub accepted: Prefix,
51    /// Highest contiguous quorum-committed operation under the configured policy.
52    pub committed: Prefix,
53    /// Highest operation reported applied by the deterministic state adapter.
54    pub applied: Prefix,
55    /// Retained live descriptors awaiting application or local persistence.
56    /// Excludes a disk-backed `journal_tail`.
57    pub pending_operations: usize,
58    /// Live canonical body bytes reserved in the external payload adapter.
59    pub pending_body_bytes: usize,
60    /// Operations above this number occupy the primary's live window: the
61    /// commit it announced or the released prefix, whichever is later.
62    pub live_base: OpNumber,
63    /// Descriptor and body capacity this role retains beyond its released prefix.
64    pub retained_limits: PipelineLimits,
65    /// Installed/recovered suffix retained only in the journal, not the live
66    /// descriptor ring. Fresh work waits until this entire tail is applied.
67    pub journal_tail: Option<Prefix>,
68    /// Primary only: the highest operation each backup acknowledged.
69    pub backups: Option<[(NodeId, OpNumber); 2]>,
70    /// Highest commit the primary announced, as this replica knows it.
71    pub known_commit: OpNumber,
72    /// Whether new operations may be admitted. Duplicates can still be retried
73    /// while an installed primary reestablishes its selected-tail quorum. Entries
74    /// inside a disk-backed tail require journal lookup, not descriptor retries.
75    pub ready_for_appends: bool,
76}
77
78/// Fixed-three-voter normal VSR path with a bounded pipelined prepare window.
79///
80/// This is one component of VSR, not a complete election or recovery protocol.
81/// The caller authenticates sender IDs and validates canonical operations plus
82/// application transitions before admission. Returned write tickets schedule
83/// disk work; the primary may concurrently send the same prepares to backups.
84/// Producer success additionally requires ordered application of committed ops.
85#[derive(Debug)]
86pub struct NormalReplica {
87    configuration: Configuration,
88    local: VoterIndex,
89    status: Status,
90    journal: JournalProgress,
91    limits: PipelineLimits,
92    operations: VecDeque<PreparedOperation>,
93    body_bytes: usize,
94    accepted: Prefix,
95    committed: Prefix,
96    applied: Prefix,
97    // Descriptors start after the applied prefix stored under the group policy.
98    retained_base: Prefix,
99    known_commit: OpNumber,
100    acknowledged: Voters<OpNumber>,
101    activation_tail: OpNumber,
102    installed_view: Option<crate::StartView>,
103    journal_tail: Option<Prefix>,
104}
105
106impl NormalReplica {
107    /// Exact prefix released from the live window after application and persistence.
108    pub const fn reclaimed_prefix(&self) -> Prefix {
109        self.retained_base
110    }
111    pub(crate) const fn driver_identity(&self) -> (Configuration, VoterIndex) {
112        (self.configuration, self.local)
113    }
114
115    /// Consume normal-role ownership and immediately fence its decisions.
116    ///
117    /// The new view must exceed the installed view. Outstanding journal work
118    /// moves into the view-change core and must settle before a view promise.
119    /// This transition cannot resume the old normal role, even on failure.
120    pub fn into_view_change(self, view: u64) -> Result<crate::ViewChange, crate::ViewChangeError> {
121        let configuration = self.configuration;
122        let local = self.local;
123        crate::ViewChange::begin(self, configuration, local, view)
124    }
125
126    /// Start a newly formatted, empty group at view zero.
127    ///
128    /// All three durable stores must have been explicitly initialized with this
129    /// exact configuration. Never use this constructor on restart, lost storage,
130    /// or a previously used voter identity. It cannot recover existing history.
131    /// `generation` must be unique across all writer instances/completions.
132    pub fn bootstrap(
133        configuration: Configuration,
134        local: NodeId,
135        generation: JournalGeneration,
136        limits: PipelineLimits,
137    ) -> Result<Self, ReplicationError> {
138        let local = configuration.voter_index(local)?;
139        if limits.max_operations == 0
140            || limits.max_body_bytes == 0
141            || u64::try_from(limits.max_operations).is_err()
142        {
143            return Err(ReplicationError::InvalidLimits);
144        }
145        // A later installed view can change the role; the ring then grows once.
146        let retained =
147            if configuration.primary(configuration.scope.view) == configuration.voter(local) {
148                limits
149            } else {
150                limits.backup()
151            };
152        let mut operations = VecDeque::new();
153        operations
154            .try_reserve_exact(retained.max_operations)
155            .map_err(|_| ReplicationError::Capacity)?;
156        Ok(Self {
157            configuration,
158            local,
159            status: Status::Normal,
160            journal: JournalProgress::recover(generation, OpNumber(0)),
161            limits,
162            operations,
163            body_bytes: 0,
164            accepted: Prefix::GENESIS,
165            committed: Prefix::GENESIS,
166            applied: Prefix::GENESIS,
167            retained_base: Prefix::GENESIS,
168            known_commit: OpNumber(0),
169            acknowledged: Voters::splat(OpNumber(0)),
170            activation_tail: OpNumber(0),
171            installed_view: None,
172            journal_tail: None,
173        })
174    }
175
176    /// Reconstruct only fenced disk evidence, not application or normal authority.
177    /// Complete retained history stays in the journal, outside the live pipeline.
178    pub(crate) fn recover_fenced(
179        mut configuration: Configuration,
180        local: NodeId,
181        generation: JournalGeneration,
182        log: crate::FrozenLog,
183        limits: PipelineLimits,
184    ) -> Result<Self, ReplicationError> {
185        configuration.scope.view = log.last_normal_view;
186        let mut normal = Self::bootstrap(configuration, local, generation, limits)?;
187        normal.restore_fenced(generation, log);
188        Ok(normal)
189    }
190
191    /// Reuse the bounded ring after storage confirms unpublished staging was aborted.
192    /// Restore only the old durable history. Never restore votes or application authority.
193    pub(crate) fn restore_fenced(&mut self, generation: JournalGeneration, log: crate::FrozenLog) {
194        self.configuration.scope.view = log.last_normal_view;
195        self.status = Status::Fenced;
196        self.journal = JournalProgress::recover(generation, log.accepted.op);
197        self.operations.clear();
198        self.body_bytes = 0;
199        self.accepted = log.accepted;
200        self.committed = log.committed;
201        self.applied = Prefix::GENESIS;
202        self.retained_base = Prefix::GENESIS;
203        self.known_commit = log.committed.op;
204        self.acknowledged = Voters::splat(OpNumber(0));
205        self.activation_tail = log.accepted.op;
206        self.installed_view = None;
207        self.journal_tail = (log.accepted != Prefix::GENESIS).then_some(log.accepted);
208    }
209
210    /// Current group/configuration/view identity, without a full snapshot.
211    pub const fn scope(&self) -> Scope {
212        self.configuration.scope()
213    }
214
215    /// Inspect state without obtaining mutable journal or quorum access.
216    pub fn snapshot(&self) -> ReplicaSnapshot {
217        ReplicaSnapshot {
218            scope: self.scope(),
219            status: self.status,
220            journal: self.journal.snapshot(),
221            accepted: self.accepted,
222            committed: self.committed,
223            applied: self.applied,
224            pending_operations: self.operations.len(),
225            pending_body_bytes: self.body_bytes,
226            live_base: self.live_base(),
227            retained_limits: self.retained_limits(),
228            journal_tail: self.journal_tail,
229            backups: self.is_primary().then(|| {
230                self.local
231                    .others()
232                    .map(|voter| (self.configuration.voter(voter), self.acknowledged[voter]))
233            }),
234            known_commit: self.known_commit,
235            ready_for_appends: self.ready_for_appends(),
236        }
237    }
238
239    /// Operations above this number occupy the primary's live window.
240    fn live_base(&self) -> OpNumber {
241        self.known_commit.max(self.retained_base.op)
242    }
243
244    /// Descriptor and body capacity retained beyond the released prefix.
245    ///
246    /// The primary keeps one window in flight beyond its released prefix. A
247    /// backup releases only what the primary has announced committed, and its
248    /// own persistence can trail that announcement by a whole window while the
249    /// other backup confirms the primary's writes, so it retains two.
250    fn retained_limits(&self) -> PipelineLimits {
251        if self.is_primary() {
252            self.limits
253        } else {
254            self.limits.backup()
255        }
256    }
257
258    /// Observe a complete local write. Page-cache completion casts no vote.
259    pub fn complete_write(&mut self, ticket: WriteTicket) -> Result<(), ReplicationError> {
260        self.journal.complete_write(ticket)?;
261        self.reclaim()
262    }
263
264    /// Local synchronized prefix available for replay and releasing applied
265    /// live capacity. It alone never establishes a quorum.
266    pub fn stored_through(&self) -> OpNumber {
267        self.journal.snapshot().durable
268    }
269
270    /// Capture the complete written prefix before submitting a disk barrier.
271    pub fn begin_sync(&self) -> Result<SyncTicket, ReplicationError> {
272        Ok(self.journal.begin_sync()?)
273    }
274
275    /// Observe an actual successful barrier for its captured prefix only.
276    /// Fenced completions update disk evidence but cannot advance commit.
277    pub fn complete_sync(&mut self, ticket: SyncTicket) -> Result<(), ReplicationError> {
278        self.journal.complete_sync(ticket)?;
279        self.advance_commit()?;
280        self.reclaim()
281    }
282
283    /// Observe an adapter write that also synchronized its complete prefix.
284    pub fn complete_durable_write(&mut self, ticket: WriteTicket) -> Result<(), ReplicationError> {
285        self.journal.complete_durable_write(ticket)?;
286        self.advance_commit()?;
287        self.reclaim()
288    }
289
290    /// Report ordered application. Release capacity only through `stored_through`.
291    ///
292    /// The caller asserts every operation through this exact prefix was applied
293    /// in order, not only the last one. This is not consumer processing. Payloads
294    /// needed for retries, replay, or lagging replicas remain in the journal.
295    pub fn apply_through(&mut self, through: Prefix) -> Result<(), ReplicationError> {
296        self.require_normal()?;
297        if through.op < self.applied.op {
298            return Err(ReplicationError::HistoryUnavailable);
299        }
300        if through.op > self.committed.op {
301            return Err(ReplicationError::ApplyBeyondCommit);
302        }
303        if self.prefix_at(through.op)? != through {
304            return Err(self.conflict());
305        }
306        self.applied = through;
307        self.reclaim()?;
308        if self.journal_tail.is_some_and(|tail| through.op >= tail.op) {
309            self.journal_tail = None;
310        }
311        Ok(())
312    }
313
314    /// Immediately stop old-view ACKs, commit, admission, and application.
315    /// Disk completions can still settle for a subsequent view-change report.
316    /// There is deliberately no un-fence or same-view restart method.
317    pub fn fence(&mut self) {
318        if self.status == Status::Normal {
319            self.status = Status::Fenced;
320        }
321    }
322
323    /// Report uncertain storage failure. Stale generations cannot fault this core.
324    pub fn fail_io(&mut self, generation: JournalGeneration) -> Result<(), ReplicationError> {
325        self.journal.fail(generation)?;
326        self.fault();
327        Ok(())
328    }
329
330    /// Fault the current generation after this core found conflicting history.
331    pub(crate) fn fault(&mut self) {
332        self.journal.fault();
333        self.status = Status::Faulted;
334    }
335
336    fn is_primary(&self) -> bool {
337        self.configuration.voter(self.local)
338            == self.configuration.primary(self.configuration.scope.view)
339    }
340
341    /// Highest matching evidence from either distinct backup.
342    fn backup_acknowledged(&self) -> OpNumber {
343        let [left, right] = self.local.others();
344        self.acknowledged[left].max(self.acknowledged[right])
345    }
346
347    fn ready_for_appends(&self) -> bool {
348        self.status == Status::Normal
349            && self.journal_tail.is_none()
350            && (!self.is_primary()
351                || (self.applied.op >= self.activation_tail
352                    && self.journal.snapshot().durable >= self.activation_tail
353                    && self.backup_acknowledged() >= self.activation_tail))
354    }
355
356    fn require_normal(&self) -> Result<(), ReplicationError> {
357        if self.status != Status::Normal {
358            return Err(ReplicationError::NotNormal);
359        }
360        Ok(())
361    }
362
363    fn require_scope(&self, scope: Scope) -> Result<(), ReplicationError> {
364        self.require_normal()?;
365        if scope != self.configuration.scope() {
366            return Err(ReplicationError::ScopeMismatch);
367        }
368        Ok(())
369    }
370
371    fn require_primary(&self, from: NodeId) -> Result<(), ReplicationError> {
372        self.configuration.voter_index(from)?;
373        if from != self.configuration.primary(self.configuration.scope.view) {
374            return Err(ReplicationError::WrongRole);
375        }
376        Ok(())
377    }
378
379    fn prefix_at(&self, op: OpNumber) -> Result<Prefix, ReplicationError> {
380        if op == self.applied.op {
381            return Ok(self.applied);
382        }
383        if op == self.retained_base.op {
384            return Ok(self.retained_base);
385        }
386        if op < self.retained_base.op {
387            return Err(ReplicationError::HistoryUnavailable);
388        }
389        if op > self.accepted.op {
390            return Err(ReplicationError::HistoryGap);
391        }
392        if let Some(tail) = self.journal_tail {
393            if op == tail.op {
394                return Ok(tail);
395            }
396            if op == self.committed.op {
397                return Ok(self.committed);
398            }
399            // Intermediate prefixes need pinned journal evidence. Do not index
400            // the empty live ring or promote partial, unverified tail evidence.
401            return Err(ReplicationError::HistoryUnavailable);
402        }
403        // The first retained operation follows the retained base.
404        op.0.checked_sub(self.retained_base.op.0)
405            .and_then(|distance| distance.checked_sub(1))
406            .and_then(|index| usize::try_from(index).ok())
407            .and_then(|index| self.operations.get(index))
408            .map(|operation| operation.prefix)
409            .ok_or(ReplicationError::HistoryUnavailable)
410    }
411
412    fn reclaim(&mut self) -> Result<(), ReplicationError> {
413        let through = self.applied.op.min(self.stored_through());
414        if through <= self.retained_base.op {
415            return Ok(());
416        }
417        let prefix = self.prefix_at(through)?;
418        self.body_bytes -= crate::release_through(&mut self.operations, through);
419        self.retained_base = prefix;
420        Ok(())
421    }
422
423    fn advance_commit(&mut self) -> Result<(), ReplicationError> {
424        if self.status != Status::Normal {
425            return Ok(());
426        }
427        let available = self.journal.snapshot().durable;
428        let eligible = if self.is_primary() {
429            // Own matching evidence AND either distinct backup's matching evidence.
430            available.min(self.backup_acknowledged())
431        } else {
432            available.min(self.known_commit)
433        };
434        if eligible > self.committed.op {
435            self.committed = self.prefix_at(eligible)?;
436        }
437        Ok(())
438    }
439
440    fn conflict(&mut self) -> ReplicationError {
441        self.fault();
442        ReplicationError::ConflictingHistory
443    }
444}
ozzy_replication/normal.rs

ozzy_replication/
normal.rs
