ozzy_replication/driver/
poll.rs

1//! Time-driven polling: deadlines, heartbeats, commit announcements, and
2//! election sends, at most one action per poll.
3
4use super::{Action, DriverError, ReplicaDriver, Role};
5use crate::wire::Control;
6use crate::{ReplicaSnapshot, ReplicationError, Status, ViewChangeError, VoterIndex};
7use std::time::Duration;
8
9impl ReplicaDriver {
10    /// Advance deadlines and emit at most one action, without waiting for I/O.
11    /// Repeated polls at one instant do not create unbounded retransmission.
12    pub fn poll(&mut self, now: Duration) -> Result<Option<Action>, DriverError> {
13        self.observe_time(now)?;
14        match &self.role {
15            Role::Installing(installing) => {
16                installing.require_pending()?;
17                return self.poll_installing(now);
18            }
19            Role::ViewChanging(changing)
20                if changing.normal_snapshot().status == Status::Faulted =>
21            {
22                return Err(ViewChangeError::Faulted.into());
23            }
24            Role::Fenced(_) => return Err(ViewChangeError::Faulted.into()),
25            Role::Normal(_) | Role::ViewChanging(_) => {}
26        }
27        if let Some(action) = self.exit_echo_action() {
28            return Ok(Some(action));
29        }
30        if let Some(normal) = self.normal() {
31            let snapshot = normal.snapshot();
32            if snapshot.status != Status::Normal {
33                return Err(ReplicationError::NotNormal.into());
34            }
35            if !self.normal_timed_out(&snapshot) {
36                return self.poll_normal(&snapshot, now);
37            }
38            self.request_exit()?;
39            if self.normal().is_some() {
40                return Ok(self.exit_action());
41            }
42        } else if now >= self.election_deadline && !self.promise_pending() {
43            // A voter still waiting on its own promise has no evidence about
44            // the view. With stalled storage it would keep helping a peer's
45            // timeout end views the others are installing.
46            self.request_exit()?;
47        }
48        if let Some(action) = self.exit_action() {
49            return Ok(Some(action));
50        }
51        if now >= self.retry_at {
52            self.start_pending = true;
53            self.report_pending = true;
54            self.retry_at = now.saturating_add(self.timing.retransmit);
55        }
56        self.poll_election()
57    }
58
59    fn poll_installing(&mut self, now: Duration) -> Result<Option<Action>, DriverError> {
60        if let Some(action) = self.exit_echo_action() {
61            return Ok(Some(action));
62        }
63        if now >= self.election_deadline {
64            self.request_exit()?;
65        }
66        // Suspicion can travel, but publication must settle before a promise.
67        Ok(self.exit_action())
68    }
69
70    /// A healthy normal role withdraws suspicion; its primary announces progress.
71    fn poll_normal(
72        &mut self,
73        snapshot: &ReplicaSnapshot,
74        now: Duration,
75    ) -> Result<Option<Action>, DriverError> {
76        self.withdraw_exit();
77        if snapshot.accepted == snapshot.committed && snapshot.ready_for_appends {
78            self.progress_deadline = now.saturating_add(self.timing.primary_timeout);
79        }
80        if VoterIndex::primary(snapshot.scope.view) != self.local {
81            return Ok(None);
82        }
83        if self.commit_pending {
84            self.commit_pending = false;
85            return self.announcement().map(Some);
86        }
87        if let Some(action) = self.activation_action()? {
88            return Ok(Some(action));
89        }
90        if now < self.retry_at {
91            return Ok(None);
92        }
93        self.retry_at = now.saturating_add(self.timing.heartbeat);
94        if let Some(start) = self.require_normal()?.start_view()? {
95            self.commit_pending = true;
96            return Ok(Some(Action::Broadcast(Control::StartView(start))));
97        }
98        self.announcement().map(Some)
99    }
100
101    fn announcement(&self) -> Result<Action, DriverError> {
102        let commit = self.require_normal()?.announcement()?;
103        Ok(Action::Broadcast(Control::Commit(commit)))
104    }
105
106    /// Persist the view promise, then send the first- and second-phase messages.
107    fn poll_election(&mut self) -> Result<Option<Action>, DriverError> {
108        // A normal role returned or entered this election above.
109        let Role::ViewChanging(changing) = &mut self.role else {
110            return Ok(None);
111        };
112        if changing.promised_view() != changing.scope().view {
113            return match changing.begin_promise() {
114                Ok(ticket) => Ok(Some(Action::PersistPromise(ticket))),
115                Err(ViewChangeError::StoragePending | ViewChangeError::PromisePending) => Ok(None),
116                Err(error) => Err(error.into()),
117            };
118        }
119        if self.start_pending {
120            let message = changing.start_message()?;
121            self.start_pending = false;
122            return Ok(Some(Action::Broadcast(Control::StartViewChange(message))));
123        }
124        if self.report_pending {
125            match changing.report() {
126                Ok(report) => {
127                    self.report_pending = false;
128                    let primary = VoterIndex::primary(report.scope.view);
129                    if primary != self.local {
130                        return Ok(Some(Action::Send {
131                            to: self.configuration.voter(primary),
132                            message: Control::DoViewChange(report),
133                        }));
134                    }
135                }
136                Err(ViewChangeError::StartQuorumMissing) => {}
137                Err(error) => return Err(error.into()),
138            }
139        }
140        Ok(None)
141    }
142}
ozzy_replication/driver/poll.rs

ozzy_replication/driver/
poll.rs
