ozzy_replication/wire/
transfer.rs

1//! Correlated, generation-bound canonical history transfer. No implicit voting.
2
3use ozzy_proto::{Envelope, EnvelopeLimits, Opcode};
4use ozzy_proto::{LinkSessionId, NodeId, RequestId};
5
6use super::prepare::{DecodedBatch, decode_batch, metadata_length, validate_batch, write_batch};
7use super::{
8    COMMON_BYTES, ControlEncoding, Operation, Operations, PeerBinding, PrepareEncoding, Reader,
9    WireError, WireLimits, Writer, validate_prefix, validate_scope,
10};
11use crate::{JournalGeneration, LogSource, Prefix, Scope};
12
13const SOURCE_BYTES: usize = 16 + 16 + 40;
14pub(super) const FETCH_BYTES: usize = COMMON_BYTES + SOURCE_BYTES + 40 + 4 + 4;
15pub(super) const OPS_FIXED_BYTES: usize = COMMON_BYTES + SOURCE_BYTES + 40 + 4;
16
17/// One bounded request into a previously advertised immutable history.
18///
19/// The serving adapter must check that it owns and still pins this exact source.
20/// A request does not create a pin, prove history, or authorize role changes.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub struct FetchOps {
23    /// Active transfer view and complete group configuration.
24    pub scope: Scope,
25    /// Unique outstanding exchange in the established link session.
26    pub request_id: RequestId,
27    /// Exact configured voter, frozen writer incarnation, and accepted tail.
28    pub source: LogSource,
29    /// Last verified prefix before the requested consecutive operations.
30    pub predecessor: Prefix,
31    /// Upper bound on complete operations in the response, never zero.
32    pub max_operations: u32,
33    /// Upper bound on canonical body bytes in the response, never zero.
34    pub max_body_bytes: u32,
35}
36
37impl FetchOps {
38    fn validate(self) -> Result<(), WireError> {
39        validate_scope(self.scope)?;
40        validate_source(self.source)?;
41        validate_prefix(self.predecessor)?;
42        if self.predecessor.op >= self.source.accepted.op {
43            return Err(WireError::History);
44        }
45        if self.max_operations == 0 || self.max_body_bytes == 0 {
46            return Err(WireError::Limit);
47        }
48        Ok(())
49    }
50}
51
52/// Borrowed, hash-validated response from one frozen source generation.
53///
54/// Intermediate chunks prove only their own chain from the predecessor. The
55/// full selected lineage is verified only when that chain reaches its tail hash.
56/// No transfer response supplies normal voting or application commit authority.
57#[derive(Debug, Clone, Copy, PartialEq, Eq)]
58pub struct OpsBatch<'a> {
59    request_id: RequestId,
60    source: LogSource,
61    batch: DecodedBatch<'a>,
62}
63
64impl<'a> OpsBatch<'a> {
65    /// Active transfer view and immutable group configuration.
66    pub const fn scope(self) -> Scope {
67        self.batch.scope
68    }
69    /// Correlation echoed from `FETCH_OPS`.
70    pub const fn request_id(self) -> RequestId {
71        self.request_id
72    }
73    /// Exact frozen source named by the response, still requiring request matching.
74    pub const fn source(self) -> LogSource {
75        self.source
76    }
77    /// Exact prefix immediately before the returned operations.
78    pub const fn predecessor(self) -> Prefix {
79        self.batch.predecessor
80    }
81    /// End of this nonempty chunk, not necessarily the selected source's tail.
82    pub const fn end(self) -> Prefix {
83        self.batch.accepted
84    }
85    /// Iterate immutable canonical bodies without allocating or rehashing.
86    pub fn operations(self) -> Operations<'a> {
87        self.batch.operations()
88    }
89
90    /// Match the live outstanding request after authenticated session decoding.
91    ///
92    /// The adapter invalidates pending requests when their session or role ends.
93    /// Success here does not advance the installation cursor or imply disk sync.
94    pub fn validate_response(self, request: FetchOps) -> Result<(), WireError> {
95        request.validate()?;
96        if self.request_id != request.request_id
97            || self.scope() != request.scope
98            || self.source != request.source
99            || self.predecessor() != request.predecessor
100            || self.operations().len() > request.max_operations as usize
101            || self.batch.payload.len() > request.max_body_bytes as usize
102        {
103            return Err(WireError::Transfer);
104        }
105        Ok(())
106    }
107}
108
109/// Encode complete canonical operations from the exact pinned request source.
110///
111/// Reuses verified body digests and caller buffers. Honors both request bounds
112/// and local wire limits; errors leave metadata and payload output unchanged.
113/// The serving adapter independently verifies the source pin and role lifetime.
114pub fn encode_ops(
115    sender: NodeId,
116    session: LinkSessionId,
117    request: FetchOps,
118    operations: &[Operation<'_>],
119    metadata: &mut [u8],
120    payload: &mut [u8],
121    limits: WireLimits,
122) -> Result<PrepareEncoding, WireError> {
123    request.validate()?;
124    if sender != request.source.voter {
125        return Err(WireError::Peer);
126    }
127    let limits = WireLimits {
128        max_operations: limits.max_operations.min(request.max_operations as usize),
129        envelope: EnvelopeLimits {
130            max_metadata_bytes: limits.envelope.max_metadata_bytes,
131            max_payload_bytes: limits
132                .envelope
133                .max_payload_bytes
134                .min(request.max_body_bytes as usize),
135        },
136    };
137    let metadata_bytes = metadata_length(operations.len(), OPS_FIXED_BYTES, limits)?;
138    let payload_bytes =
139        validate_batch(request.scope, request.predecessor, None, operations, limits)?;
140    validate_tail(
141        request.source,
142        operations.last().expect("nonempty validated").prefix(),
143    )?;
144    let header = Envelope {
145        opcode: Opcode::Ops,
146        response: true,
147        request_id: Some(request.request_id),
148        sender,
149        session: Some(session),
150    }
151    .encode_header(metadata_bytes, payload_bytes, limits.envelope)?;
152    if metadata.len() < metadata_bytes || payload.len() < payload_bytes {
153        return Err(WireError::Capacity);
154    }
155    let mut writer = Writer::new(&mut metadata[..metadata_bytes]);
156    writer.scope(request.scope, sender);
157    write_source(&mut writer, request.source);
158    writer.prefix(request.predecessor);
159    write_batch(&mut writer, Some(payload), operations);
160    Ok(PrepareEncoding {
161        header,
162        metadata_bytes,
163        payload_bytes,
164    })
165}
166
167pub(super) fn decode_ops<'a>(
168    scope: Scope,
169    envelope: Envelope,
170    mut reader: Reader<'a>,
171    payload: &'a [u8],
172    limits: WireLimits,
173) -> Result<OpsBatch<'a>, WireError> {
174    let request_id = envelope.request_id.ok_or(WireError::Correlation)?;
175    if !envelope.response {
176        return Err(WireError::Correlation);
177    }
178    let source = read_source(&mut reader)?;
179    if source.voter != envelope.sender {
180        return Err(WireError::Peer);
181    }
182    let predecessor = reader.prefix()?;
183    let batch = decode_batch(scope, predecessor, None, reader, payload, limits)?;
184    validate_tail(source, batch.accepted)?;
185    Ok(OpsBatch {
186        request_id,
187        source,
188        batch,
189    })
190}
191
192pub(super) fn validate_routing(
193    request: FetchOps,
194    envelope: Envelope,
195    mut reader: Reader<'_>,
196    payload_bytes: usize,
197) -> Result<(), WireError> {
198    request.validate()?;
199    if !envelope.response || envelope.request_id != Some(request.request_id) {
200        return Err(WireError::Correlation);
201    }
202    if envelope.sender != request.source.voter
203        || read_source(&mut reader)? != request.source
204        || reader.prefix()? != request.predecessor
205    {
206        return Err(WireError::Transfer);
207    }
208    let operations = reader.u32()?;
209    if operations == 0
210        || operations > request.max_operations
211        || payload_bytes > request.max_body_bytes as usize
212    {
213        return Err(WireError::Limit);
214    }
215    Ok(())
216}
217
218fn validate_tail(source: LogSource, end: Prefix) -> Result<(), WireError> {
219    if end.op > source.accepted.op || (end.op == source.accepted.op && end != source.accepted) {
220        return Err(WireError::History);
221    }
222    Ok(())
223}
224
225/// Encode a correlated request without allocating or modifying output on error.
226///
227/// Response limits are upper bounds, not allocation instructions. The source
228/// also bounds each read/response by its own storage, scheduling, and wire limits.
229pub fn encode_fetch(
230    sender: NodeId,
231    session: LinkSessionId,
232    request: FetchOps,
233    metadata: &mut [u8],
234    limits: WireLimits,
235) -> Result<ControlEncoding, WireError> {
236    request.validate()?;
237    let header = Envelope {
238        opcode: Opcode::FetchOps,
239        response: false,
240        request_id: Some(request.request_id),
241        sender,
242        session: Some(session),
243    }
244    .encode_header(FETCH_BYTES, 0, limits.envelope)?;
245    if metadata.len() < FETCH_BYTES {
246        return Err(WireError::Capacity);
247    }
248    let mut writer = Writer::new(&mut metadata[..FETCH_BYTES]);
249    writer.scope(request.scope, sender);
250    write_source(&mut writer, request.source);
251    writer.prefix(request.predecessor);
252    writer.bytes(&request.max_operations.to_be_bytes());
253    writer.bytes(&request.max_body_bytes.to_be_bytes());
254    Ok(ControlEncoding {
255        header,
256        metadata_bytes: FETCH_BYTES,
257    })
258}
259
260pub(super) fn decode_fetch(
261    scope: Scope,
262    envelope: Envelope,
263    mut reader: Reader<'_>,
264    payload: &[u8],
265    binding: PeerBinding,
266) -> Result<FetchOps, WireError> {
267    let request_id = envelope.request_id.ok_or(WireError::Correlation)?;
268    if envelope.response {
269        return Err(WireError::Correlation);
270    }
271    if !payload.is_empty() {
272        return Err(WireError::Payload);
273    }
274    let source = read_source(&mut reader)?;
275    binding.configuration.voter_index(source.voter)?;
276    let request = FetchOps {
277        scope,
278        request_id,
279        source,
280        predecessor: reader.prefix()?,
281        max_operations: reader.u32()?,
282        max_body_bytes: reader.u32()?,
283    };
284    reader.finish()?;
285    request.validate()?;
286    Ok(request)
287}
288
289pub(super) fn validate_source(source: LogSource) -> Result<(), WireError> {
290    validate_prefix(source.accepted)?;
291    if source.generation.0 == 0 || source.voter.as_bytes() == &[0; 16] {
292        return Err(WireError::History);
293    }
294    Ok(())
295}
296
297pub(super) fn write_source(writer: &mut Writer<'_>, source: LogSource) {
298    writer.bytes(source.voter.as_bytes());
299    writer.bytes(&source.generation.0.to_be_bytes());
300    writer.prefix(source.accepted);
301}
302
303pub(super) fn read_source(reader: &mut Reader<'_>) -> Result<LogSource, WireError> {
304    let source = LogSource {
305        voter: NodeId::from_bytes(reader.bytes()?),
306        generation: JournalGeneration(u128::from_be_bytes(reader.bytes()?)),
307        accepted: reader.prefix()?,
308    };
309    validate_source(source)?;
310    Ok(source)
311}
ozzy_replication/wire/transfer.rs

ozzy_replication/wire/
transfer.rs
