From ef34943c14993d4db155d7f6ea07650266732e05 Mon Sep 17 00:00:00 2001 From: Zaki Manian Date: Thu, 12 Mar 2026 11:10:04 -0700 Subject: [PATCH] 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 (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 (cherry picked from commit 1d5a7bdc8ec071cdddf0e69d6053d49ca20a2b18) * chore(ci): rerun regression gate [skip-regression-check] (cherry picked from commit 784d444701471a1311b1f985e2f07be6f0527abf) --------- Co-authored-by: Umesh Kumar Singh Co-authored-by: Claude Opus 4.6 --- .github/workflows/regression-test-check.yml | 17 ++-- src/channels/http.rs | 61 ++++++++++++++ src/channels/wasm/wrapper.rs | 90 ++++++++++++--------- src/channels/web/handlers/chat.rs | 32 +++++--- src/channels/web/server.rs | 32 +++++--- src/channels/web/ws.rs | 16 +++- 6 files changed, 179 insertions(+), 69 deletions(-) diff --git a/.github/workflows/regression-test-check.yml b/.github/workflows/regression-test-check.yml index 18b8c76f..6d97c4ce 100644 --- a/.github/workflows/regression-test-check.yml +++ b/.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 } diff --git a/src/channels/http.rs b/src/channels/http.rs index cf2a9945..42fc54f8 100644 --- a/src/channels/http.rs +++ b/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")); diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index a9fa4dbf..914ffbf0 100644 --- a/src/channels/wasm/wrapper.rs +++ b/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, diff --git a/src/channels/web/handlers/chat.rs b/src/channels/web/handlers/chat.rs index 91c4533b..909a252c 100644 --- a/src/channels/web/handlers/chat.rs +++ b/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(|_| { ( diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index 4dc58390..f08c95c2 100644 --- a/src/channels/web/server.rs +++ b/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(|_| { ( diff --git a/src/channels/web/ws.rs b/src/channels/web/ws.rs index 343112a1..7287902e 100644 --- a/src/channels/web/ws.rs +++ b/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; } }