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
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
use serde::Serialize;
use thiserror::Error;
/// Listener handoff strategy used during a zero-downtime upgrade.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum UpgradeStrategy {
/// A newly spawned process receives the already-bound listener.
InheritedSocket,
/// Old and new processes overlap on the same address through reuse-port style binding.
ReusePort,
}
impl UpgradeStrategy {
/// Return the default strategy for the current platform.
pub fn platform_default() -> Self {
if cfg!(windows) {
Self::ReusePort
} else {
Self::InheritedSocket
}
}
}
/// Upgrade phases visible to readiness checks and diagnostics.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum UpgradePhase {
/// Handoff has been prepared, but the replacement is not ready yet.
Prepared,
/// Replacement process is ready to accept traffic.
NewReady,
/// Old process stopped accepting and is draining active work.
OldDraining,
/// Upgrade finished without dropped in-flight work.
Complete,
}
/// Operator-provided upgrade plan.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UpgradePlan {
generation: u64,
member_id: String,
strategy: UpgradeStrategy,
}
impl UpgradePlan {
/// Create a plan for the next process generation.
pub fn new(generation: u64, member_id: impl Into<String>) -> Self {
Self {
generation,
member_id: member_id.into(),
strategy: UpgradeStrategy::platform_default(),
}
}
/// Override the listener handoff strategy.
pub fn with_strategy(mut self, strategy: UpgradeStrategy) -> Self {
self.strategy = strategy;
self
}
/// Validate and prepare the handoff.
pub fn prepare(self) -> Result<GracefulUpgrade, UpgradeError> {
if self.generation == 0 {
return Err(UpgradeError::InvalidGeneration);
}
if self.member_id.trim().is_empty() {
return Err(UpgradeError::MissingMemberId);
}
Ok(GracefulUpgrade {
plan: self,
phase: UpgradePhase::Prepared,
old_accepting: true,
new_ready: false,
in_flight: 0,
completed: 0,
})
}
}
/// Deterministic zero-downtime handoff model.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GracefulUpgrade {
plan: UpgradePlan,
phase: UpgradePhase,
old_accepting: bool,
new_ready: bool,
in_flight: usize,
completed: usize,
}
impl GracefulUpgrade {
/// Mark the replacement process ready before the old process drains.
pub fn mark_new_ready(&mut self) {
self.new_ready = true;
self.phase = UpgradePhase::NewReady;
}
/// Stop the old process from accepting new work and start drain.
pub fn start_draining_old(&mut self) -> Result<(), UpgradeError> {
if !self.new_ready {
return Err(UpgradeError::ReplacementNotReady);
}
self.old_accepting = false;
self.phase = UpgradePhase::OldDraining;
Ok(())
}
/// Record work accepted by the old process while it is still serving.
pub fn record_request(&mut self) -> bool {
if !self.old_accepting {
return false;
}
self.in_flight = self.in_flight.saturating_add(1);
true
}
/// Mark one in-flight request as completed.
pub fn finish_request(&mut self) {
if self.in_flight > 0 {
self.in_flight -= 1;
self.completed = self.completed.saturating_add(1);
}
}
/// Finish upgrade after all in-flight work is drained.
pub fn complete(mut self) -> Result<UpgradeReport, UpgradeError> {
if !self.new_ready {
return Err(UpgradeError::ReplacementNotReady);
}
if self.in_flight > 0 {
return Err(UpgradeError::InFlightRequestsRemaining(self.in_flight));
}
self.phase = UpgradePhase::Complete;
Ok(UpgradeReport {
generation: self.plan.generation,
member_id: self.plan.member_id,
strategy: self.plan.strategy,
phase: self.phase,
completed_requests: self.completed,
dropped_requests: 0,
})
}
/// Return current phase.
pub fn phase(&self) -> UpgradePhase {
self.phase
}
/// Return whether old and replacement process keep the same member identity.
pub fn membership_stable(&self) -> bool {
!self.plan.member_id.trim().is_empty()
}
/// Return active work still attached to the old process.
pub fn in_flight(&self) -> usize {
self.in_flight
}
/// Return whether the old process still accepts new traffic.
pub fn old_accepting(&self) -> bool {
self.old_accepting
}
}
/// Successful upgrade result.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct UpgradeReport {
/// Process generation that completed.
pub generation: u64,
/// Stable cluster member identity.
pub member_id: String,
/// Listener handoff strategy used.
pub strategy: UpgradeStrategy,
/// Final phase.
pub phase: UpgradePhase,
/// Requests completed while old process drained.
pub completed_requests: usize,
/// Requests dropped by the handoff.
pub dropped_requests: usize,
}
/// Fail-loud upgrade errors.
#[derive(Debug, Error, PartialEq, Eq)]
pub enum UpgradeError {
/// Generation zero is reserved for uninitialized runtimes.
#[error("upgrade generation must be greater than zero")]
InvalidGeneration,
/// Cluster member identity must remain explicit across handoff.
#[error("upgrade requires a non-empty member id")]
MissingMemberId,
/// The old process cannot drain before the replacement is ready.
#[error("replacement process is not ready")]
ReplacementNotReady,
/// Upgrade cannot complete while old process still owns work.
#[error("{0} in-flight request(s) remain during upgrade")]
InFlightRequestsRemaining(usize),
}