ozzy_replication/wire/
recovery.rs

1//! Nonvoting, nonce-scoped recovery exchanges with optional checkpoint anchors.
2
3use ozzy_proto::{Envelope, Opcode};
4use ozzy_proto::{LinkSessionId, NodeId, RequestId};
5
6use super::{
7    COMMON_BYTES, ControlEncoding, Incoming, PeerBinding, ReadFields, Reader, WireError,
8    WireLimits, WriteFields, Writer, validate_prefix, validate_scope,
9};
10use crate::recovery::{CheckpointAnchor, RecoveryLog, RecoveryResponse};
11use crate::{JournalGeneration, Scope};
12
13const NONCE_BYTES: usize = COMMON_BYTES + 16;
14const REQUEST_BYTES: usize = NONCE_BYTES + 1;
15const BACKUP_BYTES: usize = NONCE_BYTES + 1;
16const PRIMARY_BYTES: usize = BACKUP_BYTES + 16 + 40 + 40 + 1;
17const ANCHOR_BYTES: usize = 40 + 40 + 32 + 32 + 8 + 4;
18
19/// Fresh recovery attempt carried by one independently correlated link exchange.
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21pub struct RecoveryRequest {
22    /// Trusted group/configuration and a view hint, never a promise or vote.
23    pub scope: Scope,
24    /// Outstanding request in the authenticated link session.
25    pub request_id: RequestId,
26    /// Globally fresh recovery attempt, stable across retries and link replacement.
27    pub nonce: RequestId,
28    /// The requester persisted its nonvoting marker. A restart's retained
29    /// boundary probe asks before it votes and sends false.
30    pub nonvoting: bool,
31}
32
33impl RecoveryRequest {
34    fn validate(self) -> Result<(), WireError> {
35        validate_exchange(self.scope, self.request_id, self.nonce)
36    }
37}
38
39fn validate_exchange(
40    scope: Scope,
41    request_id: RequestId,
42    nonce: RequestId,
43) -> Result<(), WireError> {
44    validate_scope(scope)?;
45    if request_id.as_bytes() == &[0; 16] || nonce.as_bytes() == &[0; 16] {
46        return Err(WireError::Correlation);
47    }
48    Ok(())
49}
50
51/// Correlated reply retaining the core's nonce-scoped normal-state evidence.
52#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53pub struct RecoveryState {
54    /// Echoed link exchange ID, independently checked against the current request.
55    pub request_id: RequestId,
56    /// Immutable response for this recovery nonce and normal view, not a vote.
57    pub response: RecoveryResponse,
58}
59
60impl RecoveryState {
61    /// Match both the outstanding link request and the logical recovery attempt.
62    ///
63    /// Call after authenticated peer/session decoding. The responder's actual
64    /// normal view can differ from the request hint; the recovery core selects
65    /// the highest observed view. Matching alone grants no voting authority.
66    pub fn validate_response(self, request: RecoveryRequest) -> Result<(), WireError> {
67        self.validate()?;
68        request.validate()?;
69        if self.request_id != request.request_id || self.response.nonce != request.nonce {
70            return Err(WireError::Correlation);
71        }
72        if self.response.scope
73            != (Scope {
74                view: self.response.scope.view,
75                ..request.scope
76            })
77        {
78            return Err(WireError::Scope);
79        }
80        Ok(())
81    }
82
83    fn validate(self) -> Result<(), WireError> {
84        validate_exchange(self.response.scope, self.request_id, self.response.nonce)?;
85        if let Some(log) = self.response.primary {
86            if let Some(anchor) = log.checkpoint {
87                anchor
88                    .validate(log.committed)
89                    .map_err(|_| WireError::History)?;
90            }
91            validate_prefix(log.accepted)?;
92            validate_prefix(log.committed)?;
93            if log.generation.0 == 0
94                || log.committed.op > log.accepted.op
95                || (log.committed.op == log.accepted.op && log.committed != log.accepted)
96            {
97                return Err(WireError::History);
98            }
99        }
100        Ok(())
101    }
102}
103
104/// Separate recovery family. These messages never supply normal durable votes.
105#[derive(Debug, Clone, Copy, PartialEq, Eq)]
106#[expect(
107    clippy::large_enum_variant,
108    reason = "bounded sans-I/O decoding returns inline evidence without allocation"
109)]
110pub enum RecoveryMessage {
111    /// Ask activated normal voters for fresh evidence under an existing membership.
112    Request(RecoveryRequest),
113    /// A normal responder's evidence, still subject to the recovery core's checks.
114    State(RecoveryState),
115}
116
117/// Encode `RECOVERY_STATE` into reusable storage without mutation on error.
118///
119/// Obtain the response from an activated normal core and retain it unchanged per
120/// nonce/view. Encoding proves neither that role nor retention of source bytes.
121pub fn encode_recovery_state(
122    sender: NodeId,
123    session: LinkSessionId,
124    state: RecoveryState,
125    output: &mut [u8],
126    limits: WireLimits,
127) -> Result<ControlEncoding, WireError> {
128    state.validate()?;
129    let metadata_bytes = match state.response.primary {
130        Some(log) if log.checkpoint.is_some() => PRIMARY_BYTES + ANCHOR_BYTES,
131        Some(_) => PRIMARY_BYTES,
132        None => BACKUP_BYTES,
133    };
134    let header = Envelope {
135        opcode: Opcode::RecoveryState,
136        response: true,
137        request_id: Some(state.request_id),
138        sender,
139        session: Some(session),
140    }
141    .encode_header(metadata_bytes, 0, limits.envelope)?;
142    if output.len() < metadata_bytes {
143        return Err(WireError::Capacity);
144    }
145    let mut writer = Writer::with_len(output, metadata_bytes)?;
146    writer.scope(state.response.scope, sender)?;
147    writer.bytes(state.response.nonce.as_bytes())?;
148    writer.bytes(&[u8::from(state.response.primary.is_some())])?;
149    if let Some(log) = state.response.primary {
150        writer.bytes(&log.generation.0.to_be_bytes())?;
151        writer.prefix(log.accepted)?;
152        writer.prefix(log.committed)?;
153        writer.bytes(&[u8::from(log.checkpoint.is_some())])?;
154        if let Some(anchor) = log.checkpoint {
155            encode_anchor(&mut writer, anchor)?;
156        }
157    }
158    Ok(ControlEncoding {
159        header,
160        metadata_bytes,
161    })
162}
163
164/// Encode full-WAL `RECOVERY` without allocations or output mutation on error.
165pub fn encode_recovery(
166    sender: NodeId,
167    session: LinkSessionId,
168    request: RecoveryRequest,
169    output: &mut [u8],
170    limits: WireLimits,
171) -> Result<ControlEncoding, WireError> {
172    request.validate()?;
173    let header = Envelope {
174        opcode: Opcode::Recovery,
175        response: false,
176        request_id: Some(request.request_id),
177        sender,
178        session: Some(session),
179    }
180    .encode_header(REQUEST_BYTES, 0, limits.envelope)?;
181    if output.len() < REQUEST_BYTES {
182        return Err(WireError::Capacity);
183    }
184    let mut writer = Writer::with_len(output, REQUEST_BYTES)?;
185    writer.scope(request.scope, sender)?;
186    writer.bytes(request.nonce.as_bytes())?;
187    writer.bytes(&[u8::from(request.nonvoting)])?;
188    Ok(ControlEncoding {
189        header,
190        metadata_bytes: REQUEST_BYTES,
191    })
192}
193
194pub(super) fn decode_request(
195    Incoming {
196        scope,
197        envelope,
198        mut reader,
199        payload,
200    }: Incoming<'_>,
201) -> Result<RecoveryRequest, WireError> {
202    if !payload.is_empty() {
203        return Err(WireError::Payload);
204    }
205    if envelope.response {
206        return Err(WireError::Correlation);
207    }
208    let request = RecoveryRequest {
209        scope,
210        request_id: envelope.request_id.ok_or(WireError::Correlation)?,
211        nonce: RequestId::from_bytes(reader.array()?),
212        nonvoting: match reader.array::<1>()? {
213            [0] => false,
214            [1] => true,
215            _ => return Err(WireError::History),
216        },
217    };
218    reader.finish()?;
219    request.validate()?;
220    Ok(request)
221}
222
223pub(super) fn decode_state(
224    Incoming {
225        scope,
226        envelope,
227        mut reader,
228        payload,
229    }: Incoming<'_>,
230    binding: PeerBinding,
231) -> Result<RecoveryState, WireError> {
232    if !payload.is_empty() {
233        return Err(WireError::Payload);
234    }
235    if !envelope.response {
236        return Err(WireError::Correlation);
237    }
238    let request_id = envelope.request_id.ok_or(WireError::Correlation)?;
239    let nonce = RequestId::from_bytes(reader.array()?);
240    let primary = match reader.array::<1>()? {
241        [0] => None,
242        [1] => {
243            let generation = JournalGeneration(u128::from_be_bytes(reader.array()?));
244            let accepted = reader.prefix()?;
245            let committed = reader.prefix()?;
246            let checkpoint = match reader.array::<1>()? {
247                [0] => None,
248                [1] => Some(decode_anchor(&mut reader)?),
249                _ => return Err(WireError::History),
250            };
251            Some(RecoveryLog {
252                generation,
253                accepted,
254                committed,
255                checkpoint,
256            })
257        }
258        _ => return Err(WireError::History),
259    };
260    reader.finish()?;
261    if primary.is_some() != (binding.peer == binding.configuration.primary(scope.view)) {
262        return Err(WireError::History);
263    }
264    let state = RecoveryState {
265        request_id,
266        response: RecoveryResponse {
267            scope,
268            nonce,
269            primary,
270        },
271    };
272    state.validate()?;
273    Ok(state)
274}
275
276fn encode_anchor(writer: &mut Writer<'_>, anchor: CheckpointAnchor) -> Result<(), WireError> {
277    writer.prefix(anchor.predecessor)?;
278    writer.prefix(anchor.position)?;
279    writer.bytes(anchor.schema.as_bytes())?;
280    writer.bytes(anchor.state_digest.as_bytes())?;
281    writer.bytes(&anchor.state_bytes.to_be_bytes())?;
282    writer.bytes(&anchor.chunk_bytes.to_be_bytes())?;
283    Ok(())
284}
285
286fn decode_anchor(reader: &mut Reader<'_>) -> Result<CheckpointAnchor, WireError> {
287    Ok(CheckpointAnchor {
288        predecessor: reader.prefix()?,
289        position: reader.prefix()?,
290        schema: crate::Digest::from_bytes(reader.array()?),
291        state_digest: crate::Digest::from_bytes(reader.array()?),
292        state_bytes: u64::from_be_bytes(reader.array()?),
293        chunk_bytes: u32::from_be_bytes(reader.array()?),
294    })
295}
ozzy_replication/wire/recovery.rs

ozzy_replication/wire/
recovery.rs
