1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
//! Quorum-supported timeout suspicion, independent of durable election evidence.
use super::{Action, DriverError, ReplicaDriver, Role};
use crate::wire::Control;
use crate::{Scope, VoterIndex, VoterSet};
use ozzy_proto::NodeId;
use std::time::Duration;
impl ReplicaDriver {
pub(super) fn receive_exit(
&mut self,
voter: VoterIndex,
scope: Scope,
) -> Result<(), DriverError> {
self.require_healthy()?;
// Neither a distant wish nor a duplicate from another view is evidence
// that anyone has entered that view. START_VIEW_CHANGE supplies that fence.
if scope.view < self.scope().view {
// This voter left that view, possibly on the sender's own request
// plus a local timeout, before its suspicion went out. Without
// support the sender cannot leave until this voter's durable
// promise lets START_VIEW_CHANGE reach it. After that promise,
// election and normal traffic carry the newer view themselves.
//
// A peer that also left the view cannot tell this answer from a
// request. Answer only the sender, and each sender at most once
// per retransmission interval, so two such voters fall silent.
if self.promise_pending() && self.now >= self.exit_echo_at[voter] {
self.exit_echo_at[voter] = self.now.saturating_add(self.timing.retransmit);
let mut asked = match self.exit_echo {
Some((old, asked)) if old == scope => asked,
_ => VoterSet::EMPTY,
};
asked.insert(voter);
self.exit_echo = Some((scope, asked));
}
return Ok(());
}
if scope != self.scope() {
return Ok(());
}
if let Some(normal) = self.normal() {
let snapshot = normal.snapshot();
if !self.normal_timed_out(&snapshot) {
self.withdraw_exit();
}
}
self.exit_heard_at[voter] = self.now;
self.exit_votes.insert(voter);
self.advance_if_exit_quorum()
}
/// `from` asked for recovery as a nonvoting copy. When it is the primary of the
/// view this replica elects, it cannot install that view, so suspect the
/// view now instead of at the election deadline. Leaving the view still
/// needs a second voter's suspicion.
pub fn observe_recovering(&mut self, from: NodeId, now: Duration) -> Result<(), DriverError> {
self.observe_time(now)?;
self.require_healthy()?;
let voter = self.configuration.voter_index(from)?;
if voter != self.local
&& voter == VoterIndex::primary(self.scope().view)
&& matches!(self.role, Role::ViewChanging(_))
{
self.request_exit()?;
}
Ok(())
}
pub(super) fn request_exit(&mut self) -> Result<(), DriverError> {
self.exit_retry_at.get_or_insert(self.now);
self.exit_votes.insert(self.local);
self.advance_if_exit_quorum()
}
fn advance_if_exit_quorum(&mut self) -> Result<(), DriverError> {
// A suspicious peer repeats its request every retransmission
// interval. One that recovered goes quiet, so its request lapses
// instead of helping a later timeout end a healthy view.
for voter in self.local.others() {
let heard = self.exit_heard_at[voter];
if self.now >= heard.saturating_add(self.timing.primary_timeout) {
self.exit_votes.remove(voter);
}
}
if self.exit_votes.len() >= 2 {
if self
.normal()
.is_none_or(|normal| !normal.snapshot().ready_for_appends)
{
self.election_delay = self
.election_delay
.saturating_mul(2)
.min(self.timing.max_election_timeout);
}
self.next_view()?;
}
Ok(())
}
pub(super) fn withdraw_exit(&mut self) {
self.exit_votes.remove(self.local);
self.exit_retry_at = None;
}
pub(super) fn promise_pending(&self) -> bool {
matches!(
&self.role,
Role::ViewChanging(changing) if changing.promised_view() != changing.scope().view
)
}
/// Support one sender's exit from an older view per call. The sender
/// retransmits, so a lost answer is repeated after that interval.
pub(super) fn exit_echo_action(&mut self) -> Option<Action> {
let (scope, mut asked) = self.exit_echo.take()?;
let voter = asked.pop_first()?;
if !asked.is_empty() {
self.exit_echo = Some((scope, asked));
}
Some(Action::Send {
to: self.configuration.voter(voter),
message: Control::ExitView(scope),
})
}
pub(super) fn exit_action(&mut self) -> Option<Action> {
if self.now < self.exit_retry_at? {
return None;
}
self.exit_retry_at = Some(self.now.saturating_add(self.timing.retransmit));
Some(Action::Broadcast(Control::ExitView(self.scope())))
}
}