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

ozzy_replication/wire/
transfer.rs
