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(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        self.0.fmt(formatter)
85    }
86}
87
88/// One retained invocation with an opaque runtime-owned payload.
89#[derive(Clone, Debug, Eq, PartialEq)]
90pub struct Invocation {
91    pub id: InvocationId,
92    pub operation: String,
93    pub payload: Vec<u8>,
94    /// Publication time carried by the Orbit frame version.
95    pub submitted_at: OrbitEpoch,
96}
97
98/// A runtime-owned typed payload codec.
99///
100/// Orbit selects no serializer. Implementations may use a compact binary
101/// schema, a language value codec or an existing framework wire format.
102pub trait InvocationCodec: Sized {
103    const OPERATION: &'static str;
104
105    fn encode_invocation(&self) -> std::result::Result<Vec<u8>, String>;
106
107    fn decode_invocation(payload: &[u8]) -> std::result::Result<Self, String>;
108}
109
110impl Invocation {
111    pub fn decode<C: InvocationCodec>(&self) -> Result<C> {
112        if self.operation != C::OPERATION {
113            return Err(Error::OperationMismatch {
114                expected: C::OPERATION,
115                actual: self.operation.clone(),
116            });
117        }
118        C::decode_invocation(&self.payload).map_err(Error::Codec)
119    }
120}
121
122/// Result of advancing one caller-owned invocation cursor.
123#[derive(Clone, Debug, Default, Eq, PartialEq)]
124pub struct InvocationPoll {
125    pub invocations: Vec<Invocation>,
126    /// Frames no longer available or invalid for this invocation ring.
127    pub lagged: u64,
128}
129
130impl InvocationPoll {
131    pub fn is_empty(&self) -> bool {
132        self.invocations.is_empty() && self.lagged == 0
133    }
134}
135
136/// Caller-owned positions for every fleet-node lane.
137#[derive(Clone, Debug, PartialEq, Eq)]
138pub struct InvocationCursor {
139    inner: FleetLaneCursor,
140}
141
142/// Cheap fleet-local handle for publishing and polling invocations.
143#[derive(Clone)]
144pub struct InvocationBus {
145    fleet: Arc<Fleet>,
146}
147
148impl InvocationBus {
149    pub fn new(fleet: Arc<Fleet>) -> Self {
150        Self { fleet }
151    }
152
153    pub fn node_id(&self) -> NodeId {
154        self.fleet.node_id()
155    }
156
157    pub fn cursor_at_head(&self) -> InvocationCursor {
158        InvocationCursor {
159            inner: self.fleet.lane_cursor_at_head::<InvocationRecord>(),
160        }
161    }
162
163    pub fn cursor_from_start(&self) -> InvocationCursor {
164        InvocationCursor {
165            inner: self.fleet.lane_cursor_from_start::<InvocationRecord>(),
166        }
167    }
168
169    /// Subscribe to future invocations for one operation.
170    ///
171    /// Every subscription owns an independent cursor. This is filtering, not
172    /// work claiming: executor placement remains the caller's responsibility.
173    #[cfg(feature = "tokio")]
174    pub fn subscribe(
175        self: &Arc<Self>,
176        operation: impl Into<String>,
177    ) -> Result<InvocationSubscription> {
178        let operation = operation.into();
179        validate_operation(&operation)?;
180        #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
181        let event_fd = self.subscription_event_fd()?;
182        Ok(InvocationSubscription {
183            cursor: self.cursor_at_head(),
184            bus: Arc::clone(self),
185            operation,
186            pending: VecDeque::new(),
187            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
188            event_fd,
189        })
190    }
191
192    /// Clear the invocation ring during quiescent owner boot.
193    pub fn reset_ring(&self) -> Result<()> {
194        self.fleet
195            .reset_ring::<InvocationRecord>()
196            .map_err(Error::Io)
197    }
198
199    /// Create this process' readiness fd for the invocation ring.
200    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
201    pub fn event_fd(&self) -> Result<RingEventFd> {
202        self.fleet
203            .ring_event_fd::<InvocationRecord>()
204            .map_err(Error::Io)
205    }
206
207    #[cfg(all(
208        feature = "tokio",
209        any(target_os = "linux", target_os = "freebsd", target_os = "macos")
210    ))]
211    fn subscription_event_fd(&self) -> Result<Option<tokio::io::unix::AsyncFd<RingEventFd>>> {
212        if !self.fleet.is_shm() {
213            return Ok(None);
214        }
215        match self.event_fd() {
216            Ok(event_fd) => Ok(Some(
217                tokio::io::unix::AsyncFd::new(event_fd).map_err(Error::Io)?,
218            )),
219            #[cfg(target_os = "macos")]
220            Err(Error::Io(error)) if error.kind() == std::io::ErrorKind::Unsupported => Ok(None),
221            Err(error) => Err(error),
222        }
223    }
224
225    /// Commit a raw invocation. Success means publication, not execution.
226    pub fn submit(&self, operation: &str, payload: &[u8]) -> Result<InvocationId> {
227        validate_operation(operation)?;
228        let submitted_at = OrbitEpoch::now();
229        let frame = encode_frame(operation.as_bytes(), payload)?;
230        #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
231        let id = self
232            .fleet
233            .publish_notified::<InvocationRecord>(
234                FRAME_KIND_INVOCATION,
235                submitted_at.as_unix_ms(),
236                frame,
237            )
238            .map_err(Error::Io)?;
239        #[cfg(not(any(target_os = "linux", target_os = "freebsd", target_os = "macos")))]
240        let id = self.fleet.publish::<InvocationRecord>(
241            FRAME_KIND_INVOCATION,
242            submitted_at.as_unix_ms(),
243            frame,
244        );
245        Ok(InvocationId(id))
246    }
247
248    pub fn submit_typed<C: InvocationCodec>(&self, invocation: &C) -> 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(&self, cursor: &mut InvocationCursor) -> InvocationPoll {
255        let poll = self.fleet.poll_lanes::<InvocationRecord>(&mut cursor.inner);
256        let mut lagged = poll.loss.total();
257        let mut invocations = Vec::with_capacity(poll.frames.len());
258        for frame in poll.frames {
259            if frame.kind != FRAME_KIND_INVOCATION {
260                lagged = lagged.saturating_add(1);
261                continue;
262            }
263            let Some(decoded) = decode_frame(&frame.payload) else {
264                lagged = lagged.saturating_add(1);
265                continue;
266            };
267            let Ok(operation) = std::str::from_utf8(decoded.operation) else {
268                lagged = lagged.saturating_add(1);
269                continue;
270            };
271            invocations.push(Invocation {
272                id: InvocationId(frame.id),
273                operation: operation.to_owned(),
274                payload: decoded.payload.to_vec(),
275                submitted_at: OrbitEpoch::from_unix_ms(frame.ver),
276            });
277        }
278        InvocationPoll {
279            invocations,
280            lagged,
281        }
282    }
283
284    /// Poll one operation while still advancing past every ring frame.
285    pub fn poll_operation(&self, cursor: &mut InvocationCursor, operation: &str) -> InvocationPoll {
286        let mut poll = self.poll(cursor);
287        poll.invocations
288            .retain(|invocation| invocation.operation == operation);
289        poll
290    }
291}
292
293/// Async readiness adapter over one operation-filtered invocation cursor.
294#[cfg(feature = "tokio")]
295pub struct InvocationSubscription {
296    bus: Arc<InvocationBus>,
297    operation: String,
298    cursor: InvocationCursor,
299    pending: VecDeque<Invocation>,
300    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
301    event_fd: Option<tokio::io::unix::AsyncFd<RingEventFd>>,
302}
303
304#[cfg(feature = "tokio")]
305impl InvocationSubscription {
306    /// Wait for the next retained invocation.
307    ///
308    /// Cancellation is deliberately external: callers may select this future
309    /// against the lifecycle token they own without teaching Orbit about that
310    /// lifecycle.
311    pub async fn receive(&mut self) -> Result<Invocation> {
312        loop {
313            if let Some(invocation) = self.pending.pop_front() {
314                return Ok(invocation);
315            }
316
317            let poll = self.bus.poll_operation(&mut self.cursor, &self.operation);
318            self.pending.extend(poll.invocations);
319            if poll.lagged > 0 {
320                return Err(Error::Lagged(poll.lagged));
321            }
322            if let Some(invocation) = self.pending.pop_front() {
323                return Ok(invocation);
324            }
325
326            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
327            if let Some(event_fd) = &self.event_fd {
328                let mut ready = event_fd.readable().await.map_err(Error::Io)?;
329                event_fd.get_ref().drain().map_err(Error::Io)?;
330                ready.clear_ready();
331                continue;
332            }
333
334            tokio::time::sleep(FALLBACK_POLL_INTERVAL).await;
335        }
336    }
337}
338
339#[derive(Debug)]
340pub enum Error {
341    EmptyOperation,
342    OperationTooLong(usize),
343    FrameTooLarge {
344        operation_len: usize,
345        payload_len: usize,
346        max: usize,
347    },
348    OperationMismatch {
349        expected: &'static str,
350        actual: String,
351    },
352    Lagged(u64),
353    Codec(String),
354    Io(std::io::Error),
355}
356
357impl std::fmt::Display for Error {
358    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
359        match self {
360            Self::EmptyOperation => formatter.write_str("invocation operation cannot be empty"),
361            Self::OperationTooLong(len) => {
362                write!(
363                    formatter,
364                    "invocation operation is too long: len={len} max=65535"
365                )
366            }
367            Self::FrameTooLarge {
368                operation_len,
369                payload_len,
370                max,
371            } => write!(
372                formatter,
373                "invocation frame is too large: operation_len={operation_len} payload_len={payload_len} max={max}"
374            ),
375            Self::OperationMismatch { expected, actual } => write!(
376                formatter,
377                "invocation operation mismatch: expected {expected}, got {actual}"
378            ),
379            Self::Lagged(count) => {
380                write!(
381                    formatter,
382                    "invocation subscription lagged by {count} frame(s)"
383                )
384            }
385            Self::Codec(error) => write!(formatter, "invocation codec failed: {error}"),
386            Self::Io(error) => write!(formatter, "invocation ring io error: {error}"),
387        }
388    }
389}
390
391impl std::error::Error for Error {
392    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
393        match self {
394            Self::Io(error) => Some(error),
395            _ => None,
396        }
397    }
398}
399
400struct DecodedFrame<'a> {
401    operation: &'a [u8],
402    payload: &'a [u8],
403}
404
405fn validate_operation(operation: &str) -> Result<()> {
406    if operation.trim().is_empty() {
407        return Err(Error::EmptyOperation);
408    }
409    if operation.len() > usize::from(u16::MAX) {
410        return Err(Error::OperationTooLong(operation.len()));
411    }
412    Ok(())
413}
414
415fn encode_frame(operation: &[u8], payload: &[u8]) -> Result<Bytes> {
416    let total = HEADER_LEN
417        .checked_add(operation.len())
418        .and_then(|len| len.checked_add(payload.len()))
419        .ok_or(Error::FrameTooLarge {
420            operation_len: operation.len(),
421            payload_len: payload.len(),
422            max: INVOCATION_PAYLOAD_MAX,
423        })?;
424    if operation.len() > usize::from(u16::MAX)
425        || payload.len() > u32::MAX as usize
426        || total > INVOCATION_PAYLOAD_MAX
427    {
428        return Err(Error::FrameTooLarge {
429            operation_len: operation.len(),
430            payload_len: payload.len(),
431            max: INVOCATION_PAYLOAD_MAX,
432        });
433    }
434    let mut frame = BytesMut::with_capacity(total);
435    frame.put_u16_le(operation.len() as u16);
436    frame.put_u32_le(payload.len() as u32);
437    frame.put_slice(operation);
438    frame.put_slice(payload);
439    Ok(frame.freeze())
440}
441
442fn decode_frame(frame: &Bytes) -> Option<DecodedFrame<'_>> {
443    if frame.len() < HEADER_LEN {
444        return None;
445    }
446    let operation_len = u16::from_le_bytes(frame[0..2].try_into().ok()?) as usize;
447    let payload_len = u32::from_le_bytes(frame[2..6].try_into().ok()?) as usize;
448    let operation_end = HEADER_LEN.checked_add(operation_len)?;
449    let payload_end = operation_end.checked_add(payload_len)?;
450    if payload_end != frame.len() {
451        return None;
452    }
453    Some(DecodedFrame {
454        operation: &frame[HEADER_LEN..operation_end],
455        payload: &frame[operation_end..payload_end],
456    })
457}
orbit_invoke/lib.rs

orbit_invoke/
lib.rs
