ozzy_replication/
recovery.rs

1//! Fresh-quorum admission for a fixed voter that lost its previous state.
2//!
3//! Recovery never contributes its own vote. Only normal replicas answer; the
4//! highest observed view's primary supplies the complete accepted history.
5//! Descriptors are not proof of retained bytes, completed I/O, or application
6//! recovery. The adapter authenticates peers and pins the exact selected source.
7
8mod transfer;
9
10use ozzy_proto::{NodeId, RequestId};
11
12use crate::{
13    Configuration, JournalGeneration, LogSource, NormalReplica, PipelineLimits, Prefix,
14    ReplicationError, Scope, Status,
15};
16
17/// Primary's immutable full-log snapshot for one recovery nonce.
18#[derive(Debug, Clone, Copy, PartialEq, Eq)]
19pub struct RecoveryLog {
20    /// Pinned source writer incarnation, never mixed with a replacement writer.
21    pub generation: JournalGeneration,
22    /// Complete accepted prefix, including unsynced or uncommitted prepares.
23    pub accepted: Prefix,
24    /// Known commit floor in this snapshot, not a truncation boundary.
25    pub committed: Prefix,
26    /// Canonical state and retained-chain anchor when older operations expired.
27    pub checkpoint: Option<CheckpointAnchor>,
28}
29
30/// Exact pinned checkpoint and the predecessor of required retained operations.
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub struct CheckpointAnchor {
33    /// Hash anchor before the first retained operation, possibly before state.
34    pub predecessor: Prefix,
35    /// Original confirmed operation prefix represented by canonical state.
36    pub position: Prefix,
37    /// Exact canonical state schema digest.
38    pub schema: crate::Digest,
39    /// Integrity digest over all checkpoint state bytes.
40    pub state_digest: crate::Digest,
41    /// Bounded logical state size, independently checked by the storage adapter.
42    pub state_bytes: u64,
43    /// Maximum bytes in each immutable source chunk.
44    pub chunk_bytes: u32,
45}
46
47impl CheckpointAnchor {
48    /// Validate coordinates; bytes and canonical state require separate checks.
49    pub fn validate(self, committed: Prefix) -> Result<(), RecoveryError> {
50        if !crate::view_change::valid_prefix(self.predecessor)
51            || !crate::view_change::valid_prefix(self.position)
52            || self.position.op.0 == 0
53            || self.predecessor.op > self.position.op
54            || self.position.op > committed.op
55            || (self.predecessor.op == self.position.op && self.predecessor != self.position)
56            || (self.position.op == committed.op && self.position != committed)
57            || self.schema == crate::Digest::ZERO
58            || self.state_digest == crate::Digest::ZERO
59            || self.state_bytes == 0
60            || self.chunk_bytes == 0
61        {
62            return Err(RecoveryError::InvalidResponse);
63        }
64        Ok(())
65    }
66}
67
68/// Fresh response from an authenticated normal replica, not a durable vote.
69#[derive(Debug, Clone, Copy, PartialEq, Eq)]
70pub struct RecoveryResponse {
71    /// Exact configuration and responder's normal view.
72    pub scope: Scope,
73    /// Globally fresh request nonce echoed unchanged across retransmissions.
74    pub nonce: RequestId,
75    /// Present exactly when the responder is primary in `scope.view`.
76    pub primary: Option<RecoveryLog>,
77}
78
79impl NormalReplica {
80    /// Capture a recovery response without adding I/O to normal commit.
81    ///
82    /// The adapter pins this exact accepted prefix, including in-flight writes,
83    /// before offering its bytes. An unsent candidate may be refreshed before
84    /// pin admission. Once pinning starts, retain/retransmit that exact response
85    /// for this nonce. A durable-prefix-only snapshot is unsafe: an earlier ACK
86    /// from the lost voter may still commit an unsynced accepted operation.
87    pub fn recovery_response(&self, nonce: RequestId) -> Result<RecoveryResponse, RecoveryError> {
88        let snapshot = self.snapshot();
89        if snapshot.status != Status::Normal || !snapshot.ready_for_appends {
90            return Err(RecoveryError::NotReady);
91        }
92        if nonce.as_bytes() == &[0; 16] {
93            return Err(RecoveryError::InvalidIdentity);
94        }
95        let (configuration, local) = self.driver_identity();
96        Ok(RecoveryResponse {
97            scope: snapshot.scope,
98            nonce,
99            primary: (local == configuration.primary(snapshot.scope.view)).then_some(RecoveryLog {
100                generation: snapshot.journal.generation,
101                accepted: snapshot.accepted,
102                committed: snapshot.committed,
103                checkpoint: None,
104            }),
105        })
106    }
107}
108
109/// Quorum-authorized full-history transfer. No voting or storage-completion proof.
110#[derive(Debug, Clone, Copy, PartialEq, Eq)]
111pub struct RecoveryTicket {
112    checkpoint: Option<CheckpointAnchor>,
113    local: NodeId,
114    generation: JournalGeneration,
115    nonce: RequestId,
116    scope: Scope,
117    source: LogSource,
118    committed: Prefix,
119    voters: u8,
120}
121
122impl RecoveryTicket {
123    /// Pinned state/retained-chain source authorized by the recovery quorum.
124    pub const fn checkpoint(self) -> Option<CheckpointAnchor> {
125        self.checkpoint
126    }
127
128    /// Existing fixed voter being recovered, never a new membership entry.
129    pub const fn local(self) -> NodeId {
130        self.local
131    }
132    /// Fresh replacement writer incarnation; binds every completion.
133    pub const fn generation(self) -> JournalGeneration {
134        self.generation
135    }
136    /// Recovery attempt identity, independent of transient transport sessions.
137    pub const fn nonce(self) -> RequestId {
138        self.nonce
139    }
140    /// Highest observed view, with a matching normal primary response.
141    pub const fn scope(self) -> Scope {
142        self.scope
143    }
144    /// Complete immutable accepted lineage to transfer, not only committed data.
145    pub const fn source(self) -> LogSource {
146        self.source
147    }
148    /// Application state must be rebuilt through this exact snapshot commit floor.
149    pub const fn committed(self) -> Prefix {
150        self.committed
151    }
152    /// Configured response participants. Never contains the recovering voter.
153    pub const fn voter_mask(self) -> u8 {
154        self.voters
155    }
156}
157
158/// Bounded nonvoting recovery role. Owns no normal or view-change voting state.
159#[derive(Debug)]
160pub struct Recovery {
161    configuration: Configuration,
162    local: usize,
163    generation: JournalGeneration,
164    nonce: RequestId,
165    view: u64,
166    limits: PipelineLimits,
167    responses: [Option<RecoveryResponse>; 3],
168    pending: Option<Transfer>,
169    completed: bool,
170    faulted: bool,
171}
172
173#[derive(Debug, Clone, Copy)]
174struct Transfer {
175    ticket: RecoveryTicket,
176    through: Prefix,
177    checkpoint_ready: bool,
178}
179
180impl Recovery {
181    /// Start explicit lost-state recovery under an externally trusted configuration.
182    /// `nonce` and `generation` must be globally fresh, including after process
183    /// restart. The caller must not have another live incarnation of this voter.
184    pub fn new(
185        configuration: Configuration,
186        local: NodeId,
187        generation: JournalGeneration,
188        nonce: RequestId,
189        limits: PipelineLimits,
190    ) -> Result<Self, RecoveryError> {
191        let local = configuration.voter_index(local)?;
192        if generation.0 == 0 || nonce.as_bytes() == &[0; 16] {
193            return Err(RecoveryError::InvalidIdentity);
194        }
195        if limits.max_operations == 0 || limits.max_body_bytes == 0 {
196            return Err(ReplicationError::InvalidLimits.into());
197        }
198        Ok(Self {
199            configuration,
200            local,
201            generation,
202            nonce,
203            view: 0,
204            limits,
205            responses: [None; 3],
206            pending: None,
207            completed: false,
208            faulted: false,
209        })
210    }
211
212    /// Record one authenticated normal response. Duplicate replies count once.
213    pub fn receive(
214        &mut self,
215        from: NodeId,
216        response: RecoveryResponse,
217    ) -> Result<(), RecoveryError> {
218        self.require_active()?;
219        let voter = self.configuration.voter_index(from)?;
220        if voter == self.local {
221            return Err(RecoveryError::SelfResponse);
222        }
223        if response.nonce != self.nonce {
224            return Err(RecoveryError::StaleNonce);
225        }
226        if response.scope
227            != (Scope {
228                view: response.scope.view,
229                ..self.configuration.scope()
230            })
231        {
232            return Err(ReplicationError::ScopeMismatch.into());
233        }
234        if response.primary.is_some() != (from == self.configuration.primary(response.scope.view)) {
235            return Err(RecoveryError::InvalidResponse);
236        }
237        if let Some(log) = response.primary
238            && (log.generation.0 == 0
239                || !crate::view_change::valid_prefix(log.accepted)
240                || !crate::view_change::valid_prefix(log.committed)
241                || log.committed.op > log.accepted.op
242                || (log.committed.op == log.accepted.op && log.committed != log.accepted))
243        {
244            return Err(RecoveryError::InvalidResponse);
245        }
246        if let Some(anchor) = response.primary.and_then(|log| log.checkpoint) {
247            anchor.validate(response.primary.expect("checkpoint primary").committed)?;
248        }
249        if let Some(previous) = self.responses[voter] {
250            if response.scope.view < previous.scope.view {
251                return Ok(());
252            }
253            if response.scope.view == previous.scope.view && response != previous {
254                self.faulted = true;
255                return Err(RecoveryError::ConflictingResponse);
256            }
257        }
258        self.view = self.view.max(response.scope.view);
259        self.responses[voter] = Some(response);
260        Ok(())
261    }
262
263    /// Observe configured peer authority outside the recovery response exchange.
264    /// A higher election/normal view supersedes pending recovery but grants no
265    /// vote. Drain/settle its worker action, then start a fresh nonce-scoped
266    /// recovery attempt. A timeout alone must not invent higher-view evidence.
267    pub fn observe_view(&mut self, from: NodeId, scope: Scope) -> Result<(), RecoveryError> {
268        self.require_active()?;
269        if self.configuration.voter_index(from)? == self.local {
270            return Err(RecoveryError::SelfResponse);
271        }
272        if scope
273            != (Scope {
274                view: scope.view,
275                ..self.configuration.scope()
276            })
277        {
278            return Err(ReplicationError::ScopeMismatch.into());
279        }
280        self.view = self.view.max(scope.view);
281        Ok(())
282    }
283
284    /// Freeze the highest-view primary's source after two distinct other responses.
285    /// Older-view normal responses with this fresh nonce may witness the quorum.
286    pub fn begin_transfer(&mut self) -> Result<RecoveryTicket, RecoveryError> {
287        self.require_active()?;
288        if self.pending.is_some() {
289            return Err(RecoveryError::TransferPending);
290        }
291        let voters = self
292            .responses
293            .iter()
294            .enumerate()
295            .fold(0_u8, |mask, (index, response)| {
296                mask | (u8::from(response.is_some()) << index)
297            });
298        if voters.count_ones() < 2 {
299            return Err(RecoveryError::QuorumMissing);
300        }
301        let primary = self.configuration.primary(self.view);
302        let response = self.responses[self.configuration.voter_index(primary)?]
303            .filter(|response| response.scope.view == self.view)
304            .ok_or(RecoveryError::PrimaryMissing)?;
305        let log = response.primary.ok_or(RecoveryError::PrimaryMissing)?;
306        if log.generation == self.generation {
307            return Err(RecoveryError::InvalidIdentity);
308        }
309        let ticket = RecoveryTicket {
310            checkpoint: log.checkpoint,
311            local: self.configuration.voters()[self.local],
312            generation: self.generation,
313            nonce: self.nonce,
314            scope: response.scope,
315            source: LogSource {
316                voter: primary,
317                generation: log.generation,
318                accepted: log.accepted,
319            },
320            committed: log.committed,
321            voters,
322        };
323        self.pending = Some(Transfer {
324            ticket,
325            through: log
326                .checkpoint
327                .map_or(Prefix::GENESIS, |anchor| anchor.predecessor),
328            checkpoint_ready: log.checkpoint.is_none(),
329        });
330        Ok(ticket)
331    }
332
333    fn require_active(&self) -> Result<(), RecoveryError> {
334        if self.faulted {
335            Err(RecoveryError::Faulted)
336        } else if self.completed {
337            Err(RecoveryError::Completed)
338        } else {
339            Ok(())
340        }
341    }
342}
343
344/// Rejected recovery evidence. No error grants a vote or replaces valid history.
345#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
346pub enum RecoveryError {
347    /// Existing configuration, membership, or bounds validation failed.
348    #[error(transparent)]
349    Replication(#[from] ReplicationError),
350    /// Nonces and replacement writer incarnations must be nonzero and fresh.
351    #[error("invalid recovery identity")]
352    InvalidIdentity,
353    /// Only an activated normal replica may respond to recovery.
354    #[error("replica is not ready to answer recovery")]
355    NotReady,
356    /// Lost state cannot contribute evidence about its own past votes.
357    #[error("recovering replica cannot count itself")]
358    SelfResponse,
359    /// Response belongs to another recovery attempt.
360    #[error("stale recovery nonce")]
361    StaleNonce,
362    /// Invalid role, source incarnation, or prefix coordinates.
363    #[error("invalid recovery response")]
364    InvalidResponse,
365    /// Same responder changed a frozen response within one view and nonce.
366    #[error("conflicting recovery response")]
367    ConflictingResponse,
368    /// Two distinct other normal replicas have not responded.
369    #[error("recovery quorum missing")]
370    QuorumMissing,
371    /// No response from the primary of the highest observed view.
372    #[error("recovery primary missing")]
373    PrimaryMissing,
374    /// Previous transfer must settle before another can be admitted.
375    #[error("recovery transfer already pending")]
376    TransferPending,
377    /// Callback or chunk belongs to another writer, nonce, source, or attempt.
378    #[error("stale recovery transfer")]
379    StaleTransfer,
380    /// A newer configured view superseded this transfer's recovery authority.
381    #[error("recovery view superseded")]
382    StaleView,
383    /// The entire selected full-WAL chain has not been validated.
384    #[error("recovery history missing")]
385    HistoryMissing,
386    /// Actual durable publication does not cover the selected accepted prefix.
387    #[error("recovery storage publication pending")]
388    StoragePending,
389    /// Canonical committed state has not been rebuilt at the selected commit floor.
390    #[error("recovery application validation pending")]
391    ApplicationPending,
392    /// Recovery evidence has already been consumed by a fenced restart handoff.
393    #[error("recovery already completed")]
394    Completed,
395    /// Conflicting evidence or uncertain publication requires a new recovery attempt.
396    #[error("recovery faulted")]
397    Faulted,
398}
ozzy_replication/recovery.rs

ozzy_replication/
recovery.rs
