fix: release lock guards before awaiting channel send (#869) (#1003)

* fix: release lock guards before awaiting channel send (#869)

Clone `mpsc::Sender` out of `RwLock` before `.send().await` to prevent
read guards from blocking write lock acquisition (shutdown/start) when
the channel buffer is full.

Fixed call sites:
- src/channels/http.rs: process_message()
- src/channels/web/server.rs: chat_send_handler(), chat_approval_handler()
- src/channels/web/handlers/chat.rs: chat_send_handler(), chat_approval_handler()
- src/channels/web/ws.rs: handle_client_message() (2 sites)
- src/channels/wasm/wrapper.rs: process_emitted_messages() (2 impls, also
  scoped rate_limiter write lock per-iteration)

Includes regression test: shutdown_completes_while_process_message_blocked

Co-Authored-By: Claude Opus 4.6 <[email protected]>
(cherry picked from commit 84802e1b89aaf07ba976db20bdbfdf749edbe332)

* ci: fetch base branch before regression test check

The regression-test-check workflow failed because origin/main wasn't
available as a ref in the CI environment. actions/checkout@v4 fetches
the PR merge ref history but doesn't make the base branch ref available
for three-dot diff comparisons.

Co-Authored-By: Claude Opus 4.6 <[email protected]>
(cherry picked from commit 1d5a7bdc8ec071cdddf0e69d6053d49ca20a2b18)

* chore(ci): rerun regression gate [skip-regression-check]

(cherry picked from commit 784d444701471a1311b1f985e2f07be6f0527abf)

---------

Co-authored-by: Umesh Kumar Singh <[email protected]>
Co-authored-by: Claude Opus 4.6 <[email protected]>

src/channels/http.rs

@@ -807,6 +807,67 @@ mod tests {
         assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
     }
 
+    /// Regression test for issue #869: RwLock read guard was held across
+    /// tx.send(msg).await in `process_message()`, blocking shutdown() from
+    /// acquiring the write lock when the channel buffer was full.
+    ///
+    /// This test exercises the actual production code path (`process_message`)
+    /// with a full channel buffer, then verifies shutdown() can still complete.
+    #[tokio::test]
+    async fn shutdown_completes_while_process_message_blocked() {
+        let channel = Arc::new(test_channel(Some("secret")));
+        let stream = channel.start().await.unwrap();
+
+        // Fill all 256 slots in the channel buffer
+        {
+            let tx = {
+                let guard = channel.state.tx.read().await;
+                guard.as_ref().unwrap().clone()
+            };
+            for i in 0..256 {
+                let msg = IncomingMessage::new("http", "user", format!("fill-{}", i));
+                tx.send(msg).await.unwrap();
+            }
+        }
+
+        // Signal so we know the spawned task has started and is about to
+        // call process_message (which will block on the full channel).
+        let started = Arc::new(tokio::sync::Notify::new());
+        let started_clone = started.clone();
+
+        // Spawn a task that calls the actual production code path.
+        // process_message() internally acquires the RwLock read guard and
+        // sends on the channel. With the fix, the guard is released before
+        // send().await; without the fix, shutdown() would deadlock.
+        let state = channel.state.clone();
+        let blocked_send = tokio::spawn(async move {
+            started_clone.notify_one();
+            let msg = IncomingMessage::new("http", "user", "blocked-257th");
+            let _ = process_message(state, msg, false).await;
+        });
+
+        // Wait for the spawned task to start, then give it time to reach
+        // the send().await and verify that it is still pending (i.e., blocked).
+        started.notified().await;
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        assert!(
+            !blocked_send.is_finished(),
+            "process_message task should still be pending before shutdown()"
+        );
+
+        // shutdown() must complete even though process_message is blocked on
+        // send(). Before the fix, the read guard held across send().await
+        // would prevent shutdown() from acquiring the write lock.
+        let result =
+            tokio::time::timeout(std::time::Duration::from_secs(2), channel.shutdown()).await;
+        assert!(result.is_ok(), "shutdown() must not deadlock");
+        assert!(result.unwrap().is_ok());
+
+        // Drop the stream (receiver) so the blocked send task can complete
+        drop(stream);
+        let _ = blocked_send.await;
+    }
+
     #[tokio::test]
     async fn webhook_missing_all_auth_returns_unauthorized() {
         let channel = test_channel(Some("correct-secret"));

src/channels/wasm/wrapper.rs

@@ -1994,28 +1994,33 @@ impl WasmChannel {
             return Ok(());
         }
 
-        let tx_guard = self.message_tx.read().await;
-        let Some(tx) = tx_guard.as_ref() else {
-            tracing::error!(
-                channel = %self.name,
-                count = messages.len(),
-                "Messages emitted but no sender available - channel may not be started!"
-            );
-            return Ok(());
+        // Clone sender to avoid holding RwLock read guard across send().await in the loop
+        let tx = {
+            let tx_guard = self.message_tx.read().await;
+            let Some(tx) = tx_guard.as_ref() else {
+                tracing::error!(
+                    channel = %self.name,
+                    count = messages.len(),
+                    "Messages emitted but no sender available - channel may not be started!"
+                );
+                return Ok(());
+            };
+            tx.clone()
         };
 
-        let mut rate_limiter = self.rate_limiter.write().await;
-
         for emitted in messages {
-            // Check rate limit
-            if !rate_limiter.check_and_record() {
-                tracing::warn!(
-                    channel = %self.name,
-                    "Message emission rate limited"
-                );
-                return Err(WasmChannelError::EmitRateLimited {
-                    name: self.name.clone(),
-                });
+            // Check rate limit — acquire and release the write lock before send().await
+            {
+                let mut rate_limiter = self.rate_limiter.write().await;
+                if !rate_limiter.check_and_record() {
+                    tracing::warn!(
+                        channel = %self.name,
+                        "Message emission rate limited"
+                    );
+                    return Err(WasmChannelError::EmitRateLimited {
+                        name: self.name.clone(),
+                    });
+                }
             }
 
             // Convert to IncomingMessage
@@ -2057,7 +2062,7 @@ impl WasmChannel {
                 self.update_broadcast_metadata(&emitted.metadata_json).await;
             }
 
-            // Send to stream
+            // Send to stream — no locks held across this await
             tracing::info!(
                 channel = %self.name,
                 user_id = %emitted.user_id,
@@ -2281,28 +2286,33 @@ impl WasmChannel {
             "Processing emitted messages from polling callback"
         );
 
-        let tx_guard = message_tx.read().await;
-        let Some(tx) = tx_guard.as_ref() else {
-            tracing::error!(
-                channel = %channel_name,
-                count = messages.len(),
-                "Messages emitted but no sender available - channel may not be started!"
-            );
-            return Ok(());
+        // Clone sender to avoid holding RwLock read guard across send().await in the loop
+        let tx = {
+            let tx_guard = message_tx.read().await;
+            let Some(tx) = tx_guard.as_ref() else {
+                tracing::error!(
+                    channel = %channel_name,
+                    count = messages.len(),
+                    "Messages emitted but no sender available - channel may not be started!"
+                );
+                return Ok(());
+            };
+            tx.clone()
         };
 
-        let mut limiter = rate_limiter.write().await;
-
         for emitted in messages {
-            // Check rate limit
-            if !limiter.check_and_record() {
-                tracing::warn!(
-                    channel = %channel_name,
-                    "Message emission rate limited"
-                );
-                return Err(WasmChannelError::EmitRateLimited {
-                    name: channel_name.to_string(),
-                });
+            // Check rate limit — acquire and release the write lock before send().await
+            {
+                let mut limiter = rate_limiter.write().await;
+                if !limiter.check_and_record() {
+                    tracing::warn!(
+                        channel = %channel_name,
+                        "Message emission rate limited"
+                    );
+                    return Err(WasmChannelError::EmitRateLimited {
+                        name: channel_name.to_string(),
+                    });
+                }
             }
 
             // Convert to IncomingMessage
@@ -2350,7 +2360,7 @@ impl WasmChannel {
                 .await;
             }
 
-            // Send to stream
+            // Send to stream — no locks held across this await
             tracing::info!(
                 channel = %channel_name,
                 user_id = %emitted.user_id,

src/channels/web/handlers/chat.rs

@@ -37,11 +37,17 @@ pub async fn chat_send_handler(
     let msg_id = msg.id;
     let thread_id = msg.thread_id.clone();
 
-    let tx_guard = state.msg_tx.read().await;
-    let tx = tx_guard.as_ref().ok_or((
-        StatusCode::SERVICE_UNAVAILABLE,
-        "Channel not started".to_string(),
-    ))?;
+    // Clone sender to avoid holding RwLock read guard across send().await
+    let tx = {
+        let tx_guard = state.msg_tx.read().await;
+        tx_guard
+            .as_ref()
+            .ok_or((
+                StatusCode::SERVICE_UNAVAILABLE,
+                "Channel not started".to_string(),
+            ))?
+            .clone()
+    };
 
     tx.send(msg).await.map_err(|_| {
         (
@@ -111,11 +117,17 @@ pub async fn chat_approval_handler(
 
     let msg_id = msg.id;
 
-    let tx_guard = state.msg_tx.read().await;
-    let tx = tx_guard.as_ref().ok_or((
-        StatusCode::SERVICE_UNAVAILABLE,
-        "Channel not started".to_string(),
-    ))?;
+    // Clone sender to avoid holding RwLock read guard across send().await
+    let tx = {
+        let tx_guard = state.msg_tx.read().await;
+        tx_guard
+            .as_ref()
+            .ok_or((
+                StatusCode::SERVICE_UNAVAILABLE,
+                "Channel not started".to_string(),
+            ))?
+            .clone()
+    };
 
     tx.send(msg).await.map_err(|_| {
         (

src/channels/web/server.rs

@@ -973,11 +973,17 @@ async fn chat_send_handler(
         req.images.len()
     );
 
-    let tx_guard = state.msg_tx.read().await;
-    let tx = tx_guard.as_ref().ok_or((
-        StatusCode::SERVICE_UNAVAILABLE,
-        "Channel not started".to_string(),
-    ))?;
+    // Clone sender to avoid holding RwLock read guard across send().await
+    let tx = {
+        let tx_guard = state.msg_tx.read().await;
+        tx_guard
+            .as_ref()
+            .ok_or((
+                StatusCode::SERVICE_UNAVAILABLE,
+                "Channel not started".to_string(),
+            ))?
+            .clone()
+    };
 
     tracing::debug!("[chat_send_handler] Sending message through channel");
     tx.send(msg).await.map_err(|_| {
@@ -1043,11 +1049,17 @@ async fn chat_approval_handler(
 
     let msg_id = msg.id;
 
-    let tx_guard = state.msg_tx.read().await;
-    let tx = tx_guard.as_ref().ok_or((
-        StatusCode::SERVICE_UNAVAILABLE,
-        "Channel not started".to_string(),
-    ))?;
+    // Clone sender to avoid holding RwLock read guard across send().await
+    let tx = {
+        let tx_guard = state.msg_tx.read().await;
+        tx_guard
+            .as_ref()
+            .ok_or((
+                StatusCode::SERVICE_UNAVAILABLE,
+                "Channel not started".to_string(),
+            ))?
+            .clone()
+    };
 
     tx.send(msg).await.map_err(|_| {
         (

src/channels/web/ws.rs

@@ -176,8 +176,12 @@ async fn handle_client_message(
                 incoming = incoming.with_attachments(attachments);
             }
 
-            let tx_guard = state.msg_tx.read().await;
-            if let Some(ref tx) = *tx_guard {
+            // Clone sender to avoid holding RwLock read guard across send().await
+            let tx = {
+                let tx_guard = state.msg_tx.read().await;
+                tx_guard.as_ref().cloned()
+            };
+            if let Some(tx) = tx {
                 if tx.send(incoming).await.is_err() {
                     let _ = direct_tx
                         .send(WsServerMessage::Error {
@@ -245,8 +249,12 @@ async fn handle_client_message(
             if let Some(ref tid) = thread_id {
                 msg = msg.with_thread(tid);
             }
-            let tx_guard = state.msg_tx.read().await;
-            if let Some(ref tx) = *tx_guard {
+            // Clone sender to avoid holding RwLock read guard across send().await
+            let tx = {
+                let tx_guard = state.msg_tx.read().await;
+                tx_guard.as_ref().cloned()
+            };
+            if let Some(tx) = tx {
                 let _ = tx.send(msg).await;
             }
         }