From 1a492fc4c8343e7966841030e18b7d24e468447b Mon Sep 17 00:00:00 2001 From: NimBold Date: Sat, 1 Aug 2026 19:12:15 +0330 Subject: [PATCH] fix(torrents): harden RPC probe shutdown and retry classification --- src-tauri/src/retry.rs | 21 +++++++++--- src-tauri/src/torrent_probe.rs | 59 ++++++++++++++++++++++++++-------- 2 files changed, 62 insertions(+), 18 deletions(-) diff --git a/src-tauri/src/retry.rs b/src-tauri/src/retry.rs index f7425b4..cc4ed8d 100644 --- a/src-tauri/src/retry.rs +++ b/src-tauri/src/retry.rs @@ -64,7 +64,7 @@ pub fn backoff_for(strike: usize) -> Duration { /// Classify an error string as a transient network condition worth retrying. /// /// Returns `true` for socket drops, connect/read timeouts, connection resets, -/// and HTTP 408 / request-timeout conditions across both download paths: +/// and transient HTTP status conditions across both download paths: /// /// - **yt-dlp**: stderr lines like `ERROR: unable to ... Connection timed out`, /// `HTTP Error 408`. @@ -164,9 +164,12 @@ pub fn is_transient_network_error(message: &str) -> bool { "timeout.", "invalid range header", ]; - contains_http_status(&m, "408") - || contains_http_status(&m, "429") - || contains_http_status(&m, "503") + const TRANSIENT_HTTP_STATUS: [&str; 11] = [ + "408", "429", "500", "502", "503", "504", "520", "521", "522", "523", "524", + ]; + TRANSIENT_HTTP_STATUS + .iter() + .any(|status| contains_http_status(&m, status)) || TRANSIENT.iter().any(|t| m.contains(t)) } @@ -292,6 +295,16 @@ mod tests { assert!(is_transient_network_error("The response status is not successful. status=429")); } + #[test] + fn classifies_rpc_http_gateway_errors_as_transient() { + for status in [500, 502, 503, 504, 520, 521, 522, 523, 524] { + assert!( + is_transient_network_error(&format!("HTTP {status} gateway failure")), + "HTTP {status} should be retryable" + ); + } + } + #[test] fn classifies_ytdlp_and_aria2_phrasing_as_transient() { assert!(is_transient_network_error( diff --git a/src-tauri/src/torrent_probe.rs b/src-tauri/src/torrent_probe.rs index 85fee99..64048a8 100644 --- a/src-tauri/src/torrent_probe.rs +++ b/src-tauri/src/torrent_probe.rs @@ -311,7 +311,7 @@ mod tests { use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Mutex; - use tokio::sync::{oneshot, Notify}; + use tokio::sync::{oneshot, watch, Notify}; enum ScriptedReply { Result(Value), @@ -332,6 +332,7 @@ mod tests { scripts: Mutex>>, calls: Mutex>, call_notification: Notify, + termination: watch::Sender, } struct ScriptedRpcServer { @@ -344,6 +345,7 @@ mod tests { impl ScriptedRpcServer { async fn start(scripts: impl IntoIterator)>) -> Self { let secret = "torrent-probe-test-secret".to_string(); + let (termination, _) = watch::channel(false); let state = Arc::new(ScriptedRpcState { secret: secret.clone(), scripts: Mutex::new( @@ -354,6 +356,7 @@ mod tests { ), calls: Mutex::new(Vec::new()), call_notification: Notify::new(), + termination, }); let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) .await @@ -415,6 +418,7 @@ mod tests { } async fn shutdown(mut self) { + self.state.termination.send_replace(true); if let Some(shutdown) = self.shutdown.take() { let _ = shutdown.send(()); } @@ -424,16 +428,27 @@ mod tests { } async fn terminate(mut self) { - if let Some(task) = self.task.take() { - task.abort(); - let _ = task.await; + self.state.termination.send_replace(true); + if let Some(shutdown) = self.shutdown.take() { + let _ = shutdown.send(()); + } + if let Some(mut task) = self.task.take() { + match tokio::time::timeout(Duration::from_secs(1), &mut task).await { + Ok(result) => { + let _ = result; + } + Err(_) => { + task.abort(); + let _ = task.await; + } + } } - self.shutdown.take(); } } impl Drop for ScriptedRpcServer { fn drop(&mut self) { + self.state.termination.send_replace(true); if let Some(task) = self.task.take() { task.abort(); } @@ -483,17 +498,29 @@ mod tests { .unwrap_or_else(|| { ScriptedReply::RpcError(format!("unexpected scripted RPC method {method}")) }); - scripted_reply_response(id, reply).await + scripted_reply_response(id, reply, state.termination.subscribe()).await } - async fn scripted_reply_response(id: Value, mut reply: ScriptedReply) -> Response { + async fn scripted_reply_response( + id: Value, + mut reply: ScriptedReply, + mut termination: watch::Receiver, + ) -> Response { loop { + if *termination.borrow() { + return terminated_response(); + } match reply { ScriptedReply::Delay(delay, next) => { - tokio::time::sleep(delay).await; - reply = *next; + tokio::select! { + _ = tokio::time::sleep(delay) => reply = *next, + _ = termination.changed() => return terminated_response(), + } + } + ScriptedReply::Hang => { + let _ = termination.changed().await; + return terminated_response(); } - ScriptedReply::Hang => return std::future::pending::().await, ScriptedReply::Result(result) => { return Json(json!({ "jsonrpc": "2.0", @@ -515,6 +542,10 @@ mod tests { } } + fn terminated_response() -> Response { + (StatusCode::OK, Body::empty()).into_response() + } + fn rpc_error_response(id: Value, status: StatusCode, message: String) -> Response { ( status, @@ -1226,8 +1257,8 @@ mod tests { "aria2.forceRemove", vec![ ScriptedReply::HttpError( - StatusCode::SERVICE_UNAVAILABLE, - "temporary cleanup outage".to_string(), + StatusCode::BAD_GATEWAY, + "temporary gateway outage".to_string(), ), ScriptedReply::RpcError("connection reset by peer".to_string()), ScriptedReply::Result(json!("gid-1")), @@ -1345,7 +1376,7 @@ mod tests { }); server.wait_for_method("aria2.tellStatus", 1).await; server.terminate().await; - let result = tokio::time::timeout(Duration::from_secs(5), task) + let result = tokio::time::timeout(Duration::from_secs(2), task) .await .expect("probe should finish after daemon shutdown") .expect("probe task should not panic") @@ -1381,7 +1412,7 @@ mod tests { }); server.wait_for_method("aria2.forceRemove", 1).await; server.terminate().await; - let result = tokio::time::timeout(Duration::from_secs(5), task) + let result = tokio::time::timeout(Duration::from_secs(2), task) .await .expect("cleanup should finish after daemon shutdown") .expect("cleanup task should not panic")