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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
//! Quorum-supported timeout suspicion, independent of durable election evidence.
use super::{Action, DriverError, ReplicaDriver, Role};
use crate::wire::Control;
use crate::{Scope, StartViewChange, 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 traffic carries the newer view while it lasts.
//
// 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.now < self.exit_echo_at[voter] {
return Ok(());
}
if self.promise_pending() {
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));
} else if matches!(self.role, Role::Normal(_) | Role::Installing(_)) {
// Election traffic stops once a view is normal, and normal
// traffic carries the view only from its primary. A sender
// that cannot hear that primary would never learn of the
// view, and with the primary gone the two survivors could
// never form another. Answer with this view's durable fence
// instead, which brings the sender into it.
self.exit_echo_at[voter] = self.now.saturating_add(self.timing.retransmit);
self.fence_answers.insert(voter);
}
return Ok(());
}
if scope.view > self.scope().view {
// A request to leave a newer view says nothing about this one,
// but it shows this view is stale. An idle primary has no timeout
// of its own, and only the newer view's primary would tell it of
// that view. With that primary gone, the sender could never form
// a view with it. Ask the sender to leave this view instead: its
// answer is the newer view's fence, which brings this voter in.
if matches!(self.role, Role::Normal(_)) && self.now >= self.exit_echo_at[voter] {
self.exit_echo_at[voter] = self.now.saturating_add(self.timing.retransmit);
self.stale_answers.insert(voter);
}
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
)
}
/// Answer one sender's request to leave another view per call: support
/// for an older view's exit, that view's fence, or this stale view's exit.
/// The sender retransmits, so a lost answer is repeated after that
/// interval.
pub(super) fn exit_echo_action(&mut self) -> Option<Action> {
if let Some((scope, mut asked)) = self.exit_echo.take()
&& let Some(voter) = asked.pop_first()
{
if !asked.is_empty() {
self.exit_echo = Some((scope, asked));
}
return Some(Action::Send {
to: self.configuration.voter(voter),
message: Control::ExitView(scope),
});
}
if let Some(voter) = self.fence_answers.pop_first() {
return Some(Action::Send {
to: self.configuration.voter(voter),
message: Control::StartViewChange(StartViewChange {
scope: self.scope(),
}),
});
}
let voter = self.stale_answers.pop_first()?;
Some(Action::Send {
to: self.configuration.voter(voter),
message: Control::ExitView(self.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())))
}
}