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 = if self.fleet.is_shm() {
182            Some(tokio::io::unix::AsyncFd::new(self.event_fd()?).map_err(Error::Io)?)
183        } else {
184            None
185        };
186        Ok(InvocationSubscription {
187            cursor: self.cursor_at_head(),
188            bus: Arc::clone(self),
189            operation,
190            pending: VecDeque::new(),
191            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
192            event_fd,
193        })
194    }
195
196    /// Clear the invocation ring during quiescent owner boot.
197    pub fn reset_ring(&self) -> Result<()> {
198        self.fleet
199            .reset_ring::<InvocationRecord>()
200            .map_err(Error::Io)
201    }
202
203    /// Create this process' readiness fd for the invocation ring.
204    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
205    pub fn event_fd(&self) -> Result<RingEventFd> {
206        self.fleet
207            .ring_event_fd::<InvocationRecord>()
208            .map_err(Error::Io)
209    }
210
211    /// Commit a raw invocation. Success means publication, not execution.
212    pub fn submit(&self, operation: &str, payload: &[u8]) -> Result<InvocationId> {
213        validate_operation(operation)?;
214        let submitted_at = OrbitEpoch::now();
215        let frame = encode_frame(operation.as_bytes(), payload)?;
216        #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
217        let id = self
218            .fleet
219            .publish_notified::<InvocationRecord>(
220                FRAME_KIND_INVOCATION,
221                submitted_at.as_unix_ms(),
222                frame,
223            )
224            .map_err(Error::Io)?;
225        #[cfg(not(any(target_os = "linux", target_os = "freebsd", target_os = "macos")))]
226        let id = self.fleet.publish::<InvocationRecord>(
227            FRAME_KIND_INVOCATION,
228            submitted_at.as_unix_ms(),
229            frame,
230        );
231        Ok(InvocationId(id))
232    }
233
234    pub fn submit_typed<C: InvocationCodec>(&self, invocation: &C) -> Result<InvocationId> {
235        let payload = invocation.encode_invocation().map_err(Error::Codec)?;
236        self.submit(C::OPERATION, &payload)
237    }
238
239    /// Advance every node lane once. Ordering is lane-local, not fleet-global.
240    pub fn poll(&self, cursor: &mut InvocationCursor) -> InvocationPoll {
241        let poll = self.fleet.poll_lanes::<InvocationRecord>(&mut cursor.inner);
242        let mut lagged = poll.loss.total();
243        let mut invocations = Vec::with_capacity(poll.frames.len());
244        for frame in poll.frames {
245            if frame.kind != FRAME_KIND_INVOCATION {
246                lagged = lagged.saturating_add(1);
247                continue;
248            }
249            let Some(decoded) = decode_frame(&frame.payload) else {
250                lagged = lagged.saturating_add(1);
251                continue;
252            };
253            let Ok(operation) = std::str::from_utf8(decoded.operation) else {
254                lagged = lagged.saturating_add(1);
255                continue;
256            };
257            invocations.push(Invocation {
258                id: InvocationId(frame.id),
259                operation: operation.to_owned(),
260                payload: decoded.payload.to_vec(),
261                submitted_at: OrbitEpoch::from_unix_ms(frame.ver),
262            });
263        }
264        InvocationPoll {
265            invocations,
266            lagged,
267        }
268    }
269
270    /// Poll one operation while still advancing past every ring frame.
271    pub fn poll_operation(&self, cursor: &mut InvocationCursor, operation: &str) -> InvocationPoll {
272        let mut poll = self.poll(cursor);
273        poll.invocations
274            .retain(|invocation| invocation.operation == operation);
275        poll
276    }
277}
278
279/// Async readiness adapter over one operation-filtered invocation cursor.
280#[cfg(feature = "tokio")]
281pub struct InvocationSubscription {
282    bus: Arc<InvocationBus>,
283    operation: String,
284    cursor: InvocationCursor,
285    pending: VecDeque<Invocation>,
286    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
287    event_fd: Option<tokio::io::unix::AsyncFd<RingEventFd>>,
288}
289
290#[cfg(feature = "tokio")]
291impl InvocationSubscription {
292    /// Wait for the next retained invocation.
293    ///
294    /// Cancellation is deliberately external: callers may select this future
295    /// against the lifecycle token they own without teaching Orbit about that
296    /// lifecycle.
297    pub async fn receive(&mut self) -> Result<Invocation> {
298        loop {
299            if let Some(invocation) = self.pending.pop_front() {
300                return Ok(invocation);
301            }
302
303            let poll = self.bus.poll_operation(&mut self.cursor, &self.operation);
304            self.pending.extend(poll.invocations);
305            if poll.lagged > 0 {
306                return Err(Error::Lagged(poll.lagged));
307            }
308            if let Some(invocation) = self.pending.pop_front() {
309                return Ok(invocation);
310            }
311
312            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
313            if let Some(event_fd) = &self.event_fd {
314                let mut ready = event_fd.readable().await.map_err(Error::Io)?;
315                event_fd.get_ref().drain().map_err(Error::Io)?;
316                ready.clear_ready();
317                continue;
318            }
319
320            tokio::time::sleep(FALLBACK_POLL_INTERVAL).await;
321        }
322    }
323}
324
325#[derive(Debug)]
326pub enum Error {
327    EmptyOperation,
328    OperationTooLong(usize),
329    FrameTooLarge {
330        operation_len: usize,
331        payload_len: usize,
332        max: usize,
333    },
334    OperationMismatch {
335        expected: &'static str,
336        actual: String,
337    },
338    Lagged(u64),
339    Codec(String),
340    Io(std::io::Error),
341}
342
343impl std::fmt::Display for Error {
344    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
345        match self {
346            Self::EmptyOperation => formatter.write_str("invocation operation cannot be empty"),
347            Self::OperationTooLong(len) => {
348                write!(
349                    formatter,
350                    "invocation operation is too long: len={len} max=65535"
351                )
352            }
353            Self::FrameTooLarge {
354                operation_len,
355                payload_len,
356                max,
357            } => write!(
358                formatter,
359                "invocation frame is too large: operation_len={operation_len} payload_len={payload_len} max={max}"
360            ),
361            Self::OperationMismatch { expected, actual } => write!(
362                formatter,
363                "invocation operation mismatch: expected {expected}, got {actual}"
364            ),
365            Self::Lagged(count) => {
366                write!(
367                    formatter,
368                    "invocation subscription lagged by {count} frame(s)"
369                )
370            }
371            Self::Codec(error) => write!(formatter, "invocation codec failed: {error}"),
372            Self::Io(error) => write!(formatter, "invocation ring io error: {error}"),
373        }
374    }
375}
376
377impl std::error::Error for Error {
378    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
379        match self {
380            Self::Io(error) => Some(error),
381            _ => None,
382        }
383    }
384}
385
386struct DecodedFrame<'a> {
387    operation: &'a [u8],
388    payload: &'a [u8],
389}
390
391fn validate_operation(operation: &str) -> Result<()> {
392    if operation.trim().is_empty() {
393        return Err(Error::EmptyOperation);
394    }
395    if operation.len() > usize::from(u16::MAX) {
396        return Err(Error::OperationTooLong(operation.len()));
397    }
398    Ok(())
399}
400
401fn encode_frame(operation: &[u8], payload: &[u8]) -> Result<Bytes> {
402    let total = HEADER_LEN
403        .checked_add(operation.len())
404        .and_then(|len| len.checked_add(payload.len()))
405        .ok_or(Error::FrameTooLarge {
406            operation_len: operation.len(),
407            payload_len: payload.len(),
408            max: INVOCATION_PAYLOAD_MAX,
409        })?;
410    if operation.len() > usize::from(u16::MAX)
411        || payload.len() > u32::MAX as usize
412        || total > INVOCATION_PAYLOAD_MAX
413    {
414        return Err(Error::FrameTooLarge {
415            operation_len: operation.len(),
416            payload_len: payload.len(),
417            max: INVOCATION_PAYLOAD_MAX,
418        });
419    }
420    let mut frame = BytesMut::with_capacity(total);
421    frame.put_u16_le(operation.len() as u16);
422    frame.put_u32_le(payload.len() as u32);
423    frame.put_slice(operation);
424    frame.put_slice(payload);
425    Ok(frame.freeze())
426}
427
428fn decode_frame(frame: &Bytes) -> Option<DecodedFrame<'_>> {
429    if frame.len() < HEADER_LEN {
430        return None;
431    }
432    let operation_len = u16::from_le_bytes(frame[0..2].try_into().ok()?) as usize;
433    let payload_len = u32::from_le_bytes(frame[2..6].try_into().ok()?) as usize;
434    let operation_end = HEADER_LEN.checked_add(operation_len)?;
435    let payload_end = operation_end.checked_add(payload_len)?;
436    if payload_end != frame.len() {
437        return None;
438    }
439    Some(DecodedFrame {
440        operation: &frame[HEADER_LEN..operation_end],
441        payload: &frame[operation_end..payload_end],
442    })
443}
orbit_invoke/lib.rs

orbit_invoke/
lib.rs
