orbit_invoke/
lib.rs

1//! Bounded, fleet-local invocation transport over a dedicated Orbit ring.
2//!
3//! This crate owns transport facts only: invocation identity, operation
4//! routing, opaque payload bytes, retained cursors and loss reporting. It does
5//! not know what an operation runs, who runs it, or what happens when it
6//! fails. Those belong to the runtime adapter consuming an invocation.
7//!
8//! Publication proves only that a frame was committed. The ring is not a
9//! durable queue and provides no acknowledgement, claim, retry, response or
10//! exactly-once guarantee; cursors report overwritten frames as lag. A claim
11//! or result protocol must add its own authoritative state rather than change
12//! the meaning of this request ring.
13//!
14//! The `tokio` feature adds [`InvocationSubscription`], an async adapter that
15//! waits on the ring's native readiness fd where the platform provides one and
16//! polls otherwise.
17
18#[cfg(feature = "tokio")]
19use std::collections::VecDeque;
20use std::sync::Arc;
21#[cfg(feature = "tokio")]
22use std::time::Duration;
23
24use bytes::{BufMut, Bytes, BytesMut};
25#[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
26use orbit_core::RingEventFd;
27use orbit_core::fleet::FleetLaneCursor;
28use orbit_core::{Fleet, NetId64, NodeId, OrbitEpoch, OrbitTyped, RingSpec};
29
30pub type Result<T, E = Error> = std::result::Result<T, E>;
31
32/// Ring kind reserved for Orbit invocations.
33pub const INVOCATION_RING_KIND: u8 = 241;
34/// Number of retained invocation frames in each fleet-node lane.
35pub const INVOCATION_RING_CAPACITY: usize =
36    orbit_core::compile::usize_from_env(option_env!("ORBIT_INVOKE_RING_CAPACITY"), 256);
37/// Maximum bytes available to one invocation envelope.
38pub const INVOCATION_RING_PAYLOAD_CAPACITY: usize =
39    orbit_core::compile::usize_from_env(option_env!("ORBIT_INVOKE_RING_PAYLOAD_CAPACITY"), 8_192);
40pub const INVOCATION_RING_SPEC: RingSpec =
41    RingSpec::per_node(INVOCATION_RING_CAPACITY, INVOCATION_RING_PAYLOAD_CAPACITY);
42pub const INVOCATION_PAYLOAD_MAX: usize = INVOCATION_RING_SPEC.payload_capacity;
43
44const FRAME_KIND_INVOCATION: u8 = 1;
45const HEADER_LEN: usize = 2 + 4;
46#[cfg(feature = "tokio")]
47const FALLBACK_POLL_INTERVAL: Duration = Duration::from_millis(10);
48
49const _: () = assert!(INVOCATION_RING_CAPACITY.is_power_of_two());
50const _: () = assert!(INVOCATION_RING_PAYLOAD_CAPACITY >= HEADER_LEN);
51const _: () = assert!(INVOCATION_RING_PAYLOAD_CAPACITY <= u32::MAX as usize);
52
53#[derive(Clone, Debug)]
54struct InvocationRecord;
55
56impl OrbitTyped for InvocationRecord {
57    const KIND: u8 = INVOCATION_RING_KIND;
58    const RING_SPEC: RingSpec = INVOCATION_RING_SPEC;
59}
60
61/// Identity assigned when an invocation is committed to the ring.
62///
63/// Its embedded node is the submitting fleet node. It identifies the
64/// invocation, not the executor that may later handle it.
65#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
66pub struct InvocationId(NetId64);
67
68impl InvocationId {
69    pub const fn from_net_id(id: NetId64) -> Self {
70        Self(id)
71    }
72
73    pub const fn net_id(self) -> NetId64 {
74        self.0
75    }
76
77    pub const fn node(self) -> u16 {
78        self.0.node()
79    }
80}
81
82impl std::fmt::Display for InvocationId {
83    fn fmt(
84        &self,
85        formatter: &mut std::fmt::Formatter<'_>
86    ) -> std::fmt::Result {
87        self.0.fmt(formatter)
88    }
89}
90
91/// One retained invocation with an opaque runtime-owned payload.
92#[derive(Clone, Debug, Eq, PartialEq)]
93pub struct Invocation {
94    pub id: InvocationId,
95    pub operation: String,
96    pub payload: Vec<u8>,
97    /// Publication time carried by the Orbit frame version.
98    pub submitted_at: OrbitEpoch
99}
100
101/// A runtime-owned typed payload codec.
102///
103/// Orbit selects no serializer. Implementations may use a compact binary
104/// schema, a language value codec or an existing framework wire format.
105pub trait InvocationCodec: Sized {
106    const OPERATION: &'static str;
107
108    fn encode_invocation(&self) -> std::result::Result<Vec<u8>, String>;
109
110    fn decode_invocation(payload: &[u8]) -> std::result::Result<Self, String>;
111}
112
113impl Invocation {
114    pub fn decode<C: InvocationCodec>(&self) -> Result<C> {
115        if self.operation != C::OPERATION {
116            return Err(Error::OperationMismatch {
117                expected: C::OPERATION,
118                actual: self.operation.clone()
119            });
120        }
121        C::decode_invocation(&self.payload).map_err(Error::Codec)
122    }
123}
124
125/// Result of advancing one caller-owned invocation cursor.
126#[derive(Clone, Debug, Default, Eq, PartialEq)]
127pub struct InvocationPoll {
128    pub invocations: Vec<Invocation>,
129    /// Frames no longer available or invalid for this invocation ring.
130    pub lagged: u64
131}
132
133impl InvocationPoll {
134    pub fn is_empty(&self) -> bool {
135        self.invocations.is_empty() && self.lagged == 0
136    }
137}
138
139/// Caller-owned positions for every fleet-node lane.
140#[derive(Clone, Debug, PartialEq, Eq)]
141pub struct InvocationCursor {
142    inner: FleetLaneCursor
143}
144
145/// Cheap fleet-local handle for publishing and polling invocations.
146#[derive(Clone)]
147pub struct InvocationBus {
148    fleet: Arc<Fleet>
149}
150
151impl InvocationBus {
152    pub fn new(fleet: Arc<Fleet>) -> Self {
153        Self { fleet }
154    }
155
156    pub fn node_id(&self) -> NodeId {
157        self.fleet.node_id()
158    }
159
160    pub fn cursor_at_head(&self) -> InvocationCursor {
161        InvocationCursor { inner: self.fleet.lane_cursor_at_head::<InvocationRecord>() }
162    }
163
164    pub fn cursor_from_start(&self) -> InvocationCursor {
165        InvocationCursor { inner: self.fleet.lane_cursor_from_start::<InvocationRecord>() }
166    }
167
168    /// Subscribe to future invocations for one operation.
169    ///
170    /// Every subscription owns an independent cursor. This is filtering, not
171    /// work claiming: executor placement remains the caller's responsibility.
172    #[cfg(feature = "tokio")]
173    pub fn subscribe(
174        self: &Arc<Self>,
175        operation: impl Into<String>
176    ) -> Result<InvocationSubscription> {
177        let operation = operation.into();
178        validate_operation(&operation)?;
179        #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
180        let event_fd = self.subscription_event_fd()?;
181        Ok(InvocationSubscription {
182            cursor: self.cursor_at_head(),
183            bus: Arc::clone(self),
184            operation,
185            pending: VecDeque::new(),
186            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
187            event_fd
188        })
189    }
190
191    /// Clear the invocation ring during quiescent owner boot.
192    pub fn reset_ring(&self) -> Result<()> {
193        self.fleet.reset_ring::<InvocationRecord>().map_err(Error::Io)
194    }
195
196    /// Create this process' readiness fd for the invocation ring.
197    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
198    pub fn event_fd(&self) -> Result<RingEventFd> {
199        self.fleet.ring_event_fd::<InvocationRecord>().map_err(Error::Io)
200    }
201
202    #[cfg(all(
203        feature = "tokio",
204        any(target_os = "linux", target_os = "freebsd", target_os = "macos")
205    ))]
206    fn subscription_event_fd(&self) -> Result<Option<tokio::io::unix::AsyncFd<RingEventFd>>> {
207        if !self.fleet.is_shm() {
208            return Ok(None);
209        }
210        match self.event_fd() {
211            Ok(event_fd) => Ok(Some(tokio::io::unix::AsyncFd::new(event_fd).map_err(Error::Io)?)),
212            #[cfg(target_os = "macos")]
213            Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::Unsupported => Ok(None),
214            Err(error) => Err(error)
215        }
216    }
217
218    /// Commit a raw invocation. Success means publication, not execution.
219    pub fn submit(
220        &self,
221        operation: &str,
222        payload: &[u8]
223    ) -> Result<InvocationId> {
224        validate_operation(operation)?;
225        let submitted_at = OrbitEpoch::now();
226        let frame = encode_frame(operation.as_bytes(), payload)?;
227        #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
228        let id = self
229            .fleet
230            .publish_notified::<InvocationRecord>(
231                FRAME_KIND_INVOCATION,
232                submitted_at.as_unix_ms(),
233                frame
234            )
235            .map_err(Error::Io)?;
236        #[cfg(not(any(target_os = "linux", target_os = "freebsd", target_os = "macos")))]
237        let id = self.fleet.publish::<InvocationRecord>(
238            FRAME_KIND_INVOCATION,
239            submitted_at.as_unix_ms(),
240            frame
241        );
242        Ok(InvocationId(id))
243    }
244
245    pub fn submit_typed<C: InvocationCodec>(
246        &self,
247        invocation: &C
248    ) -> Result<InvocationId> {
249        let payload = invocation.encode_invocation().map_err(Error::Codec)?;
250        self.submit(C::OPERATION, &payload)
251    }
252
253    /// Advance every node lane once. Ordering is lane-local, not fleet-global.
254    pub fn poll(
255        &self,
256        cursor: &mut InvocationCursor
257    ) -> InvocationPoll {
258        let poll = self.fleet.poll_lanes::<InvocationRecord>(&mut cursor.inner);
259        let mut lagged = poll.loss.total();
260        let mut invocations = Vec::with_capacity(poll.frames.len());
261        for frame in poll.frames {
262            if frame.kind != FRAME_KIND_INVOCATION {
263                lagged = lagged.saturating_add(1);
264                continue;
265            }
266            let Some(decoded) = decode_frame(&frame.payload) else {
267                lagged = lagged.saturating_add(1);
268                continue;
269            };
270            let Ok(operation) = std::str::from_utf8(decoded.operation) else {
271                lagged = lagged.saturating_add(1);
272                continue;
273            };
274            invocations.push(Invocation {
275                id: InvocationId(frame.id),
276                operation: operation.to_owned(),
277                payload: decoded.payload.to_vec(),
278                submitted_at: OrbitEpoch::from_unix_ms(frame.ver)
279            });
280        }
281        InvocationPoll { invocations, lagged }
282    }
283
284    /// Poll one operation while still advancing past every ring frame.
285    pub fn poll_operation(
286        &self,
287        cursor: &mut InvocationCursor,
288        operation: &str
289    ) -> InvocationPoll {
290        let mut poll = self.poll(cursor);
291        poll.invocations.retain(|invocation| invocation.operation == operation);
292        poll
293    }
294}
295
296/// Async readiness adapter over one operation-filtered invocation cursor.
297#[cfg(feature = "tokio")]
298pub struct InvocationSubscription {
299    bus: Arc<InvocationBus>,
300    operation: String,
301    cursor: InvocationCursor,
302    pending: VecDeque<Invocation>,
303    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
304    event_fd: Option<tokio::io::unix::AsyncFd<RingEventFd>>
305}
306
307#[cfg(feature = "tokio")]
308impl InvocationSubscription {
309    /// Wait for the next retained invocation.
310    ///
311    /// Cancellation is deliberately external: callers may select this future
312    /// against the lifecycle token they own without teaching Orbit about that
313    /// lifecycle.
314    pub async fn receive(&mut self) -> Result<Invocation> {
315        loop {
316            if let Some(invocation) = self.pending.pop_front() {
317                return Ok(invocation);
318            }
319
320            let poll = self.bus.poll_operation(&mut self.cursor, &self.operation);
321            self.pending.extend(poll.invocations);
322            if poll.lagged > 0 {
323                return Err(Error::Lagged(poll.lagged));
324            }
325            if let Some(invocation) = self.pending.pop_front() {
326                return Ok(invocation);
327            }
328
329            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
330            if let Some(event_fd) = &self.event_fd {
331                let mut ready = event_fd.readable().await.map_err(Error::Io)?;
332                event_fd.get_ref().drain().map_err(Error::Io)?;
333                ready.clear_ready();
334                continue;
335            }
336
337            tokio::time::sleep(FALLBACK_POLL_INTERVAL).await;
338        }
339    }
340}
341
342#[derive(Debug)]
343pub enum Error {
344    EmptyOperation,
345    OperationTooLong(usize),
346    FrameTooLarge { operation_len: usize, payload_len: usize, max: usize },
347    OperationMismatch { expected: &'static str, actual: String },
348    Lagged(u64),
349    Codec(String),
350    Io(std::io::Error)
351}
352
353impl std::fmt::Display for Error {
354    fn fmt(
355        &self,
356        formatter: &mut std::fmt::Formatter<'_>
357    ) -> std::fmt::Result {
358        match self {
359            Self::EmptyOperation => formatter.write_str("invocation operation cannot be empty"),
360            Self::OperationTooLong(len) => {
361                write!(formatter, "invocation operation is too long: len={len} max=65535")
362            }
363            Self::FrameTooLarge { operation_len, payload_len, max } => write!(
364                formatter,
365                "invocation frame is too large: operation_len={operation_len} payload_len={payload_len} max={max}"
366            ),
367            Self::OperationMismatch { expected, actual } => write!(
368                formatter,
369                "invocation operation mismatch: expected {expected}, got {actual}"
370            ),
371            Self::Lagged(count) => {
372                write!(formatter, "invocation subscription lagged by {count} frame(s)")
373            }
374            Self::Codec(error) => write!(formatter, "invocation codec failed: {error}"),
375            Self::Io(error) => write!(formatter, "invocation ring io error: {error}")
376        }
377    }
378}
379
380impl std::error::Error for Error {
381    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
382        match self {
383            Self::Io(error) => Some(error),
384            _ => None
385        }
386    }
387}
388
389struct DecodedFrame<'a> {
390    operation: &'a [u8],
391    payload: &'a [u8]
392}
393
394fn validate_operation(operation: &str) -> Result<()> {
395    if operation.trim().is_empty() {
396        return Err(Error::EmptyOperation);
397    }
398    if operation.len() > usize::from(u16::MAX) {
399        return Err(Error::OperationTooLong(operation.len()));
400    }
401    Ok(())
402}
403
404fn encode_frame(
405    operation: &[u8],
406    payload: &[u8]
407) -> Result<Bytes> {
408    let total = HEADER_LEN
409        .checked_add(operation.len())
410        .and_then(|len| len.checked_add(payload.len()))
411        .ok_or(Error::FrameTooLarge {
412            operation_len: operation.len(),
413            payload_len: payload.len(),
414            max: INVOCATION_PAYLOAD_MAX
415        })?;
416    if operation.len() > usize::from(u16::MAX)
417        || payload.len() > u32::MAX as usize
418        || total > INVOCATION_PAYLOAD_MAX
419    {
420        return Err(Error::FrameTooLarge {
421            operation_len: operation.len(),
422            payload_len: payload.len(),
423            max: INVOCATION_PAYLOAD_MAX
424        });
425    }
426    let mut frame = BytesMut::with_capacity(total);
427    frame.put_u16_le(operation.len() as u16);
428    frame.put_u32_le(payload.len() as u32);
429    frame.put_slice(operation);
430    frame.put_slice(payload);
431    Ok(frame.freeze())
432}
433
434fn decode_frame(frame: &Bytes) -> Option<DecodedFrame<'_>> {
435    if frame.len() < HEADER_LEN {
436        return None;
437    }
438    let operation_len = u16::from_le_bytes(frame[0..2].try_into().ok()?) as usize;
439    let payload_len = u32::from_le_bytes(frame[2..6].try_into().ok()?) as usize;
440    let operation_end = HEADER_LEN.checked_add(operation_len)?;
441    let payload_end = operation_end.checked_add(payload_len)?;
442    if payload_end != frame.len() {
443        return None;
444    }
445    Some(DecodedFrame {
446        operation: &frame[HEADER_LEN..operation_end],
447        payload: &frame[operation_end..payload_end]
448    })
449}
orbit_invoke/lib.rs

orbit_invoke/
lib.rs
