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

ozzy_replication/driver/
storage.rs
