ozzy_replication/wire/
checkpoint.rs

1//! Nonvoting checkpoint bytes over bounded, source/nonce-bound PEER exchanges.
2
3use super::{
4    COMMON_BYTES, ControlEncoding, Incoming, PeerBinding, ReadFields, WireError, WireLimits,
5    WriteFields, Writer,
6    transfer::{read_source, validate_source, write_source},
7    validate_scope,
8};
9use crate::{LogSource, Scope};
10use ozzy_proto::{Envelope, LinkSessionId, NodeId, Opcode, RequestId};
11
12const METADATA_BYTES: usize = COMMON_BYTES + 16 + 72 + 8 + 4;
13
14/// One bounded byte range of a previously pinned recovery checkpoint.
15#[derive(Clone, Copy, Debug, Eq, PartialEq)]
16pub struct CheckpointRequest {
17    /// Exact configuration/view of the recovery attempt.
18    pub scope: Scope,
19    /// Outstanding exchange, checked again against the current link session.
20    pub request_id: RequestId,
21    /// Fresh recovery nonce, independent of transient link identities.
22    pub nonce: RequestId,
23    /// Exact donor incarnation and accepted tail retained by its pin.
24    pub source: LogSource,
25    /// Next logical checkpoint-state byte, beginning at zero.
26    pub offset: u64,
27    /// Positive maximum returned bytes; a source chunk boundary may return less.
28    pub max_bytes: u32,
29}
30
31impl CheckpointRequest {
32    fn validate(self, limits: WireLimits) -> Result<(), WireError> {
33        validate_scope(self.scope)?;
34        validate_source(self.source)?;
35        if self.request_id.as_bytes() == &[0; 16] || self.nonce.as_bytes() == &[0; 16] {
36            return Err(WireError::Correlation);
37        }
38        if self.max_bytes == 0 || self.max_bytes as usize > limits.envelope.max_payload_bytes {
39            return Err(WireError::Limit);
40        }
41        Ok(())
42    }
43}
44
45/// Source-bound checkpoint exchange, supplying no normal quorum evidence.
46#[derive(Clone, Copy, Debug, Eq, PartialEq)]
47pub enum CheckpointMessage<'a> {
48    /// Request bytes only after the donor has offered its immutable descriptor.
49    Request(CheckpointRequest),
50    /// Echo the exact request alongside one positive bounded state slice.
51    Chunk {
52        /// Exact nonce/source-bound request being answered.
53        request: CheckpointRequest,
54        /// Positive state slice within the requested limit.
55        bytes: &'a [u8],
56    },
57}
58
59/// Encode checkpoint request/chunk metadata without copying state bytes.
60pub fn encode_checkpoint(
61    sender: NodeId,
62    session: LinkSessionId,
63    message: CheckpointMessage<'_>,
64    output: &mut [u8],
65    limits: WireLimits,
66) -> Result<ControlEncoding, WireError> {
67    let (request, payload_bytes, response) = match message {
68        CheckpointMessage::Request(request) => (request, 0, false),
69        CheckpointMessage::Chunk { request, bytes } => {
70            if bytes.is_empty()
71                || bytes.len() > request.max_bytes as usize
72                || request.source.voter != sender
73            {
74                return Err(WireError::History);
75            }
76            (request, bytes.len(), true)
77        }
78    };
79    request.validate(limits)?;
80    let header = Envelope {
81        opcode: if response {
82            Opcode::SnapshotChunk
83        } else {
84            Opcode::SnapshotBegin
85        },
86        response,
87        request_id: Some(request.request_id),
88        sender,
89        session: Some(session),
90    }
91    .encode_header(METADATA_BYTES, payload_bytes, limits.envelope)?;
92    if output.len() < METADATA_BYTES {
93        return Err(WireError::Capacity);
94    }
95    let mut writer = Writer::with_len(output, METADATA_BYTES)?;
96    writer.scope(request.scope, sender)?;
97    writer.bytes(request.nonce.as_bytes())?;
98    write_source(&mut writer, request.source)?;
99    writer.bytes(&request.offset.to_be_bytes())?;
100    writer.bytes(&request.max_bytes.to_be_bytes())?;
101    Ok(ControlEncoding {
102        header,
103        metadata_bytes: METADATA_BYTES,
104    })
105}
106
107pub(super) fn decode(
108    Incoming {
109        scope,
110        envelope,
111        mut reader,
112        payload,
113    }: Incoming<'_>,
114    binding: PeerBinding,
115    limits: WireLimits,
116) -> Result<CheckpointMessage<'_>, WireError> {
117    let request = CheckpointRequest {
118        scope,
119        request_id: envelope.request_id.ok_or(WireError::Correlation)?,
120        nonce: RequestId::from_bytes(reader.array()?),
121        source: read_source(&mut reader)?,
122        offset: u64::from_be_bytes(reader.array()?),
123        max_bytes: reader.u32()?,
124    };
125    reader.finish()?;
126    request.validate(limits)?;
127    binding.configuration.voter_index(request.source.voter)?;
128    match envelope.opcode {
129        Opcode::SnapshotBegin if !envelope.response && payload.is_empty() => {
130            Ok(CheckpointMessage::Request(request))
131        }
132        Opcode::SnapshotChunk
133            if envelope.response
134                && request.source.voter == binding.peer
135                && !payload.is_empty()
136                && payload.len() <= request.max_bytes as usize =>
137        {
138            Ok(CheckpointMessage::Chunk {
139                request,
140                bytes: payload,
141            })
142        }
143        _ => Err(WireError::Correlation),
144    }
145}
146
147/// Exact old receiver or history request being withdrawn. Neither is authority.
148#[derive(Debug, Clone, Copy, PartialEq, Eq)]
149pub enum HistoryFence {
150    /// Volatile normal receive incarnation.
151    Receive(crate::flow::ReceiveEpoch),
152    /// Outstanding election history request.
153    Fetch(RequestId),
154}
155
156/// Primary hint to withdraw a lagging receiver before fresh quorum recovery.
157/// This carries no vote, commit proof, or checkpoint installation authority.
158#[derive(Debug, Clone, Copy, PartialEq, Eq)]
159pub struct HistoryRetired {
160    /// Exact normal authority view.
161    pub scope: Scope,
162    /// Receiver incarnation whose independently checked history is unavailable.
163    pub fence: HistoryFence,
164    /// Last original operation preceding all required retained operations.
165    /// For a fetch, the first predecessor the source serves from: its
166    /// retained boundary, when the request predecessor is retired or not in
167    /// the source. Only a fetch fence may name genesis.
168    pub before: crate::Prefix,
169}
170
171/// Encode a receiver-fenced withdrawal hint on the control plane.
172pub fn encode_history_retired(
173    sender: NodeId,
174    session: LinkSessionId,
175    notice: HistoryRetired,
176    output: &mut [u8],
177    limits: WireLimits,
178) -> Result<ControlEncoding, WireError> {
179    validate_scope(notice.scope)?;
180    super::validate_prefix(notice.before)?;
181    if (notice.before.op.0 == 0 && matches!(notice.fence, HistoryFence::Receive(_)))
182        || matches!(notice.fence, HistoryFence::Fetch(id) if id.as_bytes() == &[0; 16])
183    {
184        return Err(WireError::History);
185    }
186    let length = COMMON_BYTES + 1 + 16 + 40;
187    let header = Envelope {
188        opcode: Opcode::HistoryRetired,
189        response: false,
190        request_id: None,
191        sender,
192        session: Some(session),
193    }
194    .encode_header(length, 0, limits.envelope)?;
195    if output.len() < length {
196        return Err(WireError::Capacity);
197    }
198    let mut writer = Writer::with_len(output, length)?;
199    writer.scope(notice.scope, sender)?;
200    match notice.fence {
201        HistoryFence::Receive(epoch) => {
202            writer.bytes(&[0])?;
203            writer.bytes(&epoch.get().to_be_bytes())?;
204        }
205        HistoryFence::Fetch(id) => {
206            writer.bytes(&[1])?;
207            writer.bytes(id.as_bytes())?;
208        }
209    }
210    writer.prefix(notice.before)?;
211    Ok(ControlEncoding {
212        header,
213        metadata_bytes: length,
214    })
215}
216
217pub(super) fn decode_retired(
218    Incoming {
219        scope,
220        envelope,
221        mut reader,
222        payload,
223    }: Incoming<'_>,
224) -> Result<HistoryRetired, WireError> {
225    if envelope.response || envelope.request_id.is_some() || !payload.is_empty() {
226        return Err(WireError::Correlation);
227    }
228    let tag: [u8; 1] = reader.array()?;
229    let id: [u8; 16] = reader.array()?;
230    let fence = match tag[0] {
231        0 => HistoryFence::Receive(crate::flow::ReceiveEpoch::new(u128::from_be_bytes(id))?),
232        1 if id != [0; 16] => HistoryFence::Fetch(RequestId::from_bytes(id)),
233        _ => return Err(WireError::Correlation),
234    };
235    let before = reader.prefix()?;
236    reader.finish()?;
237    super::validate_prefix(before)?;
238    if before.op.0 == 0 && matches!(fence, HistoryFence::Receive(_)) {
239        return Err(WireError::History);
240    }
241    Ok(HistoryRetired {
242        scope,
243        fence,
244        before,
245    })
246}
ozzy_replication/wire/checkpoint.rs

ozzy_replication/wire/
checkpoint.rs
