liminal_server/
link_route_call.rs

1//! Calls carry the caller's original admitted wire request inside Noise.
2//! The far exchange executes it through its ordinary admitted connection path.
3
4use crate::link_handshake::{LinkTransport, NOISE_MAX_MESSAGE, NOISE_OVERHEAD};
5use liminal::protocol::{Frame, codec};
6use zeroize::Zeroizing;
7
8#[path = "link_route_call/client.rs"]
9pub mod client;
10#[path = "link_route_call/pending.rs"]
11pub mod pending;
12#[cfg(test)]
13#[path = "link_route_call_tests.rs"]
14mod tests;
15#[path = "link_route_call/wire.rs"]
16mod wire;
17
18/// A full far connection execution, owned until its exact call settles.
19pub trait CallExecution: Send + std::fmt::Debug {
20    /// Cancels this call's full connection and joins its owned reader.
21    ///
22    /// # Errors
23    /// The connection's cancellation or worker join failed.
24    fn finish(&mut self) -> Result<(), CallError>;
25}
26
27/// The receiving supervisor supplies its ordinary admitted connection path.
28pub trait CallExecutor: Send + Sync + std::fmt::Debug {
29    /// Actual retained execution owners, for capacity and leak observations.
30    fn retained(&self) -> usize;
31    /// Waits on execution retirement signals, with no clock or polling.
32    ///
33    /// # Errors
34    /// The retirement signal state was poisoned.
35    fn observe_empty(&self) -> Result<(), CallError>;
36    /// Starts only after the keyed link has decided its two crossing grants.
37    ///
38    /// # Errors
39    /// Full connection admission, wire validation or owned reader startup failed.
40    fn start(
41        &self,
42        request: CallRequest,
43        completed: Box<dyn FnOnce(Result<CallReply, CallError>) + Send>,
44    ) -> Result<Box<dyn CallExecution>, CallError>;
45}
46
47/// A named crossing refusal; no request or credential bytes enter its text.
48#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
49pub enum CallError {
50    /// The complete call cannot fit its configured bound or the Noise ceiling.
51    #[error("link_call_too_large")]
52    Bound,
53    /// No additional call is admitted until an existing reservation settles.
54    #[error("link_call_capacity")]
55    Capacity,
56    /// The call envelope is malformed or has unexpected framing.
57    #[error("link_call_invalid")]
58    Invalid,
59    /// A reply did not match this exact call and subject.
60    #[error("link_call_correlation_mismatch")]
61    Correlation,
62    /// The keyed peer disappeared before its answer.
63    #[error("link_call_peer_gone")]
64    PeerGone,
65    /// The local caller cancelled this exact pending operation.
66    #[error("link_call_cancelled")]
67    Cancelled,
68    /// The caller's already configured pending-reply deadline elapsed.
69    #[error("link_call_deadline")]
70    Deadline,
71    /// State was poisoned or a response delivery could not be completed.
72    #[error("link_call_unavailable")]
73    Unavailable,
74    /// No live server on either end advertises this subject.
75    #[error("no_server")]
76    NoServer,
77    /// A machine no longer holds its crossing right when the answer is released.
78    #[error("link_call_grant_refused")]
79    GrantRefused,
80}
81
82/// A zeroized sequence of Connect, `ConversationOpen` and `ConversationMessage`.
83/// Private fields prevent the receiving adapter from accepting arbitrary acts.
84pub struct CallRequest {
85    id: u64,
86    subject: String,
87    bytes: Zeroizing<Vec<u8>>,
88}
89
90impl std::fmt::Debug for CallRequest {
91    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
92        out.debug_struct("CallRequest")
93            .field("id", &self.id)
94            .field("bytes", &self.bytes.len())
95            .finish_non_exhaustive()
96    }
97}
98
99impl CallRequest {
100    /// Builds the call from its caller's original credential and request frames.
101    ///
102    /// # Errors
103    /// Wrong frame sequence, correlations/subject, non-Lys credential or oversize call.
104    pub fn new(
105        id: u64,
106        subject: String,
107        frames: &[Frame],
108        bound: usize,
109    ) -> Result<Self, CallError> {
110        let bytes = wire::encode_frames(frames, bound)?;
111        Self::decode(id, subject, bytes, bound)
112    }
113
114    fn decode(
115        id: u64,
116        subject: String,
117        bytes: Zeroizing<Vec<u8>>,
118        bound: usize,
119    ) -> Result<Self, CallError> {
120        if id == 0 || bytes.len() > bound {
121            return Err(CallError::Invalid);
122        }
123        crate::auth_lys::SubjectBinding::new("wire".into(), subject.clone(), subject.clone())
124            .map_err(|_| CallError::Invalid)?;
125        let mut frames = wire::request_frames(&bytes, &subject)?;
126        for frame in &mut frames {
127            if let Frame::Connect { auth_token, .. } = frame {
128                zeroize::Zeroize::zeroize(auth_token);
129            }
130        }
131        drop(frames);
132        Ok(Self { id, subject, bytes })
133    }
134    /// The monotonic link-local operation id, never reused on this transport.
135    #[must_use]
136    pub const fn id(&self) -> u64 {
137        self.id
138    }
139    /// The concrete placed service subject.
140    #[must_use]
141    pub fn subject(&self) -> &str {
142        &self.subject
143    }
144    /// The checked original frames, for the full far admission path.
145    ///
146    /// # Errors
147    /// The stored wire bytes could not be decoded.
148    pub fn frames(&self) -> Result<Vec<Frame>, CallError> {
149        wire::request_frames(&self.bytes, &self.subject)
150    }
151    /// The checked wire image, borrowed only while it is sent to the far door.
152    #[must_use]
153    pub fn wire_bytes(&self) -> &[u8] {
154        &self.bytes
155    }
156    /// The caller's original wire correlation, with copied credential bytes zeroized.
157    ///
158    /// # Errors
159    /// The checked stored request could not be read.
160    pub fn correlation(&self) -> Result<(u32, u64), CallError> {
161        let mut frames = self.frames()?;
162        for frame in &mut frames {
163            if let Frame::Connect { auth_token, .. } = frame {
164                zeroize::Zeroize::zeroize(auth_token);
165            }
166        }
167        let Some(Frame::ConversationOpen {
168            stream_id,
169            conversation_id,
170            ..
171        }) = frames.get(1)
172        else {
173            return Err(CallError::Invalid);
174        };
175        Ok((*stream_id, *conversation_id))
176    }
177}
178
179/// The exact far wire answer, including its named refusal and correlation fields.
180pub struct CallReply {
181    id: u64,
182    frame: Frame,
183}
184impl std::fmt::Debug for CallReply {
185    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
186        out.debug_struct("CallReply")
187            .field("id", &self.id)
188            .finish_non_exhaustive()
189    }
190}
191impl CallReply {
192    /// Keeps the far connection's answer byte-for-byte at the wire-frame level.
193    ///
194    /// # Errors
195    /// A zero operation id or a frame that is not a conversation answer.
196    pub fn new(id: u64, frame: Frame) -> Result<Self, CallError> {
197        if id == 0
198            || !matches!(
199                frame,
200                Frame::ConversationMessage { .. } | Frame::ConversationError { .. }
201            )
202        {
203            return Err(CallError::Invalid);
204        }
205        Ok(Self { id, frame })
206    }
207    /// This reply's exact link-local operation.
208    #[must_use]
209    pub const fn id(&self) -> u64 {
210        self.id
211    }
212    /// Releases the answer only after the near side checks the operation binding.
213    #[must_use]
214    pub fn into_frame(self) -> Frame {
215        self.frame
216    }
217}
218
219/// A control/body kind inside the existing sealed `LinkMessage` envelope.
220/// Nothing here derives Debug over a credential, payload or contract.
221pub enum CallPacket {
222    /// Execute one original request through the far admission path.
223    Request(CallRequest),
224    /// Return its original wire answer.
225    Reply(CallReply),
226    /// Cancel only this exact link-local operation.
227    Cancel(u64),
228    /// A far exchange's own whole topology, never a guessed station list.
229    Topology(Zeroizing<Vec<u8>>),
230}
231
232impl CallPacket {
233    /// Whether a decrypted crossing uses this explicitly versioned body.
234    #[must_use]
235    pub fn is_call(bytes: &[u8]) -> bool {
236        bytes.starts_with(wire::MAGIC)
237    }
238    /// Encodes within the configured body bound before Noise allocation.
239    ///
240    /// # Errors
241    /// The packet or nested frame exceeds the bound or is invalid.
242    pub fn encode(&self, bound: usize) -> Result<Zeroizing<Vec<u8>>, CallError> {
243        wire::encode(self, bound)
244    }
245    /// Decodes exact lengths and refuses any trailing bytes.
246    ///
247    /// # Errors
248    /// Unknown version/kind, malformed nested wire or a configured bound refusal.
249    pub fn decode(bytes: &[u8], bound: usize) -> Result<Self, CallError> {
250        wire::decode(bytes, bound)
251    }
252    /// Seals an encoded packet once on the existing keyed session.
253    ///
254    /// # Errors
255    /// Bound or Noise framing refusal; payload bytes are excluded from the error.
256    pub fn seal(&self, transport: &mut LinkTransport, bound: usize) -> Result<Vec<u8>, CallError> {
257        let bytes = self.encode(bound)?;
258        let mut sealed = vec![0; bytes.len() + NOISE_OVERHEAD];
259        let written = transport
260            .seal(&bytes, &mut sealed)
261            .map_err(|_| CallError::Invalid)?;
262        sealed.truncate(written);
263        Ok(sealed)
264    }
265    /// Opens the exact keyed message then applies its configured body bound.
266    ///
267    /// # Errors
268    /// Oversize, Noise authentication or packet decoding refusal.
269    pub fn open(
270        transport: &mut LinkTransport,
271        sealed: &[u8],
272        bound: usize,
273    ) -> Result<Self, CallError> {
274        if sealed.len() > NOISE_MAX_MESSAGE || sealed.len() > bound.saturating_add(NOISE_OVERHEAD) {
275            return Err(CallError::Bound);
276        }
277        let mut plain = Zeroizing::new(vec![0; sealed.len()]);
278        let written = transport
279            .open(sealed, &mut plain)
280            .map_err(|_| CallError::Invalid)?;
281        plain.truncate(written);
282        Self::decode(&plain, bound)
283    }
284}
liminal_server/link_route_call.rs

liminal_server/
link_route_call.rs
