ozzy_replication/driver/
exit_view.rs1use super::{Action, DriverError, ReplicaDriver, Role};
4use crate::wire::Control;
5use crate::{Scope, VoterIndex, VoterSet};
6use ozzy_proto::NodeId;
7use std::time::Duration;
8
9impl ReplicaDriver {
10 pub(super) fn receive_exit(
11 &mut self,
12 voter: VoterIndex,
13 scope: Scope,
14 ) -> Result<(), DriverError> {
15 self.require_healthy()?;
16 if scope.view < self.scope().view {
19 if self.promise_pending() && self.now >= self.exit_echo_at[voter] {
29 self.exit_echo_at[voter] = self.now.saturating_add(self.timing.retransmit);
30 let mut asked = match self.exit_echo {
31 Some((old, asked)) if old == scope => asked,
32 _ => VoterSet::EMPTY,
33 };
34 asked.insert(voter);
35 self.exit_echo = Some((scope, asked));
36 }
37 return Ok(());
38 }
39 if scope != self.scope() {
40 return Ok(());
41 }
42 if let Some(normal) = self.normal() {
43 let snapshot = normal.snapshot();
44 if !self.normal_timed_out(&snapshot) {
45 self.withdraw_exit();
46 }
47 }
48 self.exit_heard_at[voter] = self.now;
49 self.exit_votes.insert(voter);
50 self.advance_if_exit_quorum()
51 }
52
53 pub fn observe_recovering(&mut self, from: NodeId, now: Duration) -> Result<(), DriverError> {
58 self.observe_time(now)?;
59 self.require_healthy()?;
60 let voter = self.configuration.voter_index(from)?;
61 if voter != self.local
62 && voter == VoterIndex::primary(self.scope().view)
63 && matches!(self.role, Role::ViewChanging(_))
64 {
65 self.request_exit()?;
66 }
67 Ok(())
68 }
69
70 pub(super) fn request_exit(&mut self) -> Result<(), DriverError> {
71 self.exit_retry_at.get_or_insert(self.now);
72 self.exit_votes.insert(self.local);
73 self.advance_if_exit_quorum()
74 }
75
76 fn advance_if_exit_quorum(&mut self) -> Result<(), DriverError> {
77 for voter in self.local.others() {
81 let heard = self.exit_heard_at[voter];
82 if self.now >= heard.saturating_add(self.timing.primary_timeout) {
83 self.exit_votes.remove(voter);
84 }
85 }
86 if self.exit_votes.len() >= 2 {
87 if self
88 .normal()
89 .is_none_or(|normal| !normal.snapshot().ready_for_appends)
90 {
91 self.election_delay = self
92 .election_delay
93 .saturating_mul(2)
94 .min(self.timing.max_election_timeout);
95 }
96 self.next_view()?;
97 }
98 Ok(())
99 }
100
101 pub(super) fn withdraw_exit(&mut self) {
102 self.exit_votes.remove(self.local);
103 self.exit_retry_at = None;
104 }
105
106 pub(super) fn promise_pending(&self) -> bool {
107 matches!(
108 &self.role,
109 Role::ViewChanging(changing) if changing.promised_view() != changing.scope().view
110 )
111 }
112
113 pub(super) fn exit_echo_action(&mut self) -> Option<Action> {
116 let (scope, mut asked) = self.exit_echo.take()?;
117 let voter = asked.pop_first()?;
118 if !asked.is_empty() {
119 self.exit_echo = Some((scope, asked));
120 }
121 Some(Action::Send {
122 to: self.configuration.voter(voter),
123 message: Control::ExitView(scope),
124 })
125 }
126
127 pub(super) fn exit_action(&mut self) -> Option<Action> {
128 if self.now < self.exit_retry_at? {
129 return None;
130 }
131 self.exit_retry_at = Some(self.now.saturating_add(self.timing.retransmit));
132 Some(Action::Broadcast(Control::ExitView(self.scope())))
133 }
134}