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 {
43            self.request_exit()?;
44        }
45        if let Some(action) = self.exit_action() {
46            return Ok(Some(action));
47        }
48        if now >= self.retry_at {
49            self.start_pending = true;
50            self.report_pending = true;
51            self.retry_at = now.saturating_add(self.timing.retransmit);
52        }
53        self.poll_election()
54    }
55
56    fn poll_installing(&mut self, now: Duration) -> Result<Option<Action>, DriverError> {
57        if let Some(action) = self.exit_echo_action() {
58            return Ok(Some(action));
59        }
60        if now >= self.election_deadline {
61            self.request_exit()?;
62        }
63        // Suspicion can travel, but publication must settle before a promise.
64        Ok(self.exit_action())
65    }
66
67    /// A healthy normal role withdraws suspicion; its primary announces progress.
68    fn poll_normal(
69        &mut self,
70        snapshot: &ReplicaSnapshot,
71        now: Duration,
72    ) -> Result<Option<Action>, DriverError> {
73        self.withdraw_exit();
74        if snapshot.accepted == snapshot.committed && snapshot.ready_for_appends {
75            self.progress_deadline = now.saturating_add(self.timing.primary_timeout);
76        }
77        if VoterIndex::primary(snapshot.scope.view) != self.local {
78            return Ok(None);
79        }
80        if self.commit_pending {
81            self.commit_pending = false;
82            return self.announcement().map(Some);
83        }
84        if let Some(action) = self.activation_action()? {
85            return Ok(Some(action));
86        }
87        if now < self.retry_at {
88            return Ok(None);
89        }
90        self.retry_at = now.saturating_add(self.timing.heartbeat);
91        if let Some(start) = self.require_normal()?.start_view()? {
92            self.commit_pending = true;
93            return Ok(Some(Action::Broadcast(Control::StartView(start))));
94        }
95        self.announcement().map(Some)
96    }
97
98    fn announcement(&self) -> Result<Action, DriverError> {
99        let commit = self.require_normal()?.announcement()?;
100        Ok(Action::Broadcast(Control::Commit(commit)))
101    }
102
103    /// Persist the view promise, then send the first- and second-phase messages.
104    fn poll_election(&mut self) -> Result<Option<Action>, DriverError> {
105        // A normal role returned or entered this election above.
106        let Role::ViewChanging(changing) = &mut self.role else {
107            return Ok(None);
108        };
109        if changing.promised_view() != changing.scope().view {
110            return match changing.begin_promise() {
111                Ok(ticket) => Ok(Some(Action::PersistPromise(ticket))),
112                Err(ViewChangeError::StoragePending | ViewChangeError::PromisePending) => Ok(None),
113                Err(error) => Err(error.into()),
114            };
115        }
116        if self.start_pending {
117            let message = changing.start_message()?;
118            self.start_pending = false;
119            return Ok(Some(Action::Broadcast(Control::StartViewChange(message))));
120        }
121        if self.report_pending {
122            match changing.report() {
123                Ok(report) => {
124                    self.report_pending = false;
125                    let primary = VoterIndex::primary(report.scope.view);
126                    if primary != self.local {
127                        return Ok(Some(Action::Send {
128                            to: self.configuration.voter(primary),
129                            message: Control::DoViewChange(report),
130                        }));
131                    }
132                }
133                Err(ViewChangeError::StartQuorumMissing) => {}
134                Err(error) => return Err(error.into()),
135            }
136        }
137        Ok(None)
138    }
139}
ozzy_replication/driver/poll.rs

ozzy_replication/driver/
poll.rs
