fix(engine): use flume async API in shared MPMC channel

rust/otap-dataflow/.chloggen/engine-flume-mpmc-async-blocking.yaml

@@ -0,0 +1,23 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. engine, otap).
+# Must be one of the components listed in rust/otap-dataflow/.chloggen/config.yaml.
+component: engine
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Use flume's async interface in the shared MPMC channel so `SharedSender::send` and `SharedReceiver::recv` no longer block the Tokio worker thread
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [1704]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  The MPMC variant of the shared channel called flume's synchronous `send()`/`recv()`
+  from inside `async fn`, which parked the underlying async runtime thread instead of
+  yielding cooperatively. These now use `send_async().await` / `recv_async().await`,
+  matching the local channel abstraction.

rust/otap-dataflow/crates/engine/src/shared/message.rs

@@ -97,7 +97,10 @@ impl<T> SharedSender<T> {
             SharedSenderInner::Mpsc(sender) => {
                 sender.send(msg).await.map_err(|e| SendError::Closed(e.0))
             }
-            SharedSenderInner::Mpmc(sender) => sender.send(msg).map_err(|e| SendError::Closed(e.0)),
+            SharedSenderInner::Mpmc(sender) => sender
+                .send_async(msg)
+                .await
+                .map_err(|e| SendError::Closed(e.0)),
         };
 
         if let Some(metrics) = &self.metrics {
@@ -217,7 +220,9 @@ impl<T> SharedReceiver<T> {
     pub async fn recv(&mut self) -> Result<T, RecvError> {
         let result = match &mut self.inner {
             SharedReceiverInner::Mpsc(receiver) => receiver.recv().await.ok_or(RecvError::Closed),
-            SharedReceiverInner::Mpmc(receiver) => receiver.recv().map_err(|_| RecvError::Closed),
+            SharedReceiverInner::Mpmc(receiver) => {
+                receiver.recv_async().await.map_err(|_| RecvError::Closed)
+            }
         };
 
         if let Some(metrics) = &self.metrics {
@@ -327,4 +332,46 @@ mod tests {
             "expected Closed, got {result:?}"
         );
     }
+
+    /// Regression test for https://github.com/open-telemetry/otel-arrow/issues/1704.
+    #[test]
+    fn test_mpmc_recv_send_do_not_block_runtime_thread() {
+        use std::sync::mpsc;
+        use std::time::Duration;
+
+        let (done_tx, done_rx) = mpsc::channel();
+
+        let worker = std::thread::spawn(move || {
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .build()
+                .expect("failed to build current-thread runtime");
+
+            let outcome = rt.block_on(async {
+                let (tx, rx) = flume::bounded::<String>(1);
+                let mut receiver = SharedReceiver::mpmc(rx);
+                let sender = SharedSender::mpmc(tx);
+
+                // On a single-threaded runtime `recv` is polled first and must
+                // yield (Pending) so `send` can make progress on the same thread.
+                // A blocking flume `recv()` would deadlock here.
+                tokio::join!(async { receiver.recv().await }, async {
+                    sender.send("hello".to_owned()).await
+                },)
+            });
+
+            let _ = done_tx.send(outcome);
+        });
+
+        match done_rx.recv_timeout(Duration::from_secs(5)) {
+            Ok((received, sent)) => {
+                worker.join().expect("worker thread panicked");
+                assert!(sent.is_ok(), "send failed: {sent:?}");
+                assert_eq!(received.expect("recv failed"), "hello");
+            }
+            Err(_) => panic!(
+                "shared MPMC recv()/send() blocked the runtime thread; flume's \
+                 sync interface must not be used inside async fns (issue #1704)"
+            ),
+        }
+    }
 }