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