dora_runtime_shared_lib/
lib.rs

1//! Shared-library (C ABI) operator runtime backend.
2//!
3//! Loads `.so`/`.dll`/`.dylib` operators via `libloading` and runs them on the
4//! [`dora_runtime_api`] event loop. Shipped inside the `dora` CLI and launched
5//! by the daemon as the `dora runtime` subcommand.
6//!
7//! [`SharedLibRunner`] is public because the Python runtime embeds it too: a
8//! daemon that is itself an embedded Python process routes *native* operators to
9//! `python -uc "import dora; dora.start_runtime()"`, so the wheel's runtime has
10//! to be able to host them. See `dora-runtime-python`.
11
12use dora_core::{
13    config::NodeId,
14    descriptor::{Descriptor, OperatorDefinition, OperatorSource},
15};
16use dora_node_api::Event;
17use dora_runtime_api::{OperatorRunner, RunnerGuard, RuntimeHandle};
18use eyre::{Context, Result};
19use tokio::sync::oneshot;
20
21mod runner;
22
23/// Runtime process entry point for shared-library operators.
24pub fn main() -> eyre::Result<()> {
25    dora_runtime_api::main(SharedLibRunner)
26}
27
28/// Backend hosting `dora_init_operator`/`dora_on_event` C-ABI operators.
29pub struct SharedLibRunner;
30
31impl OperatorRunner for SharedLibRunner {
32    fn run_operator(
33        &self,
34        node_id: &NodeId,
35        operator: OperatorDefinition,
36        incoming_events: flume::Receiver<Event>,
37        handle: RuntimeHandle,
38        init_done: oneshot::Sender<Result<()>>,
39        _dataflow_descriptor: &Descriptor,
40    ) -> eyre::Result<RunnerGuard> {
41        match &operator.config.source {
42            // The loaded library is handed back as the runner guard so it stays
43            // mapped until the event loop has joined: values whose vtable lives
44            // in this `.so` (an `OperatorEvent::Panic` payload) can still be in
45            // flight, and dropping them after an unload SIGSEGVs.
46            OperatorSource::SharedLibrary(source) => runner::run(
47                node_id,
48                &operator.id,
49                source,
50                handle,
51                incoming_events,
52                init_done,
53            )
54            .wrap_err_with(|| {
55                format!(
56                    "failed to spawn shared library operator for {}",
57                    operator.id
58                )
59            })
60            .map(|library| Some(Box::new(library) as Box<dyn std::any::Any>)),
61            // Unsupported sources must return a descriptive error rather than
62            // `Ok(())` with a silently dropped `init_done` sender, which would
63            // leave the runtime task blocked in `init_done.await` until it fails
64            // with the misleading "the `init_done` channel was closed
65            // unexpectedly" (#2595).
66            OperatorSource::Python(_) => eyre::bail!(
67                "operator `{}` uses a Python source, but this is the shared-library \
68                 runtime; Python operators are spawned by the Python runtime \
69                 (`dora-runtime-python`)",
70                operator.id
71            ),
72            OperatorSource::Wasm(_) => eyre::bail!(
73                "operator `{}` uses a WASM source, which is not supported yet",
74                operator.id
75            ),
76        }
77    }
78}
79
80#[cfg(test)]
81mod tests {
82    use super::*;
83    use dora_runtime_api::SharedAllocator;
84
85    /// Drives `run_operator` for a source this backend cannot host and returns
86    /// the error plus the still-unsignalled `init_done` receiver.
87    fn run_unsupported(yaml: &str) -> (eyre::Report, oneshot::Receiver<Result<()>>) {
88        let operator: OperatorDefinition =
89            serde_yaml::from_str(yaml).expect("operator definition parses");
90        let dataflow: Descriptor =
91            serde_yaml::from_str("nodes:\n  - id: a\n").expect("descriptor parses");
92        let (_events_in_tx, incoming_events) = flume::unbounded::<Event>();
93        let (events_tx, _events_rx) = tokio::sync::mpsc::channel(1);
94        let (init_done_tx, init_done_rx) = oneshot::channel();
95
96        let err = SharedLibRunner
97            .run_operator(
98                &NodeId::from("node".to_string()),
99                operator,
100                incoming_events,
101                RuntimeHandle::new(events_tx, SharedAllocator::default()),
102                init_done_tx,
103                &dataflow,
104            )
105            .expect_err("unsupported operator source must return an error");
106        (err, init_done_rx)
107    }
108
109    /// An unsupported operator source must surface a descriptive error from
110    /// `run_operator` rather than returning `Ok(())` while silently dropping the
111    /// `init_done` sender — which would leave the runtime task blocked in
112    /// `init_done.await` until it fails with the misleading "the `init_done`
113    /// channel was closed unexpectedly".
114    #[test]
115    fn wasm_source_returns_descriptive_error() {
116        let (err, mut init_done_rx) = run_unsupported("id: op\nwasm: model.wasm\n");
117        assert!(
118            err.to_string().contains("WASM"),
119            "expected a descriptive WASM error, got: {err}"
120        );
121        assert!(
122            init_done_rx.try_recv().is_err(),
123            "init_done must not receive a value for an unsupported source"
124        );
125    }
126
127    /// The cross-language arm this split introduces: a Python operator reaching
128    /// the shared-library runtime is a routing bug, and it must fail the same
129    /// way — descriptive error, `init_done` left unsignalled — rather than hang
130    /// the runtime task.
131    #[test]
132    fn python_source_returns_descriptive_error() {
133        let (err, mut init_done_rx) = run_unsupported("id: op\npython: op.py\n");
134        let msg = err.to_string();
135        assert!(
136            msg.contains("Python") && msg.contains("shared-library runtime"),
137            "expected an error naming the wrong runtime, got: {err}"
138        );
139        assert!(
140            init_done_rx.try_recv().is_err(),
141            "init_done must not receive a value for a wrongly routed source"
142        );
143    }
144}
dora_runtime_shared_lib/lib.rs

dora_runtime_shared_lib/
lib.rs
