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

.github/workflows/regression-test-check.yml

@@ -13,6 +13,11 @@ jobs:
         with:
           fetch-depth: 0
 
+      - name: Fetch PR head and base
+        run: |
+          git fetch origin ${{ github.event.pull_request.base.ref }}
+          git fetch origin pull/${{ github.event.pull_request.number }}/head:pr-head
+
       - name: Check for regression tests
         env:
           PR_TITLE: ${{ github.event.pull_request.title }}
@@ -21,6 +26,8 @@ jobs:
           set -euo pipefail
 
           BASE_REF="origin/${{ github.event.pull_request.base.ref }}"
+          # Use the actual PR head, not the merge commit that actions/checkout checks out
+          HEAD_REF="pr-head"
 
           # --- 1. Is this a fix PR? Check title first, then commit messages ---
           IS_FIX=false
@@ -30,7 +37,7 @@ jobs:
           fi
 
           if [ "$IS_FIX" = false ]; then
-            COMMITS=$(git log --format='%s' "${BASE_REF}..HEAD")
+            COMMITS=$(git log --format='%s' "${BASE_REF}..${HEAD_REF}")
             if grep -qiE '^(fix(\(.*\))?|hotfix|bugfix):' <<< "$COMMITS"; then
               IS_FIX=true
             fi
@@ -49,14 +56,14 @@ jobs:
             exit 0
           fi
 
-          COMMIT_BODIES=$(git log --format='%B' "${BASE_REF}..HEAD")
+          COMMIT_BODIES=$(git log --format='%B' "${BASE_REF}..${HEAD_REF}")
           if grep -qF '[skip-regression-check]' <<< "$COMMIT_BODIES"; then
             echo "[skip-regression-check] found in commit message — skipping."
             exit 0
           fi
 
           # --- 3. Exempt static-only / docs-only changes ---
-          CHANGED_FILES=$(git diff --name-only "${BASE_REF}...HEAD")
+          CHANGED_FILES=$(git diff --name-only "${BASE_REF}...${HEAD_REF}")
 
           if [ -z "$CHANGED_FILES" ]; then
             echo "No changed files — skipping."
@@ -80,13 +87,13 @@ jobs:
           # --- 4. Look for test changes ---
 
           # Fast path: new test attributes or test modules in added lines.
-          if git diff "${BASE_REF}...HEAD" -U0 -- '*.rs' | grep -qE '^\+.*(#\[test\]|#\[tokio::test\]|#\[cfg\(test\)\]|mod tests)'; then
+          if git diff "${BASE_REF}...${HEAD_REF}" -U0 -- '*.rs' | grep -qE '^\+.*(#\[test\]|#\[tokio::test\]|#\[cfg\(test\)\]|mod tests)'; then
             echo "Test changes found in .rs files."
             exit 0
           fi
 
           # Whole-function context: detect edits inside existing test functions.
-          if git diff "${BASE_REF}...HEAD" -W -- '*.rs' | awk '
+          if git diff "${BASE_REF}...${HEAD_REF}" -W -- '*.rs' | awk '
             /^@@/           { if (has_test && has_add) { found=1; exit } has_test=0; has_add=0 }
             /^ .*#\[test\]/ || /^ .*#\[tokio::test\]/ || /^ .*#\[cfg\(test\)\]/ || /^ .*mod tests/ { has_test=1 }
             /^\+.*#\[test\]/ || /^\+.*#\[tokio::test\]/ || /^\+.*#\[cfg\(test\)\]/ || /^\+.*mod tests/ { has_test=1 }

src/channels/http.rs

@@ -580,6 +580,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_secret_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

@@ -958,11 +958,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(|_| {
@@ -1028,11 +1034,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;
             }
         }