ozzy_replication/wire/
admission.rs

1//! Actor-selected receive fences checked before spending dispatch capacity.
2
3use ozzy_proto::{EnvelopeLimits, Opcode, Packet};
4
5use super::{FetchOps, Reader, WireError};
6use crate::flow::{Channel, FlowError, ReceiveEpoch};
7
8/// Exact purpose of one actor-backed replica receive reservation.
9/// This is not authentication, history validation, or replication authority.
10#[derive(Clone, Copy, Debug, Eq, PartialEq)]
11pub enum ReceiveFence {
12    /// Normal data for the receiver's current group/configuration/view/epoch.
13    Normal(Channel),
14    /// Response to one exact outstanding selected-history request.
15    History(FetchOps),
16}
17
18impl ReceiveFence {
19    /// Group selected by the owning actor, never by incoming payload bytes.
20    pub const fn scope(self) -> crate::Scope {
21        match self {
22            Self::Normal(channel) => channel.scope,
23            Self::History(request) => request.scope,
24        }
25    }
26
27    /// Inspect only bounded protocol metadata. A mismatch must not consume the
28    /// fresh reservation. The actor must still decode/hash bodies and validate
29    /// membership, history, application state, and its current authority.
30    pub fn validate_routing(
31        self,
32        packet: Packet<'_>,
33        limits: EnvelopeLimits,
34    ) -> Result<(), WireError> {
35        packet
36            .envelope
37            .validate_frames(packet.metadata.len(), packet.payload.len(), limits)?;
38        let mut reader = Reader::new(packet.metadata);
39        if reader.routing_scope(packet.envelope.sender)? != self.scope() {
40            return Err(WireError::Scope);
41        }
42        match self {
43            Self::Normal(channel) => {
44                if packet.envelope.opcode != Opcode::PrepareFlow {
45                    return Err(WireError::UnsupportedCommand);
46                }
47                if packet.envelope.response || packet.envelope.request_id.is_some() {
48                    return Err(WireError::Correlation);
49                }
50                let epoch = ReceiveEpoch::new(u128::from_be_bytes(reader.bytes()?))?;
51                if epoch != channel.epoch {
52                    return Err(FlowError::Channel.into());
53                }
54                Ok(())
55            }
56            Self::History(request) => {
57                if packet.envelope.opcode != Opcode::Ops {
58                    return Err(WireError::UnsupportedCommand);
59                }
60                super::transfer::validate_routing(
61                    request,
62                    packet.envelope,
63                    reader,
64                    packet.payload.len(),
65                )
66            }
67        }
68    }
69}
ozzy_replication/wire/admission.rs

ozzy_replication/wire/
admission.rs
