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

ozzy_replication/wire/
recovery.rs
