aion/activity/
bridge.rs

1//! Activity dispatch bridge for the `aion_flow_ffi` activity NIFs.
2//!
3//! The bridge decouples the raw NIF function pointer (which cannot capture
4//! state) from the engine's activity execution path. A concrete dispatcher
5//! is installed on the engine's NIF state during build; the NIF recovers it
6//! from its calling context's NIF private data.
7
8use std::sync::Arc;
9
10use futures::future::BoxFuture;
11
12/// Executes an activity request originating from workflow code.
13///
14/// Implementations receive the activity name, JSON-encoded input, and
15/// JSON-encoded config as strings — the same wire format that the Gleam SDK
16/// sends through the `aion_flow_ffi:dispatch_activity/3` binding — plus the
17/// one-based delivery attempt stamped by the engine. The return value is the
18/// JSON-encoded activity result or a prefixed error string matching the
19/// SDK's error-decoding convention.
20pub trait ActivityDispatcher: Send + Sync + 'static {
21    /// Dispatch the named activity and block until completion.
22    ///
23    /// `attempt` is the one-based delivery attempt for this dispatch; the
24    /// dispatcher stamps it onto whatever transport carries the task (the
25    /// worker wire's `ActivityTask.attempt`), so consumers can distinguish
26    /// retries without guessing.
27    ///
28    /// Returns `Ok(encoded_output)` on success or `Err(error_string)` on
29    /// failure. Both sides are strings matching the Gleam SDK's
30    /// `Result(String, String)` FFI contract.
31    ///
32    /// # Errors
33    ///
34    /// Returns the error string surfaced by the activity execution path —
35    /// worker rejection, decode failure, timeout, or activity body error.
36    fn dispatch(
37        &self,
38        name: &str,
39        input: &str,
40        config: &str,
41        attempt: u32,
42    ) -> Result<String, String>;
43
44    /// Dispatch the named activity with the calling workflow process id when
45    /// the runtime can provide it.
46    ///
47    /// Implementations that need to correlate a raw NIF call back to an active
48    /// workflow handle can override this method. The default preserves the
49    /// original dispatcher contract for tests and non-workflow callers.
50    ///
51    /// # Errors
52    ///
53    /// Returns the same errors as [`Self::dispatch`].
54    fn dispatch_from_process(
55        &self,
56        name: &str,
57        input: &str,
58        config: &str,
59        attempt: u32,
60        caller_pid: Option<u64>,
61    ) -> Result<String, String> {
62        let _ = caller_pid;
63        self.dispatch(name, input, config, attempt)
64    }
65
66    /// Dispatch the named activity from a Tokio task.
67    ///
68    /// The default runs the synchronous [`Self::dispatch_from_process`] on
69    /// the runtime's blocking pool via [`tokio::task::spawn_blocking`], so a
70    /// dispatcher implementation that blocks its calling thread cannot wedge
71    /// the engine's async workers (a single-threaded engine runtime keeps
72    /// servicing queries and completions while the dispatch waits).
73    /// Nonblocking dispatchers can override this method directly.
74    ///
75    /// Must be awaited inside a Tokio runtime context; the engine's
76    /// completion task guarantees that.
77    ///
78    /// # Errors
79    ///
80    /// Returns the same errors as [`Self::dispatch_from_process`], plus a
81    /// dispatch-failure reason when the blocking task itself is cancelled or
82    /// panics.
83    fn dispatch_async_from_process(
84        self: Arc<Self>,
85        name: String,
86        input: String,
87        config: String,
88        attempt: u32,
89        caller_pid: Option<u64>,
90    ) -> BoxFuture<'static, Result<String, String>> {
91        Box::pin(async move {
92            let blocking = tokio::task::spawn_blocking(move || {
93                self.dispatch_from_process(&name, &input, &config, attempt, caller_pid)
94            });
95            match blocking.await {
96                Ok(result) => result,
97                Err(join_error) => Err(format!("activity dispatch task failed: {join_error}")),
98            }
99        })
100    }
101}
102
103#[cfg(test)]
104mod tests {
105    use std::sync::Arc;
106
107    use super::ActivityDispatcher;
108    use crate::runtime::EngineNifState;
109
110    struct Echo;
111
112    impl ActivityDispatcher for Echo {
113        fn dispatch(
114            &self,
115            _name: &str,
116            input: &str,
117            _config: &str,
118            _attempt: u32,
119        ) -> Result<String, String> {
120            Ok(input.to_owned())
121        }
122    }
123
124    #[test]
125    fn dispatcher_is_accessible_after_install_on_engine_state() {
126        let state = EngineNifState::default();
127        state.set_activity_dispatcher(Arc::new(Echo));
128        let dispatcher = state.activity_dispatcher();
129        assert!(dispatcher.is_some());
130        assert_eq!(
131            dispatcher
132                .as_ref()
133                .and_then(|d| d.dispatch("test", "hello", "{}", 1).ok()),
134            Some("hello".to_owned())
135        );
136    }
137}
aion/activity/bridge.rs

aion/activity/
bridge.rs
