liminal_server/server/participant/
transport.rs

1use liminal::protocol::Frame;
2use liminal_protocol::wire::{
3    AuthenticationState, ClientRequest, CodecError, FRAME_MAX, InboundGateContext,
4    InboundGateError, NegotiatedParticipantCapability, PARTICIPANT_FRAME_TYPE,
5    ParticipantCapabilityState, ParticipantFrame, ParticipantTransportRejected, ReceiverDirection,
6    ServerPush, ServerValue, TransportRejectionReason, ValidatedFrameLimit, encode, encoded_len,
7    gate_inbound,
8};
9
10/// Reserved connection-capability bit for `participant-v1`.
11///
12/// Production advertises this bit only after a complete semantic and durable
13/// participant service is installed. The transport gate can still decode and
14/// reject participant frames while the bit remains unadvertised.
15pub const PARTICIPANT_CAPABILITY_BIT: u32 = 1;
16
17/// Participant capability negotiated on one authenticated connection.
18#[derive(Clone, Copy, Debug, PartialEq, Eq)]
19pub struct ParticipantSession {
20    capability: ParticipantCapabilityState,
21}
22
23/// Rejects an oversized participant frame from its generic header before the
24/// connection buffers the declared body.
25///
26/// The shared gate owns both the negotiated/pre-capability limit and the exact
27/// rejection payload. An incomplete frame within the limit returns `None` and
28/// remains in the ordinary incremental decoder.
29#[must_use]
30pub fn preflight_generic_bytes(
31    input: &[u8],
32    authenticated: bool,
33    session: ParticipantSession,
34) -> Option<ParticipantTransportRejected> {
35    if input.first().copied() != Some(PARTICIPANT_FRAME_TYPE) || input.len() < 10 {
36        return None;
37    }
38    match gate_inbound(input, session.gate_context(authenticated)) {
39        Err(InboundGateError::ParticipantRejected(rejection))
40            if matches!(
41                rejection.reason,
42                TransportRejectionReason::FrameTooLarge { .. }
43            ) =>
44        {
45            Some(rejection)
46        }
47        Ok(_)
48        | Err(
49            InboundGateError::NotParticipantFrame { .. } | InboundGateError::ParticipantRejected(_),
50        ) => None,
51    }
52}
53
54/// Normalizes the operator's raw configured `WF` to the generic frame ceiling.
55///
56/// # Errors
57///
58/// Returns [`CodecError::InvalidFrameLimit`] when the configured value is below
59/// the shared protocol's minimum complete-frame size. Values above the generic
60/// `u32` payload ceiling are clamped to [`FRAME_MAX`] exactly as the participant
61/// contract defines.
62pub fn normalize_configured_frame_limit(
63    configured_wf: u64,
64) -> Result<ValidatedFrameLimit, CodecError> {
65    ValidatedFrameLimit::new(configured_wf.min(FRAME_MAX))
66}
67
68impl Default for ParticipantSession {
69    fn default() -> Self {
70        Self {
71            capability: ParticipantCapabilityState::Missing,
72        }
73    }
74}
75
76impl ParticipantSession {
77    /// Stores the server's supported v1 capability and normalized configured
78    /// complete-frame limit after a successful handshake.
79    pub const fn negotiate_v1(&mut self, limit: ValidatedFrameLimit) {
80        self.capability =
81            ParticipantCapabilityState::Negotiated(NegotiatedParticipantCapability::v1(limit));
82    }
83
84    const fn gate_context(self, authenticated: bool) -> InboundGateContext {
85        InboundGateContext {
86            receiver: ReceiverDirection::Server,
87            authentication: if authenticated {
88                AuthenticationState::Authenticated
89            } else {
90                AuthenticationState::Unauthenticated
91            },
92            participant_capability: self.capability,
93        }
94    }
95}
96
97/// Result of classifying one preserved generic frame through the shared gate.
98#[derive(Clone, Debug, PartialEq, Eq)]
99pub enum ParticipantIngress {
100    /// The generic frame is not assigned to participant traffic.
101    NotParticipant,
102    /// A structurally valid and authorized client request.
103    Request(ClientRequest),
104    /// Exact pre-semantic rejection selected by `liminal-protocol`.
105    Rejected(ParticipantTransportRejected),
106    /// A locally fabricated generic frame could not be represented canonically.
107    InvalidGenericFrame,
108}
109
110/// Applies the shared participant gate to one generic frame preserved by
111/// `liminal`'s forward-compatible decoder.
112#[must_use]
113pub fn gate_generic_frame(
114    frame: &Frame,
115    authenticated: bool,
116    session: ParticipantSession,
117) -> ParticipantIngress {
118    let Frame::Unknown {
119        type_id,
120        flags,
121        stream_id,
122        payload,
123    } = frame
124    else {
125        return ParticipantIngress::NotParticipant;
126    };
127    if *type_id != PARTICIPANT_FRAME_TYPE {
128        return ParticipantIngress::NotParticipant;
129    }
130    let Ok(payload_length) = u32::try_from(payload.len()) else {
131        return ParticipantIngress::InvalidGenericFrame;
132    };
133    let Some(capacity) = 10_usize.checked_add(payload.len()) else {
134        return ParticipantIngress::InvalidGenericFrame;
135    };
136    let mut complete = Vec::with_capacity(capacity);
137    complete.push(*type_id);
138    complete.push(*flags);
139    complete.extend_from_slice(&stream_id.to_be_bytes());
140    complete.extend_from_slice(&payload_length.to_be_bytes());
141    complete.extend_from_slice(payload);
142
143    match gate_inbound(&complete, session.gate_context(authenticated)) {
144        Ok(ParticipantFrame::ClientRequest(request)) => ParticipantIngress::Request(request),
145        Ok(ParticipantFrame::ServerValue(_) | ParticipantFrame::ServerPush(_)) => {
146            ParticipantIngress::InvalidGenericFrame
147        }
148        Err(InboundGateError::NotParticipantFrame { .. }) => ParticipantIngress::NotParticipant,
149        Err(InboundGateError::ParticipantRejected(rejection)) => {
150            ParticipantIngress::Rejected(rejection)
151        }
152    }
153}
154
155/// Encodes one crate-produced semantic value into the existing generic frame.
156///
157/// # Errors
158///
159/// Returns the shared codec error if the typed value cannot be encoded.
160pub fn encode_server_value(value: ServerValue) -> Result<Frame, CodecError> {
161    let participant = ParticipantFrame::ServerValue(value);
162    encode_server_frame(&participant)
163}
164
165/// Encodes one crate-produced server push into the participant generic frame on
166/// the protocol-owned generic stream zero.
167///
168/// # Errors
169///
170/// Returns the shared codec error if the typed push cannot be encoded.
171pub fn encode_server_push(push: ServerPush) -> Result<Frame, CodecError> {
172    let participant = ParticipantFrame::ServerPush(push);
173    encode_server_frame(&participant)
174}
175
176/// Applies the one canonical participant codec to either server-to-client arm,
177/// then preserves its already encoded participant payload in the generic frame.
178fn encode_server_frame(participant: &ParticipantFrame) -> Result<Frame, CodecError> {
179    let needed = encoded_len(participant)?;
180    let mut complete = vec![0_u8; needed];
181    let written = encode(participant, &mut complete)?;
182    complete.truncate(written);
183
184    let payload = complete
185        .get(10..)
186        .ok_or(CodecError::LengthOverflow)?
187        .to_vec();
188    Ok(Frame::Unknown {
189        type_id: PARTICIPANT_FRAME_TYPE,
190        flags: 0,
191        stream_id: 0,
192        payload,
193    })
194}
liminal_server/server/participant/transport.rs

liminal_server/server/participant/
transport.rs
