ozzy_replication/driver/
storage.rs

1//! Local storage completions and failures: promise publication, writes,
2//! barriers, application, and uncertain I/O.
3
4use super::{DriverError, ReplicaDriver, Role};
5use crate::{
6    JournalGeneration, Prefix, PromiseTicket, ReplicationError, SyncTicket, ViewChangeError,
7    WriteTicket,
8};
9use std::time::Duration;
10
11impl ReplicaDriver {
12    /// Observe actual durable publication of the exact outstanding promise.
13    pub fn complete_promise(&mut self, ticket: PromiseTicket) -> Result<(), DriverError> {
14        self.changing()?.complete_promise(ticket)?;
15        Ok(())
16    }
17
18    /// Report an uncertain metadata publication. No election retry can reuse it.
19    pub fn fail_promise(&mut self, ticket: PromiseTicket) -> Result<(), DriverError> {
20        self.changing()?.fail_promise(ticket)?;
21        Ok(())
22    }
23
24    /// Settle actual complete writes, including writes admitted before a timeout.
25    /// Completion while changing view never restores normal voting or commit.
26    pub fn complete_write(&mut self, ticket: WriteTicket) -> Result<(), DriverError> {
27        match &mut self.role {
28            Role::Normal(normal) => normal.complete_write(ticket)?,
29            Role::ViewChanging(changing) => changing.complete_write(ticket)?,
30            Role::Installing(_) => return Err(ReplicationError::WrongRole.into()),
31            Role::Fenced(_) => return Err(ViewChangeError::Faulted.into()),
32        }
33        Ok(())
34    }
35
36    /// Install an exact `O_DSYNC` completion. The write is durable at once.
37    pub fn complete_durable_write(
38        &mut self,
39        ticket: WriteTicket,
40        now: Duration,
41    ) -> Result<(), DriverError> {
42        self.observe_time(now)?;
43        match &mut self.role {
44            Role::Normal(normal) => {
45                let previous = normal.snapshot();
46                normal.complete_durable_write(ticket)?;
47                let current = normal.snapshot();
48                self.observe_normal_progress(&previous, &current, now);
49            }
50            Role::ViewChanging(changing) => changing.normal.complete_durable_write(ticket)?,
51            Role::Installing(_) => return Err(ReplicationError::WrongRole.into()),
52            Role::Fenced(_) => return Err(ViewChangeError::Faulted.into()),
53        }
54        Ok(())
55    }
56
57    /// Complete deterministic ordered application through an established commit.
58    /// Application replay/identity-index I/O stays outside this driver.
59    pub fn apply_through(&mut self, through: Prefix) -> Result<(), DriverError> {
60        let Role::Normal(normal) = &mut self.role else {
61            return Err(ReplicationError::NotNormal.into());
62        };
63        let was_ready = normal.snapshot().ready_for_appends;
64        normal.apply_through(through)?;
65        if !was_ready && normal.snapshot().ready_for_appends {
66            self.progress_deadline = self.now.saturating_add(self.timing.primary_timeout);
67            self.election_delay = self.timing.election_timeout;
68        }
69        Ok(())
70    }
71
72    /// Fence on uncertain storage failure. A stale writer cannot fault a new one.
73    pub fn fail_io(&mut self, generation: JournalGeneration) -> Result<(), DriverError> {
74        match &mut self.role {
75            Role::Normal(normal) => normal.fail_io(generation)?,
76            Role::ViewChanging(changing) => changing.fail_io(generation)?,
77            Role::Installing(installing) => {
78                let ticket = installing.ticket();
79                if generation != ticket.generation() {
80                    return Err(ViewChangeError::StaleInstallation.into());
81                }
82                installing.fail(ticket)?;
83            }
84            Role::Fenced(_) => return Err(ViewChangeError::Faulted.into()),
85        }
86        Ok(())
87    }
88
89    /// Capture the written prefix for a real disk barrier, never a simulated ACK.
90    pub fn begin_sync(&self) -> Result<SyncTicket, DriverError> {
91        match &self.role {
92            Role::Normal(normal) => Ok(normal.begin_sync()?),
93            Role::ViewChanging(changing) => Ok(changing.begin_sync()?),
94            Role::Installing(_) => Err(ReplicationError::WrongRole.into()),
95            Role::Fenced(_) => Err(ViewChangeError::Faulted.into()),
96        }
97    }
98
99    /// Observe a completed barrier. Commit and persistence deadlines advance
100    /// independently; a late old-view completion only settles the frozen log.
101    pub fn complete_sync(&mut self, ticket: SyncTicket, now: Duration) -> Result<(), DriverError> {
102        self.observe_time(now)?;
103        match &mut self.role {
104            Role::Normal(normal) => {
105                let previous = normal.snapshot();
106                normal.complete_sync(ticket)?;
107                let current = normal.snapshot();
108                self.observe_normal_progress(&previous, &current, now);
109            }
110            Role::ViewChanging(changing) => changing.complete_sync(ticket)?,
111            Role::Installing(_) => return Err(ReplicationError::WrongRole.into()),
112            Role::Fenced(_) => return Err(ViewChangeError::Faulted.into()),
113        }
114        Ok(())
115    }
116}
ozzy_replication/driver/storage.rs

ozzy_replication/driver/
storage.rs
