liminal-server 0.25.4

//! Calls carry the caller's original admitted wire request inside Noise.
//! The far exchange executes it through its ordinary admitted connection path.

use crate::link_handshake::{LinkTransport, NOISE_MAX_MESSAGE, NOISE_OVERHEAD};
use liminal::protocol::{Frame, codec};
use zeroize::Zeroizing;

#[path = "link_route_call/client.rs"]
pub mod client;
#[path = "link_route_call/pending.rs"]
pub mod pending;
#[cfg(test)]
#[path = "link_route_call_tests.rs"]
mod tests;
#[path = "link_route_call/wire.rs"]
mod wire;

/// A full far connection execution, owned until its exact call settles.
pub trait CallExecution: Send + std::fmt::Debug {
    /// Cancels this call's full connection and joins its owned reader.
    ///
    /// # Errors
    /// The connection's cancellation or worker join failed.
    fn finish(&mut self) -> Result<(), CallError>;
}

/// The receiving supervisor supplies its ordinary admitted connection path.
pub trait CallExecutor: Send + Sync + std::fmt::Debug {
    /// Actual retained execution owners, for capacity and leak observations.
    fn retained(&self) -> usize;
    /// Waits on execution retirement signals, with no clock or polling.
    ///
    /// # Errors
    /// The retirement signal state was poisoned.
    fn observe_empty(&self) -> Result<(), CallError>;
    /// Starts only after the keyed link has decided its two crossing grants.
    ///
    /// # Errors
    /// Full connection admission, wire validation or owned reader startup failed.
    fn start(
        &self,
        request: CallRequest,
        completed: Box<dyn FnOnce(Result<CallReply, CallError>) + Send>,
    ) -> Result<Box<dyn CallExecution>, CallError>;
}

/// A named crossing refusal; no request or credential bytes enter its text.
#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
pub enum CallError {
    /// The complete call cannot fit its configured bound or the Noise ceiling.
    #[error("link_call_too_large")]
    Bound,
    /// No additional call is admitted until an existing reservation settles.
    #[error("link_call_capacity")]
    Capacity,
    /// The call envelope is malformed or has unexpected framing.
    #[error("link_call_invalid")]
    Invalid,
    /// A reply did not match this exact call and subject.
    #[error("link_call_correlation_mismatch")]
    Correlation,
    /// The keyed peer disappeared before its answer.
    #[error("link_call_peer_gone")]
    PeerGone,
    /// The local caller cancelled this exact pending operation.
    #[error("link_call_cancelled")]
    Cancelled,
    /// The caller's already configured pending-reply deadline elapsed.
    #[error("link_call_deadline")]
    Deadline,
    /// State was poisoned or a response delivery could not be completed.
    #[error("link_call_unavailable")]
    Unavailable,
    /// No live server on either end advertises this subject.
    #[error("no_server")]
    NoServer,
    /// A machine no longer holds its crossing right when the answer is released.
    #[error("link_call_grant_refused")]
    GrantRefused,
}

/// A zeroized sequence of Connect, `ConversationOpen` and `ConversationMessage`.
/// Private fields prevent the receiving adapter from accepting arbitrary acts.
pub struct CallRequest {
    id: u64,
    subject: String,
    bytes: Zeroizing<Vec<u8>>,
}

impl std::fmt::Debug for CallRequest {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("CallRequest")
            .field("id", &self.id)
            .field("bytes", &self.bytes.len())
            .finish_non_exhaustive()
    }
}

impl CallRequest {
    /// Builds the call from its caller's original credential and request frames.
    ///
    /// # Errors
    /// Wrong frame sequence, correlations/subject, non-Lys credential or oversize call.
    pub fn new(
        id: u64,
        subject: String,
        frames: &[Frame],
        bound: usize,
    ) -> Result<Self, CallError> {
        let bytes = wire::encode_frames(frames, bound)?;
        Self::decode(id, subject, bytes, bound)
    }

