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, PeerBinding, Reader, WireError, WireLimits, Writer,
8    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 = if state.response.primary.is_some() {
123        PRIMARY_BYTES
124            + usize::from(
125                state
126                    .response
127                    .primary
128                    .is_some_and(|log| log.checkpoint.is_some()),
129            ) * ANCHOR_BYTES
130    } else {
131        BACKUP_BYTES
132    };
133    let header = Envelope {
134        opcode: Opcode::RecoveryState,
135        response: true,
136        request_id: Some(state.request_id),
137        sender,
138        session: Some(session),
139    }
140    .encode_header(metadata_bytes, 0, limits.envelope)?;
141    if output.len() < metadata_bytes {
142        return Err(WireError::Capacity);
143    }
144    let mut writer = Writer::new(&mut output[..metadata_bytes]);
145    writer.scope(state.response.scope, sender);
146    writer.bytes(state.response.nonce.as_bytes());
147    writer.bytes(&[u8::from(state.response.primary.is_some())]);
148    if let Some(log) = state.response.primary {
149        writer.bytes(&log.generation.0.to_be_bytes());
150        writer.prefix(log.accepted);
151        writer.prefix(log.committed);
152        writer.bytes(&[u8::from(log.checkpoint.is_some())]);
153        if let Some(anchor) = log.checkpoint {
154            encode_anchor(&mut writer, anchor);
155        }
156    }
157    Ok(ControlEncoding {
158        header,
159        metadata_bytes,
160    })
161}
162
163/// Encode full-WAL `RECOVERY` without allocations or output mutation on error.
164pub fn encode_recovery(
165    sender: NodeId,
166    session: LinkSessionId,
167    request: RecoveryRequest,
168    output: &mut [u8],
169    limits: WireLimits,
170) -> Result<ControlEncoding, WireError> {
171    request.validate()?;
172    let header = Envelope {
173        opcode: Opcode::Recovery,
174        response: false,
175        request_id: Some(request.request_id),
176        sender,
177        session: Some(session),
178    }
179    .encode_header(REQUEST_BYTES, 0, limits.envelope)?;
180    if output.len() < REQUEST_BYTES {
181        return Err(WireError::Capacity);
182    }
183    let mut writer = Writer::new(&mut output[..REQUEST_BYTES]);
184    writer.scope(request.scope, sender);
185    writer.bytes(request.nonce.as_bytes());
186    Ok(ControlEncoding {
187        header,
188        metadata_bytes: REQUEST_BYTES,
189    })
190}
191
192pub(super) fn decode_request(
193    scope: Scope,
194    envelope: Envelope,
195    mut reader: Reader<'_>,
196    payload: &[u8],
197) -> Result<RecoveryRequest, WireError> {
198    if !payload.is_empty() {
199        return Err(WireError::Payload);
200    }
201    if envelope.response {
202        return Err(WireError::Correlation);
203    }
204    let request = RecoveryRequest {
205        scope,
206        request_id: envelope.request_id.ok_or(WireError::Correlation)?,
207        nonce: RequestId::from_bytes(reader.bytes()?),
208    };
209    reader.finish()?;
210    request.validate()?;
211    Ok(request)
212}
213
214pub(super) fn decode_state(
215    scope: Scope,
216    envelope: Envelope,
217    mut reader: Reader<'_>,
218    payload: &[u8],
219    binding: PeerBinding,
220) -> Result<RecoveryState, WireError> {
221    if !payload.is_empty() {
222        return Err(WireError::Payload);
223    }
224    if !envelope.response {
225        return Err(WireError::Correlation);
226    }
227    let request_id = envelope.request_id.ok_or(WireError::Correlation)?;
228    let nonce = RequestId::from_bytes(reader.bytes()?);
229    let primary = match reader.bytes::<1>()? {
230        [0] => None,
231        [1] => {
232            let generation = JournalGeneration(u128::from_be_bytes(reader.bytes()?));
233            let accepted = reader.prefix()?;
234            let committed = reader.prefix()?;
235            let checkpoint = match reader.bytes::<1>()? {
236                [0] => None,
237                [1] => Some(decode_anchor(&mut reader)?),
238                _ => return Err(WireError::History),
239            };
240            Some(RecoveryLog {
241                generation,
242                accepted,
243                committed,
244                checkpoint,
245            })
246        }
247        _ => return Err(WireError::History),
248    };
249    reader.finish()?;
250    if primary.is_some() != (binding.peer == binding.configuration.primary(scope.view)) {
251        return Err(WireError::History);
252    }
253    let state = RecoveryState {
254        request_id,
255        response: RecoveryResponse {
256            scope,
257            nonce,
258            primary,
259        },
260    };
261    state.validate()?;
262    Ok(state)
263}
264
265fn encode_anchor(writer: &mut Writer<'_>, anchor: CheckpointAnchor) {
266    writer.prefix(anchor.predecessor);
267    writer.prefix(anchor.position);
268    writer.bytes(anchor.schema.as_bytes());
269    writer.bytes(anchor.state_digest.as_bytes());
270    writer.bytes(&anchor.state_bytes.to_be_bytes());
271    writer.bytes(&anchor.chunk_bytes.to_be_bytes());
272}
273
274fn decode_anchor(reader: &mut Reader<'_>) -> Result<CheckpointAnchor, WireError> {
275    Ok(CheckpointAnchor {
276        predecessor: reader.prefix()?,
277        position: reader.prefix()?,
278        schema: crate::Digest::from_bytes(reader.bytes()?),
279        state_digest: crate::Digest::from_bytes(reader.bytes()?),
280        state_bytes: u64::from_be_bytes(reader.bytes()?),
281        chunk_bytes: u32::from_be_bytes(reader.bytes()?),
282    })
283}
ozzy_replication/wire/recovery.rs

ozzy_replication/wire/
recovery.rs
