From 94fa6810689219e797be681634db09cb5eb4aee8 Mon Sep 17 00:00:00 2001 From: Raphael Date: Wed, 26 Aug 2026 13:24:14 +0200 Subject: [PATCH 1/2] feat: implemented conccurent message handling --- rust/hercules/src/action.rs | 25 +++- rust/hercules/src/connected.rs | 243 +++++++++++++++++++++++++++----- rust/hercules/src/connection.rs | 14 +- rust/hercules/src/error.rs | 9 ++ rust/hercules/src/lib.rs | 4 +- 5 files changed, 253 insertions(+), 42 deletions(-) diff --git a/rust/hercules/src/action.rs b/rust/hercules/src/action.rs index 29d2837..db9f2af 100644 --- a/rust/hercules/src/action.rs +++ b/rust/hercules/src/action.rs @@ -22,6 +22,8 @@ use crate::registration; use crate::types::{ConfigurationDefinition, ScalingOption, Translation}; const EVENT_CHANNEL_CAPACITY: usize = 256; +/// Default maximum number of outbound requests buffered per connection. +pub const DEFAULT_REQUEST_QUEUE_CAPACITY: usize = 256; /// An action under construction: `#[hercules_sdk::runtime_function]` and friends /// register themselves automatically (see [`crate::registration`]), so most @@ -41,6 +43,7 @@ pub struct Action { version: String, aquila_url: Option, scaling_option: ScalingOption, + request_queue_capacity: usize, author: String, icon: String, documentation: String, @@ -68,6 +71,7 @@ impl Action { version: version.into(), aquila_url: None, scaling_option: ScalingOption::default(), + request_queue_capacity: DEFAULT_REQUEST_QUEUE_CAPACITY, author: String::new(), icon: String::new(), documentation: String::new(), @@ -97,6 +101,15 @@ impl Action { self } + /// Maximum number of outbound requests waiting to be written to Aquila. + /// Once full, new submissions fail with [`crate::HerculesError::Overloaded`]. + /// Defaults to [`DEFAULT_REQUEST_QUEUE_CAPACITY`]. A zero capacity is + /// rejected by [`Action::connect`]. + pub fn request_queue_capacity(mut self, capacity: usize) -> Self { + self.request_queue_capacity = capacity; + self + } + pub fn author(mut self, author: impl Into) -> Self { self.author = author.into(); self @@ -247,8 +260,14 @@ impl Action { .or(self.aquila_url) .ok_or(crate::error::HerculesError::MissingAquilaUrl)?; - let connection = - connection::connect(module, self.scaling_option, &auth_token.into(), &url).await?; + let connection = connection::connect( + module, + self.scaling_option, + &auth_token.into(), + &url, + self.request_queue_capacity, + ) + .await?; let inner = Arc::new(ConnectedInner { identifier: self.identifier, @@ -258,7 +277,9 @@ impl Action { flows: Default::default(), pending_flow_executions: Default::default(), pending_sub_flow_executions: Default::default(), + next_pending_token: Default::default(), request_tx: connection.request_tx, + queue_saturation_count: Default::default(), events_tx: self.events_tx, }); diff --git a/rust/hercules/src/connected.rs b/rust/hercules/src/connected.rs index 3a82899..5e7e1bb 100644 --- a/rust/hercules/src/connected.rs +++ b/rust/hercules/src/connected.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex, RwLock}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -29,6 +30,27 @@ pub(crate) struct RuntimeFunctionEntry { pub handler: Arc, } +pub(crate) struct PendingExecution { + token: u64, + sender: oneshot::Sender>, +} + +/// Point-in-time health counters for the outbound Aquila request path. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct ConnectedMetrics { + /// Requests currently buffered, excluding one currently being consumed + /// by the transport. + pub queue_depth: usize, + /// Configured upper bound for `queue_depth`. + pub queue_capacity: usize, + /// Number of submissions rejected because the queue was full. + pub queue_saturation_count: u64, + /// Whole-flow executions currently awaiting a response from Aquila. + pub pending_flow_executions: usize, + /// Sub-flow executions currently awaiting a response from Aquila. + pub pending_sub_flow_executions: usize, +} + pub(crate) struct ConnectedInner { pub identifier: String, pub version: String, @@ -41,7 +63,7 @@ pub(crate) struct ConnectedInner { /// [`Connected::execute_flow`]) and is still awaiting a result for, /// keyed by the `execution_identifier` that correlates the eventual /// `ActionFlowExecutionResponse`. - pub pending_flow_executions: Mutex>>>, + pub(crate) pending_flow_executions: Mutex>, /// Sub flow executions this action has requested (via /// [`Connected::execute_sub_flow`]) and is still awaiting a result for, /// keyed by a locally-generated `correlation_identifier` — not the sub @@ -50,8 +72,10 @@ pub(crate) struct ConnectedInner { /// iteration), so it cannot disambiguate concurrent or out-of-order /// invocations the way the correlation identifier can. See /// [`crate::types::FunctionContext::sub_flow_parameters`]. - pub pending_sub_flow_executions: Mutex>>>, - pub request_tx: mpsc::UnboundedSender, + pub(crate) pending_sub_flow_executions: Mutex>, + pub next_pending_token: AtomicU64, + pub request_tx: mpsc::Sender, + pub queue_saturation_count: AtomicU64, pub events_tx: broadcast::Sender, } @@ -67,6 +91,31 @@ impl ConnectedInner { fn next_execution_id(&self) -> String { uuid::Uuid::new_v4().to_string() } + + fn next_pending_token(&self) -> u64 { + self.next_pending_token.fetch_add(1, Ordering::Relaxed) + } +} + +/// Removes an unresolved pending entry when its caller's future is dropped. +/// The token prevents an old future from deleting a newer entry that happens +/// to reuse the same external identifier. +struct PendingExecutionGuard<'a> { + pending: &'a Mutex>, + identifier: String, + token: u64, +} + +impl Drop for PendingExecutionGuard<'_> { + fn drop(&mut self) { + let mut pending = sync::lock(self.pending); + if pending + .get(&self.identifier) + .is_some_and(|entry| entry.token == self.token) + { + pending.remove(&self.identifier); + } + } } /// A live, connected action. Cheap to clone (an `Arc` underneath) and safe to @@ -103,6 +152,20 @@ impl Connected { event_stream(&self.inner.events_tx) } + /// Returns a point-in-time snapshot of outbound queue pressure and + /// outstanding executions. The saturation counter is cumulative for the + /// lifetime of this connection. + pub fn metrics(&self) -> ConnectedMetrics { + let queue_capacity = self.inner.request_tx.max_capacity(); + ConnectedMetrics { + queue_depth: queue_capacity - self.inner.request_tx.capacity(), + queue_capacity, + queue_saturation_count: self.inner.queue_saturation_count.load(Ordering::Relaxed), + pending_flow_executions: sync::lock(&self.inner.pending_flow_executions).len(), + pending_sub_flow_executions: sync::lock(&self.inner.pending_sub_flow_executions).len(), + } + } + /// The configuration Aquila has resolved for `project_id`, if any has /// been pushed down yet. pub fn config(&self, project_id: i64) -> Option { @@ -167,6 +230,21 @@ impl Connected { &self, flow_id: impl Into, payload: PlainValue, + ) -> Result { + self.try_execute_flow(flow_id, payload).await + } + + /// Attempts to enqueue a flow execution without waiting for outbound + /// queue capacity. Returns [`HerculesError::Overloaded`] immediately when + /// the configured queue is full. + /// + /// [`Connected::execute_flow`] has the same bounded admission semantics; + /// this explicitly named form is useful at request boundaries where an + /// overload response is part of the caller-facing contract. + pub async fn try_execute_flow( + &self, + flow_id: impl Into, + payload: PlainValue, ) -> Result { let execution_identifier = self.inner.next_execution_id(); self.execute_flow_with_id(execution_identifier, flow_id, payload) @@ -194,7 +272,16 @@ impl Connected { payload: PlainValue, ) -> Result { let (tx, rx) = oneshot::channel(); - sync::lock(&self.inner.pending_flow_executions).insert(execution_identifier.clone(), tx); + let token = self.inner.next_pending_token(); + sync::lock(&self.inner.pending_flow_executions).insert( + execution_identifier.clone(), + PendingExecution { token, sender: tx }, + ); + let _pending_guard = PendingExecutionGuard { + pending: &self.inner.pending_flow_executions, + identifier: execution_identifier.clone(), + token, + }; let request = ActionTransferRequest { data: Some(action_transfer_request::Data::FlowExecution( @@ -205,10 +292,7 @@ impl Connected { }, )), }; - if let Err(err) = send(&self.inner, request) { - sync::lock(&self.inner.pending_flow_executions).remove(&execution_identifier); - return Err(err); - } + send(&self.inner, request)?; rx.await.map_err(|_| HerculesError::StreamClosed)? } @@ -230,8 +314,16 @@ impl Connected { // repeatedly and responses arrive out of order. let correlation_identifier = self.inner.next_execution_id(); let (tx, rx) = oneshot::channel(); - sync::lock(&self.inner.pending_sub_flow_executions) - .insert(correlation_identifier.clone(), tx); + let token = self.inner.next_pending_token(); + sync::lock(&self.inner.pending_sub_flow_executions).insert( + correlation_identifier.clone(), + PendingExecution { token, sender: tx }, + ); + let _pending_guard = PendingExecutionGuard { + pending: &self.inner.pending_sub_flow_executions, + identifier: correlation_identifier.clone(), + token, + }; let request = ActionTransferRequest { data: Some(action_transfer_request::Data::SubFlowExecution( @@ -242,10 +334,7 @@ impl Connected { }, )), }; - if let Err(err) = send(&self.inner, request) { - sync::lock(&self.inner.pending_sub_flow_executions).remove(&correlation_identifier); - return Err(err); - } + send(&self.inner, request)?; rx.await.map_err(|_| HerculesError::StreamClosed)? } @@ -253,10 +342,16 @@ impl Connected { fn send(inner: &ConnectedInner, request: ActionTransferRequest) -> Result<()> { log::trace!("sending {request:?}"); - inner - .request_tx - .send(request) - .map_err(|_| HerculesError::StreamClosed) + match inner.request_tx.try_send(request) { + Ok(()) => Ok(()), + Err(mpsc::error::TrySendError::Full(_)) => { + inner.queue_saturation_count.fetch_add(1, Ordering::Relaxed); + Err(HerculesError::Overloaded { + capacity: inner.request_tx.max_capacity(), + }) + } + Err(mpsc::error::TrySendError::Closed(_)) => Err(HerculesError::StreamClosed), + } } pub(crate) fn spawn_dispatch_loop( @@ -299,9 +394,9 @@ where /// Resolves every outstanding sender in a pending-execution map with /// [`HerculesError::StreamClosed`], used when the response stream has ended /// and nothing will ever answer them. -fn drain_pending(pending: &Mutex>>>) { - for (_, sender) in sync::lock(pending).drain() { - let _ = sender.send(Err(HerculesError::StreamClosed)); +fn drain_pending(pending: &Mutex>) { + for (_, entry) in sync::lock(pending).drain() { + let _ = entry.sender.send(Err(HerculesError::StreamClosed)); } } @@ -541,8 +636,8 @@ fn handle_flow_execution_response( inner: &Arc, response: ActionFlowExecutionResponse, ) { - let sender = sync::lock(&inner.pending_flow_executions).remove(&response.execution_identifier); - let Some(sender) = sender else { + let entry = sync::lock(&inner.pending_flow_executions).remove(&response.execution_identifier); + let Some(entry) = entry else { log::warn!( "received a flow execution response for unknown execution {:?}", response.execution_identifier @@ -559,16 +654,16 @@ fn handle_flow_execution_response( "flow execution response is missing a result".into(), )), }; - let _ = sender.send(outcome); + let _ = entry.sender.send(outcome); } fn handle_sub_flow_execution_response( inner: &Arc, response: ActionSubFlowExecutionResponse, ) { - let sender = + let entry = sync::lock(&inner.pending_sub_flow_executions).remove(&response.correlation_identifier); - let Some(sender) = sender else { + let Some(entry) = entry else { log::warn!( "received a sub flow execution response for unknown correlation id {:?}", response.correlation_identifier @@ -587,7 +682,7 @@ fn handle_sub_flow_execution_response( "sub flow execution response is missing a result".into(), )), }; - let _ = sender.send(outcome); + let _ = entry.sender.send(outcome); } fn wire_error(code: String, message: String, timestamp: i64, version: &str) -> WireError { @@ -624,11 +719,14 @@ mod tests { /// A bare `ConnectedInner` with no runtime functions and a request /// channel the test can inspect, mirroring the shape [`crate::action`] /// builds one with in [`crate::Action::connect`]. - fn test_inner() -> ( - Arc, - mpsc::UnboundedReceiver, - ) { - let (request_tx, request_rx) = mpsc::unbounded_channel(); + fn test_inner() -> (Arc, mpsc::Receiver) { + test_inner_with_capacity(16) + } + + fn test_inner_with_capacity( + capacity: usize, + ) -> (Arc, mpsc::Receiver) { + let (request_tx, request_rx) = mpsc::channel(capacity); let (events_tx, _events_rx) = broadcast::channel(16); let inner = Arc::new(ConnectedInner { identifier: "test-action".into(), @@ -638,7 +736,9 @@ mod tests { flows: Default::default(), pending_flow_executions: Default::default(), pending_sub_flow_executions: Default::default(), + next_pending_token: Default::default(), request_tx, + queue_saturation_count: Default::default(), events_tx, }); (inner, request_rx) @@ -695,9 +795,21 @@ mod tests { let (inner, _request_rx) = test_inner(); let (success_tx, success_rx) = oneshot::channel(); - sync::lock(&inner.pending_sub_flow_executions).insert("id-a".into(), success_tx); + sync::lock(&inner.pending_sub_flow_executions).insert( + "id-a".into(), + PendingExecution { + token: 1, + sender: success_tx, + }, + ); let (failure_tx, failure_rx) = oneshot::channel(); - sync::lock(&inner.pending_sub_flow_executions).insert("id-b".into(), failure_tx); + sync::lock(&inner.pending_sub_flow_executions).insert( + "id-b".into(), + PendingExecution { + token: 2, + sender: failure_tx, + }, + ); handle_sub_flow_execution_response( &inner, @@ -872,4 +984,67 @@ mod tests { .unwrap(); assert!(matches!(result, Err(HerculesError::StreamClosed))); } + + #[tokio::test] + async fn full_request_queue_rejects_flow_without_growing_pending_entries() { + let (inner, _request_rx) = test_inner_with_capacity(1); + let connected = Connected::new(inner.clone()); + + let first = tokio::spawn({ + let connected = connected.clone(); + async move { + connected + .execute_flow_with_id("flow-first".into(), "flow-1", PlainValue::Null) + .await + } + }); + + tokio::time::timeout(Duration::from_secs(2), async { + while connected.metrics().queue_depth != 1 { + tokio::task::yield_now().await; + } + }) + .await + .expect("first request was not queued"); + + let result = connected + .execute_flow_with_id("flow-overload".into(), "flow-2", PlainValue::Null) + .await; + assert!(matches!( + result, + Err(HerculesError::Overloaded { capacity: 1 }) + )); + + let metrics = connected.metrics(); + assert_eq!(metrics.queue_depth, 1); + assert_eq!(metrics.queue_capacity, 1); + assert_eq!(metrics.queue_saturation_count, 1); + assert_eq!(metrics.pending_flow_executions, 1); + + first.abort(); + assert!(first.await.unwrap_err().is_cancelled()); + assert_eq!(connected.metrics().pending_flow_executions, 0); + } + + #[tokio::test] + async fn cancelling_execute_flow_removes_its_pending_entry() { + let (inner, mut request_rx) = test_inner(); + let connected = Connected::new(inner); + + let execution = tokio::spawn({ + let connected = connected.clone(); + async move { + connected + .execute_flow_with_id("flow-cancelled".into(), "flow-1", PlainValue::Null) + .await + } + }); + + request_rx.recv().await.expect("flow request was sent"); + assert_eq!(connected.metrics().pending_flow_executions, 1); + + execution.abort(); + assert!(execution.await.unwrap_err().is_cancelled()); + assert_eq!(connected.metrics().pending_flow_executions, 0); + } } diff --git a/rust/hercules/src/connection.rs b/rust/hercules/src/connection.rs index dbde740..edfadbf 100644 --- a/rust/hercules/src/connection.rs +++ b/rust/hercules/src/connection.rs @@ -2,7 +2,7 @@ //! and sends the initial `ActionLogon` frame. use tokio::sync::mpsc; -use tokio_stream::wrappers::UnboundedReceiverStream; +use tokio_stream::wrappers::ReceiverStream; use tonic::metadata::MetadataValue; use tonic::transport::Endpoint; use tonic::{Request, Streaming}; @@ -16,7 +16,7 @@ use crate::error::{HerculesError, Result}; use crate::types::ScalingOption; pub struct Connection { - pub request_tx: mpsc::UnboundedSender, + pub request_tx: mpsc::Sender, pub responses: Streaming, } @@ -35,13 +35,18 @@ pub async fn connect( scaling_option: ScalingOption, auth_token: &str, aquila_url: &str, + request_queue_capacity: usize, ) -> Result { + if request_queue_capacity == 0 { + return Err(HerculesError::InvalidQueueCapacity); + } + let channel = Endpoint::from_shared(endpoint_uri(aquila_url))? .connect() .await?; let mut client = ActionTransferServiceClient::new(channel); - let (request_tx, request_rx) = mpsc::unbounded_channel::(); + let (request_tx, request_rx) = mpsc::channel::(request_queue_capacity); request_tx .send(ActionTransferRequest { data: Some(action_transfer_request::Data::Logon(ActionLogon { @@ -49,9 +54,10 @@ pub async fn connect( scaling_option: scaling_option.into_wire() as i32, })), }) + .await .map_err(|_| HerculesError::StreamClosed)?; - let mut request = Request::new(UnboundedReceiverStream::new(request_rx)); + let mut request = Request::new(ReceiverStream::new(request_rx)); let token: MetadataValue<_> = auth_token .parse() .map_err(|_| HerculesError::Other("auth token is not valid gRPC metadata".to_string()))?; diff --git a/rust/hercules/src/error.rs b/rust/hercules/src/error.rs index a5059aa..e36641f 100644 --- a/rust/hercules/src/error.rs +++ b/rust/hercules/src/error.rs @@ -38,6 +38,15 @@ pub enum HerculesError { #[error("failed to send on the Aquila request stream")] StreamClosed, + /// The bounded outbound request queue has no remaining capacity. Callers + /// can retry later instead of allowing queued requests to consume memory + /// without a limit. + #[error("Aquila request queue is full (capacity {capacity})")] + Overloaded { capacity: usize }, + + #[error("Aquila request queue capacity must be greater than zero")] + InvalidQueueCapacity, + #[error( "cannot generate a type string for data type {identifier:?}: it contains a recursive \ schema that can't be inlined. Register the nested schema as its own data type so it \ diff --git a/rust/hercules/src/lib.rs b/rust/hercules/src/lib.rs index deda660..6f12857 100644 --- a/rust/hercules/src/lib.rs +++ b/rust/hercules/src/lib.rs @@ -44,9 +44,9 @@ mod sync; mod types; mod value; -pub use action::Action; +pub use action::{Action, DEFAULT_REQUEST_QUEUE_CAPACITY}; pub use arguments::Arguments; -pub use connected::Connected; +pub use connected::{Connected, ConnectedMetrics}; pub use data_type::{DataType, DataTypeDef, DataTypeRule}; pub use error::{HerculesError, Result}; pub use event::{Event, RuntimeEvent}; From 866afba8763be60b0a2558cf0f49373758764ef2 Mon Sep 17 00:00:00 2001 From: Raphael Date: Wed, 26 Aug 2026 14:44:06 +0200 Subject: [PATCH 2/2] fix: let mandatory execution results wait for outbound queue capacity Runtime function results are protocol traffic, not optional caller-initiated requests, so they must not be dropped when the outbound queue is full. try_send_caller_request keeps the fail-fast admission behavior for execute_flow/execute_sub_flow; send_mandatory instead awaits capacity for execution results, so a full queue slows result delivery rather than silently discarding it. Co-Authored-By: Claude Sonnet 5 --- rust/hercules/src/connected.rs | 107 ++++++++++++++++++++++++++++++--- 1 file changed, 99 insertions(+), 8 deletions(-) diff --git a/rust/hercules/src/connected.rs b/rust/hercules/src/connected.rs index 5e7e1bb..256ec1a 100644 --- a/rust/hercules/src/connected.rs +++ b/rust/hercules/src/connected.rs @@ -43,7 +43,8 @@ pub struct ConnectedMetrics { pub queue_depth: usize, /// Configured upper bound for `queue_depth`. pub queue_capacity: usize, - /// Number of submissions rejected because the queue was full. + /// Number of outbound submissions that encountered a full queue. Caller + /// requests are rejected; mandatory execution results wait for capacity. pub queue_saturation_count: u64, /// Whole-flow executions currently awaiting a response from Aquila. pub pending_flow_executions: usize, @@ -292,7 +293,7 @@ impl Connected { }, )), }; - send(&self.inner, request)?; + try_send_caller_request(&self.inner, request)?; rx.await.map_err(|_| HerculesError::StreamClosed)? } @@ -334,13 +335,15 @@ impl Connected { }, )), }; - send(&self.inner, request)?; + try_send_caller_request(&self.inner, request)?; rx.await.map_err(|_| HerculesError::StreamClosed)? } } -fn send(inner: &ConnectedInner, request: ActionTransferRequest) -> Result<()> { +/// Caller-initiated work is optional admission: fail fast so an overloaded +/// request boundary can shed load instead of adding another waiter. +fn try_send_caller_request(inner: &ConnectedInner, request: ActionTransferRequest) -> Result<()> { log::trace!("sending {request:?}"); match inner.request_tx.try_send(request) { Ok(()) => Ok(()), @@ -354,6 +357,21 @@ fn send(inner: &ConnectedInner, request: ActionTransferRequest) -> Result<()> { } } +/// Runtime execution results are mandatory protocol traffic. They share the +/// same bounded queue, but wait for capacity instead of being discarded when +/// caller-initiated traffic has filled it. +async fn send_mandatory(inner: &ConnectedInner, request: ActionTransferRequest) -> Result<()> { + log::trace!("sending mandatory {request:?}"); + if inner.request_tx.capacity() == 0 { + inner.queue_saturation_count.fetch_add(1, Ordering::Relaxed); + } + inner + .request_tx + .send(request) + .await + .map_err(|_| HerculesError::StreamClosed) +} + pub(crate) fn spawn_dispatch_loop( inner: Arc, responses: Streaming, @@ -536,10 +554,11 @@ async fn handle_execution(inner: Arc, execution: ActionExecution }, )), }; - if send(&inner, request).is_err() { + if let Err(err) = send_mandatory(&inner, request).await { log::error!( - "failed to send execution result for {:?}: stream closed", - execution_identifier + "failed to send execution result for {:?}: {}", + execution_identifier, + err ); } } @@ -716,6 +735,15 @@ mod tests { use super::*; + struct FixedResultHandler; + + #[async_trait::async_trait] + impl RuntimeFunctionHandler for FixedResultHandler { + async fn run(&self, _context: &FunctionContext, _args: &Arguments) -> Result { + Ok(serde_json::json!("completed")) + } + } + /// A bare `ConnectedInner` with no runtime functions and a request /// channel the test can inspect, mirroring the shape [`crate::action`] /// builds one with in [`crate::Action::connect`]. @@ -725,13 +753,20 @@ mod tests { fn test_inner_with_capacity( capacity: usize, + ) -> (Arc, mpsc::Receiver) { + test_inner_with_runtime_functions(capacity, HashMap::new()) + } + + fn test_inner_with_runtime_functions( + capacity: usize, + runtime_functions: HashMap, ) -> (Arc, mpsc::Receiver) { let (request_tx, request_rx) = mpsc::channel(capacity); let (events_tx, _events_rx) = broadcast::channel(16); let inner = Arc::new(ConnectedInner { identifier: "test-action".into(), version: "0.0.0".into(), - runtime_functions: HashMap::new(), + runtime_functions, configs: Default::default(), flows: Default::default(), pending_flow_executions: Default::default(), @@ -1026,6 +1061,62 @@ mod tests { assert_eq!(connected.metrics().pending_flow_executions, 0); } + #[tokio::test] + async fn mandatory_execution_result_waits_for_capacity_and_is_transmitted() { + let runtime_functions = HashMap::from([( + "fixed-result".to_string(), + RuntimeFunctionEntry { + meta: RuntimeFunctionMeta { + identifier: "fixed-result".into(), + ..Default::default() + }, + handler: Arc::new(FixedResultHandler), + }, + )]); + let (inner, mut request_rx) = test_inner_with_runtime_functions(1, runtime_functions); + + inner + .request_tx + .try_send(ActionTransferRequest { data: None }) + .expect("test queue should start empty"); + + let execution = tokio::spawn(handle_execution( + inner.clone(), + ActionExecutionRequest { + execution_identifier: "execution-result".into(), + function_identifier: "fixed-result".into(), + parameters: vec![], + project_id: 42, + }, + )); + + tokio::time::timeout(Duration::from_secs(2), async { + while inner.queue_saturation_count.load(Ordering::Relaxed) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("execution result did not wait on the full queue"); + assert!(!execution.is_finished()); + + let placeholder = request_rx.recv().await.expect("placeholder was queued"); + assert!(placeholder.data.is_none()); + + let transmitted = tokio::time::timeout(Duration::from_secs(2), request_rx.recv()) + .await + .expect("execution result did not acquire released queue capacity") + .expect("request channel closed before the result was transmitted"); + match transmitted.data { + Some(action_transfer_request::Data::Result(response)) => { + assert_eq!(response.execution_identifier, "execution-result"); + assert!(response.node_result.is_some()); + } + other => panic!("expected an execution result, got {other:?}"), + } + + execution.await.expect("execution task panicked"); + } + #[tokio::test] async fn cancelling_execute_flow_removes_its_pending_entry() { let (inner, mut request_rx) = test_inner();