    fn decode(
        id: u64,
        subject: String,
        bytes: Zeroizing<Vec<u8>>,
        bound: usize,
    ) -> Result<Self, CallError> {
        if id == 0 || bytes.len() > bound {
            return Err(CallError::Invalid);
        }
        crate::auth_lys::SubjectBinding::new("wire".into(), subject.clone(), subject.clone())
            .map_err(|_| CallError::Invalid)?;
        let mut frames = wire::request_frames(&bytes, &subject)?;
        for frame in &mut frames {
            if let Frame::Connect { auth_token, .. } = frame {
                zeroize::Zeroize::zeroize(auth_token);
            }
        }
        drop(frames);
        Ok(Self { id, subject, bytes })
    }
    /// The monotonic link-local operation id, never reused on this transport.
    #[must_use]
    pub const fn id(&self) -> u64 {
        self.id
    }
    /// The concrete placed service subject.
    #[must_use]
    pub fn subject(&self) -> &str {
        &self.subject
    }
    /// The checked original frames, for the full far admission path.
    ///
    /// # Errors
    /// The stored wire bytes could not be decoded.
    pub fn frames(&self) -> Result<Vec<Frame>, CallError> {
        wire::request_frames(&self.bytes, &self.subject)
    }
    /// The checked wire image, borrowed only while it is sent to the far door.
    #[must_use]
    pub fn wire_bytes(&self) -> &[u8] {
        &self.bytes
    }
    /// The caller's original wire correlation, with copied credential bytes zeroized.
    ///
    /// # Errors
    /// The checked stored request could not be read.
    pub fn correlation(&self) -> Result<(u32, u64), CallError> {
        let mut frames = self.frames()?;
        for frame in &mut frames {
            if let Frame::Connect { auth_token, .. } = frame {
                zeroize::Zeroize::zeroize(auth_token);
            }
        }
        let Some(Frame::ConversationOpen {
            stream_id,
            conversation_id,
            ..
        }) = frames.get(1)
        else {
            return Err(CallError::Invalid);
        };
        Ok((*stream_id, *conversation_id))
    }
}

/// The exact far wire answer, including its named refusal and correlation fields.
pub struct CallReply {
    id: u64,
    frame: Frame,
}
impl std::fmt::Debug for CallReply {
    fn fmt(&self, out: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        out.debug_struct("CallReply")
            .field("id", &self.id)
            .finish_non_exhaustive()
    }
}
impl CallReply {
    /// Keeps the far connection's answer byte-for-byte at the wire-frame level.
    ///
    /// # Errors
    /// A zero operation id or a frame that is not a conversation answer.
    pub fn new(id: u64, frame: Frame) -> Result<Self, CallError> {
        if id == 0
            || !matches!(
                frame,
                Frame::ConversationMessage { .. } | Frame::ConversationError { .. }
            )
        {
            return Err(CallError::Invalid);
        }
        Ok(Self { id, frame })
    }
    /// This reply's exact link-local operation.
    #[must_use]
    pub const fn id(&self) -> u64 {
        self.id
    }
    /// Releases the answer only after the near side checks the operation binding.
    #[must_use]
    pub fn into_frame(self) -> Frame {
        self.frame
    }
}

/// A control/body kind inside the existing sealed `LinkMessage` envelope.
/// Nothing here derives Debug over a credential, payload or contract.
pub enum CallPacket {
    /// Execute one original request through the far admission path.
    Request(CallRequest),
    /// Return its original wire answer.
    Reply(CallReply),
    /// Cancel only this exact link-local operation.
    Cancel(u64),
    /// A far exchange's own whole topology, never a guessed station list.
    Topology(Zeroizing<Vec<u8>>),
}

impl CallPacket {
    /// Whether a decrypted crossing uses this explicitly versioned body.
    #[must_use]
    pub fn is_call(bytes: &[u8]) -> bool {
        bytes.starts_with(wire::MAGIC)
    }
    /// Encodes within the configured body bound before Noise allocation.
    ///
    /// # Errors
    /// The packet or nested frame exceeds the bound or is invalid.
    pub fn encode(&self, bound: usize) -> Result<Zeroizing<Vec<u8>>, CallError> {
        wire::encode(self, bound)
    }
    /// Decodes exact lengths and refuses any trailing bytes.
    ///
    /// # Errors
    /// Unknown version/kind, malformed nested wire or a configured bound refusal.
    pub fn decode(bytes: &[u8], bound: usize) -> Result<Self, CallError> {
        wire::decode(bytes, bound)
    }
    /// Seals an encoded packet once on the existing keyed session.
    ///
    /// # Errors
    /// Bound or Noise framing refusal; payload bytes are excluded from the error.
    pub fn seal(&self, transport: &mut LinkTransport, bound: usize) -> Result<Vec<u8>, CallError> {
        let bytes = self.encode(bound)?;
        let mut sealed = vec![0; bytes.len() + NOISE_OVERHEAD];
        let written = transport
            .seal(&bytes, &mut sealed)
            .map_err(|_| CallError::Invalid)?;
        sealed.truncate(written);
        Ok(sealed)
    }
    /// Opens the exact keyed message then applies its configured body bound.
    ///
    /// # Errors
    /// Oversize, Noise authentication or packet decoding refusal.
    pub fn open(
        transport: &mut LinkTransport,
        sealed: &[u8],
        bound: usize,
    ) -> Result<Self, CallError> {
        if sealed.len() > NOISE_MAX_MESSAGE || sealed.len() > bound.saturating_add(NOISE_OVERHEAD) {
            return Err(CallError::Bound);
        }
        let mut plain = Zeroizing::new(vec![0; sealed.len()]);
        let written = transport
            .open(sealed, &mut plain)
            .map_err(|_| CallError::Invalid)?;
        plain.truncate(written);
        Self::decode(&plain, bound)
    }
}