From 706826f9f7031fe5ffd56ed0f87d3ef25eec2f87 Mon Sep 17 00:00:00 2001 From: TeacherLi <110473389+TeacherLi07@users.noreply.github.com> Date: Tue, 13 Oct 2026 16:16:01 +0000 Subject: [PATCH 1/3] fix: preserve sessions after client write resets --- libshpool/src/daemon/shell.rs | 36 ++++++++++++-------- libshpool/src/test_hooks.rs | 23 +++++++++++++ shpool/tests/regression.rs | 62 +++++++++++++++++++++++++++++++++++ 3 files changed, 108 insertions(+), 13 deletions(-) diff --git a/libshpool/src/daemon/shell.rs b/libshpool/src/daemon/shell.rs index 2baac978..d83b702b 100644 --- a/libshpool/src/daemon/shell.rs +++ b/libshpool/src/daemon/shell.rs @@ -476,24 +476,32 @@ impl SessionInner { recv(args.heartbeat) -> request_id => { let client_present = if let ClientConnectionMsg::New(conn) = &mut client_conn { let chunk = Chunk { kind: ChunkKind::Heartbeat, buf: &[] }; - match chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()) { + let write_result = test_hooks::client_write_error("heartbeat") + .map_or_else(|| chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()), Err); + match write_result { Ok(_) => { trace!("wrote heartbeat"); true } - Err(e) if e.kind() == io::ErrorKind::BrokenPipe => { - trace!("client hangup: {:?}", e); - false - } Err(e) => { - error!("unexpected IO error while writing heartbeat: {}", e); - return Err(e).context("writing heartbeat")?; + info!("heartbeat write err, assuming hangup: {:?}", e); + // A failed client write must not terminate the + // session's long-lived output worker. Shut down + // every clone so client->shell also wakes up. + if let Err(err) = shutdown_socket(&conn.stream, net::Shutdown::Both) { + warn!("shutting down client after heartbeat error: {:?}", err); + } + false } } } else { false }; + if !client_present { + client_conn = ClientConnectionMsg::Disconnect; + } + test_hooks::emit("daemon-wrote-heartbeat"); args.heartbeat_ack.send((request_id.unwrap_or(0), client_present)) @@ -522,16 +530,18 @@ impl SessionInner { } let chunk = Chunk { kind: ChunkKind::MaybeSwitch, buf: &encoded[..] }; - match chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()) { + let write_result = test_hooks::client_write_error("maybe-switch") + .map_or_else(|| chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()), Err); + match write_result { Ok(_) => { trace!("wrote MaybeSwitch"); } - Err(e) if e.kind() == io::ErrorKind::BrokenPipe => { - trace!("writing MaybeSwitch: client hangup: {:?}", e); - } Err(e) => { - error!("unexpected IO error while writing heartbeat: {}", e); - return Err(e).context("writing MaybeSwitch")?; + info!("MaybeSwitch write err, assuming hangup: {:?}", e); + if let Err(err) = shutdown_socket(&conn.stream, net::Shutdown::Both) { + warn!("shutting down client after MaybeSwitch error: {:?}", err); + } + client_conn = ClientConnectionMsg::Disconnect; } } } diff --git a/libshpool/src/test_hooks.rs b/libshpool/src/test_hooks.rs index 5fa4b2ab..e41ef72e 100644 --- a/libshpool/src/test_hooks.rs +++ b/libshpool/src/test_hooks.rs @@ -45,6 +45,29 @@ pub fn emit(_event: &str) { // a no-op normally } +/// Inject one client write reset per daemon for the regression tests. This +/// leaves the peer open so the test also checks that shutdown wakes the input +/// thread, rather than relying on the client to close its own socket. +#[cfg(feature = "test_hooks")] +pub fn client_write_error(kind: &str) -> Option { + use std::sync::atomic::{AtomicBool, Ordering}; + + static INJECTED: AtomicBool = AtomicBool::new(false); + if std::env::var("SHPOOL_TEST__CLIENT_WRITE_RESET").as_deref() == Ok(kind) + && !INJECTED.swap(true, Ordering::Relaxed) + { + emit("daemon-injected-client-write-error"); + Some(std::io::Error::from_raw_os_error(libc::ECONNRESET)) + } else { + None + } +} + +#[cfg(not(feature = "test_hooks"))] +pub fn client_write_error(_kind: &str) -> Option { + None +} + #[cfg(feature = "test_hooks")] pub fn scoped(event: &str) -> ScopedEvent { ScopedEvent::new(event) diff --git a/shpool/tests/regression.rs b/shpool/tests/regression.rs index db18dd91..abf16583 100644 --- a/shpool/tests/regression.rs +++ b/shpool/tests/regression.rs @@ -11,6 +11,68 @@ mod support; use crate::support::{daemon::DaemonArgs, tmpdir}; +fn client_write_reset_preserves_session(kind: &str) -> anyhow::Result<()> { + let mut daemon_proc = support::daemon::Proc::new( + "norc.toml", + DaemonArgs { + extra_env: vec![(String::from("SHPOOL_TEST__CLIENT_WRITE_RESET"), kind.to_owned())], + ..DaemonArgs::default() + }, + )?; + let mut original_client = daemon_proc.attach("main", Default::default())?; + daemon_proc.await_event("daemon-bidi-stream-enter")?; + + let sessions: serde_json::Value = serde_json::from_slice(&daemon_proc.list_json()?.stdout)?; + let shell_pid = sessions["sessions"][0]["shell_pid"].as_i64().unwrap(); + + if kind == "maybe-switch" { + assert!(daemon_proc.var_set("repro", "changed")?.status.success()); + } + daemon_proc.await_event("daemon-injected-client-write-error")?; + daemon_proc.wait_until_list_matches(|out| out.contains("disconnected"))?; + // The peer did not close voluntarily. It must be released by the daemon. + original_client.proc.wait()?; + signal::kill(Pid::from_raw(shell_pid as i32), None)?; + assert!( + !fs::read_to_string(&daemon_proc.log_file)?.contains("error in shell->client:"), + "a client write reset must not terminate the session's output worker" + ); + + for _ in 0..3 { + let mut client = daemon_proc.attach("main", Default::default())?; + let mut output = client.line_matcher()?; + client.run_cmd("printf 'reset-%s\\n' recovered")?; + output.scan_until_re("reset-recovered$")?; + + let sessions: serde_json::Value = serde_json::from_slice(&daemon_proc.list_json()?.stdout)?; + assert_eq!(sessions["sessions"][0]["shell_pid"].as_i64(), Some(shell_pid)); + assert_eq!(sessions["sessions"][0]["status"], "Attached"); + + assert!(daemon_proc.detach(vec![String::from("main")])?.status.success()); + client.proc.wait()?; + daemon_proc.wait_until_list_matches(|out| out.contains("disconnected"))?; + } + let sessions: serde_json::Value = serde_json::from_slice(&daemon_proc.list_json()?.stdout)?; + assert_eq!(sessions["sessions"][0]["attachments"], serde_json::json!([])); + assert!(daemon_proc.kill(vec![String::from("main")])?.status.success()); + Ok(()) +} + +/// ECONNRESET on a heartbeat used to terminate the output worker while the +/// shell stayed alive, permanently preventing subsequent attaches. +#[test] +#[timeout(15000)] +fn heartbeat_write_reset_preserves_session() -> anyhow::Result<()> { + client_write_reset_preserves_session("heartbeat") +} + +/// MaybeSwitch writes share the heartbeat's client-disconnect failure mode. +#[test] +#[timeout(15000)] +fn maybe_switch_write_reset_preserves_session() -> anyhow::Result<()> { + client_write_reset_preserves_session("maybe-switch") +} + /// Regression test for a deadlock where the global `shells` mutex is held /// during `spawn_subshell` -> `wait_for_startup`. If `wait_for_startup` /// blocks (e.g. the shell never produces the startup sentinel), the mutex From d143180f499208a1eaf1008e8f3ba51b163df602 Mon Sep 17 00:00:00 2001 From: TeacherLi <110473389+TeacherLi07@users.noreply.github.com> Date: Sat, 7 Nov 2026 07:20:37 +0000 Subject: [PATCH 2/3] test: generalize failure injection for client write resets Support comma-separated failure points with thread-safe, one-time consumption per process. Keep error construction at the call sites and disable injection without test_hooks. Consolidate the reset regression into one MaybeSwitch test with an unconditional variable update, as requested in review. Add coverage for multiple failure points and concurrent consumption with and without test_hooks. Validation: regression 11/11, var 7/7, template-switch 2/2, production cargo check, nightly rustfmt, and git diff --check passed. Clippy passed with the existing unwrap_or_default allowance. The merged regression rejects the original error handling. --- libshpool/src/daemon/shell.rs | 14 +++++-- libshpool/src/test_hooks.rs | 71 +++++++++++++++++++++++++++-------- shpool/tests/regression.rs | 32 ++++++---------- 3 files changed, 76 insertions(+), 41 deletions(-) diff --git a/libshpool/src/daemon/shell.rs b/libshpool/src/daemon/shell.rs index d83b702b..342c1be9 100644 --- a/libshpool/src/daemon/shell.rs +++ b/libshpool/src/daemon/shell.rs @@ -476,8 +476,11 @@ impl SessionInner { recv(args.heartbeat) -> request_id => { let client_present = if let ClientConnectionMsg::New(conn) = &mut client_conn { let chunk = Chunk { kind: ChunkKind::Heartbeat, buf: &[] }; - let write_result = test_hooks::client_write_error("heartbeat") - .map_or_else(|| chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()), Err); + let write_result = if test_hooks::fail_once("heartbeat-write") { + Err(io::Error::from_raw_os_error(libc::ECONNRESET)) + } else { + chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()) + }; match write_result { Ok(_) => { trace!("wrote heartbeat"); @@ -530,8 +533,11 @@ impl SessionInner { } let chunk = Chunk { kind: ChunkKind::MaybeSwitch, buf: &encoded[..] }; - let write_result = test_hooks::client_write_error("maybe-switch") - .map_or_else(|| chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()), Err); + let write_result = if test_hooks::fail_once("maybe-switch-write") { + Err(io::Error::from_raw_os_error(libc::ECONNRESET)) + } else { + chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()) + }; match write_result { Ok(_) => { trace!("wrote MaybeSwitch"); diff --git a/libshpool/src/test_hooks.rs b/libshpool/src/test_hooks.rs index e41ef72e..07ab23cd 100644 --- a/libshpool/src/test_hooks.rs +++ b/libshpool/src/test_hooks.rs @@ -45,27 +45,34 @@ pub fn emit(_event: &str) { // a no-op normally } -/// Inject one client write reset per daemon for the regression tests. This -/// leaves the peer open so the test also checks that shutdown wakes the input -/// thread, rather than relying on the client to close its own socket. +/// Consume a named failure point at most once per process. Configure points +/// with a comma-separated SHPOOL_TEST__FAILURE_POINTS environment variable. #[cfg(feature = "test_hooks")] -pub fn client_write_error(kind: &str) -> Option { - use std::sync::atomic::{AtomicBool, Ordering}; - - static INJECTED: AtomicBool = AtomicBool::new(false); - if std::env::var("SHPOOL_TEST__CLIENT_WRITE_RESET").as_deref() == Ok(kind) - && !INJECTED.swap(true, Ordering::Relaxed) - { - emit("daemon-injected-client-write-error"); - Some(std::io::Error::from_raw_os_error(libc::ECONNRESET)) - } else { - None +pub fn fail_once(point: &str) -> bool { + use std::sync::OnceLock; + + static FAILURE_POINTS: OnceLock>> = OnceLock::new(); + let points = FAILURE_POINTS.get_or_init(|| { + Mutex::new( + std::env::var("SHPOOL_TEST__FAILURE_POINTS") + .unwrap_or_default() + .split(',') + .map(str::trim) + .filter(|point| !point.is_empty()) + .map(String::from) + .collect(), + ) + }); + let injected = points.lock().remove(point); + if injected { + emit(&format!("injected-failure-{point}")); } + injected } #[cfg(not(feature = "test_hooks"))] -pub fn client_write_error(_kind: &str) -> Option { - None +pub fn fail_once(_point: &str) -> bool { + false } #[cfg(feature = "test_hooks")] @@ -246,3 +253,35 @@ impl TestHookServer { } } } + +#[cfg(test)] +mod tests { + #[test] + fn failure_points_from_environment() { + // Use a subprocess to isolate the environment and process-wide state + // from other tests, without changing the environment of running + // threads. + if std::env::var_os("SHPOOL_TEST__FAILURE_POINTS_CHILD").is_none() { + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "test_hooks::tests::failure_points_from_environment"]) + .env("SHPOOL_TEST__FAILURE_POINTS_CHILD", "1") + .env("SHPOOL_TEST__FAILURE_POINTS", "first, second,,first") + .status() + .unwrap(); + assert!(status.success()); + return; + } + + assert!(!super::fail_once("unknown")); + assert!(!super::fail_once("")); + let workers: Vec<_> = (0..8) + .map(|_| std::thread::spawn(|| (super::fail_once("first"), super::fail_once("second")))) + .collect(); + let results: Vec<_> = workers.into_iter().map(|worker| worker.join().unwrap()).collect(); + let expected = usize::from(cfg!(feature = "test_hooks")); + assert_eq!(results.iter().filter(|(first, _)| *first).count(), expected); + assert_eq!(results.iter().filter(|(_, second)| *second).count(), expected); + assert!(!super::fail_once("first")); + assert!(!super::fail_once("second")); + } +} diff --git a/shpool/tests/regression.rs b/shpool/tests/regression.rs index abf16583..a92de5b9 100644 --- a/shpool/tests/regression.rs +++ b/shpool/tests/regression.rs @@ -11,11 +11,18 @@ mod support; use crate::support::{daemon::DaemonArgs, tmpdir}; -fn client_write_reset_preserves_session(kind: &str) -> anyhow::Result<()> { +/// A MaybeSwitch write reset must disconnect the client while preserving the +/// output worker and original shell for subsequent attaches. +#[test] +#[timeout(15000)] +fn client_write_reset_preserves_session() -> anyhow::Result<()> { let mut daemon_proc = support::daemon::Proc::new( "norc.toml", DaemonArgs { - extra_env: vec![(String::from("SHPOOL_TEST__CLIENT_WRITE_RESET"), kind.to_owned())], + extra_env: vec![( + String::from("SHPOOL_TEST__FAILURE_POINTS"), + String::from("maybe-switch-write"), + )], ..DaemonArgs::default() }, )?; @@ -25,10 +32,8 @@ fn client_write_reset_preserves_session(kind: &str) -> anyhow::Result<()> { let sessions: serde_json::Value = serde_json::from_slice(&daemon_proc.list_json()?.stdout)?; let shell_pid = sessions["sessions"][0]["shell_pid"].as_i64().unwrap(); - if kind == "maybe-switch" { - assert!(daemon_proc.var_set("repro", "changed")?.status.success()); - } - daemon_proc.await_event("daemon-injected-client-write-error")?; + assert!(daemon_proc.var_set("repro", "changed")?.status.success()); + daemon_proc.await_event("injected-failure-maybe-switch-write")?; daemon_proc.wait_until_list_matches(|out| out.contains("disconnected"))?; // The peer did not close voluntarily. It must be released by the daemon. original_client.proc.wait()?; @@ -58,21 +63,6 @@ fn client_write_reset_preserves_session(kind: &str) -> anyhow::Result<()> { Ok(()) } -/// ECONNRESET on a heartbeat used to terminate the output worker while the -/// shell stayed alive, permanently preventing subsequent attaches. -#[test] -#[timeout(15000)] -fn heartbeat_write_reset_preserves_session() -> anyhow::Result<()> { - client_write_reset_preserves_session("heartbeat") -} - -/// MaybeSwitch writes share the heartbeat's client-disconnect failure mode. -#[test] -#[timeout(15000)] -fn maybe_switch_write_reset_preserves_session() -> anyhow::Result<()> { - client_write_reset_preserves_session("maybe-switch") -} - /// Regression test for a deadlock where the global `shells` mutex is held /// during `spawn_subshell` -> `wait_for_startup`. If `wait_for_startup` /// blocks (e.g. the shell never produces the startup sentinel), the mutex From 353364a02361f243f2795652bea422089b66bc8b Mon Sep 17 00:00:00 2001 From: TeacherLi <110473389+TeacherLi07@users.noreply.github.com> Date: Tue, 10 Nov 2026 14:46:09 +0000 Subject: [PATCH 3/3] test: inject failures on selected call counts --- libshpool/src/daemon/shell.rs | 4 +- libshpool/src/test_hooks.rs | 112 ++++++++++++++++++++++++++-------- shpool/tests/regression.rs | 2 +- 3 files changed, 90 insertions(+), 28 deletions(-) diff --git a/libshpool/src/daemon/shell.rs b/libshpool/src/daemon/shell.rs index 342c1be9..70a1be5f 100644 --- a/libshpool/src/daemon/shell.rs +++ b/libshpool/src/daemon/shell.rs @@ -476,7 +476,7 @@ impl SessionInner { recv(args.heartbeat) -> request_id => { let client_present = if let ClientConnectionMsg::New(conn) = &mut client_conn { let chunk = Chunk { kind: ChunkKind::Heartbeat, buf: &[] }; - let write_result = if test_hooks::fail_once("heartbeat-write") { + let write_result = if test_hooks::maybe_fail("heartbeat-write") { Err(io::Error::from_raw_os_error(libc::ECONNRESET)) } else { chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()) @@ -533,7 +533,7 @@ impl SessionInner { } let chunk = Chunk { kind: ChunkKind::MaybeSwitch, buf: &encoded[..] }; - let write_result = if test_hooks::fail_once("maybe-switch-write") { + let write_result = if test_hooks::maybe_fail("maybe-switch-write") { Err(io::Error::from_raw_os_error(libc::ECONNRESET)) } else { chunk.write_to(&mut conn.sink).and_then(|_| conn.sink.flush()) diff --git a/libshpool/src/test_hooks.rs b/libshpool/src/test_hooks.rs index 07ab23cd..df22d7c8 100644 --- a/libshpool/src/test_hooks.rs +++ b/libshpool/src/test_hooks.rs @@ -45,25 +45,54 @@ pub fn emit(_event: &str) { // a no-op normally } -/// Consume a named failure point at most once per process. Configure points -/// with a comma-separated SHPOOL_TEST__FAILURE_POINTS environment variable. +/// Inject failures on selected calls to a named failure point. Configure +/// SHPOOL_TEST__FAILURE_POINTS with comma-separated `name:n_1:n_2:...` entries. +/// Calls are counted from 1, independently for each point, across the process. +/// For example, `heartbeat-write:2:4` fails on the second and fourth calls. #[cfg(feature = "test_hooks")] -pub fn fail_once(point: &str) -> bool { - use std::sync::OnceLock; +pub fn maybe_fail(point: &str) -> bool { + use std::{collections::HashMap, sync::OnceLock}; - static FAILURE_POINTS: OnceLock>> = OnceLock::new(); + #[derive(Default)] + struct FailurePoint { + calls: usize, + fail_on: HashSet, + } + + static FAILURE_POINTS: OnceLock>> = OnceLock::new(); let points = FAILURE_POINTS.get_or_init(|| { - Mutex::new( - std::env::var("SHPOOL_TEST__FAILURE_POINTS") - .unwrap_or_default() - .split(',') - .map(str::trim) - .filter(|point| !point.is_empty()) - .map(String::from) - .collect(), - ) + let mut points = HashMap::new(); + for spec in std::env::var("SHPOOL_TEST__FAILURE_POINTS") + .unwrap_or_default() + .split(',') + .map(str::trim) + .filter(|spec| !spec.is_empty()) + { + let (name, calls) = + spec.split_once(':').expect("failure points must use name:call[:call...]"); + let name = name.trim(); + assert!(!name.is_empty(), "failure point names must not be empty"); + let failure = points.entry(name.to_owned()).or_insert_with(FailurePoint::default); + for call in calls.split(':') { + let call = call + .trim() + .parse::() + .expect("failure point calls must be positive integers"); + assert!(call > 0, "failure point calls are counted from 1"); + failure.fail_on.insert(call); + } + } + Mutex::new(points) }); - let injected = points.lock().remove(point); + let injected = { + let mut points = points.lock(); + if let Some(failure) = points.get_mut(point) { + failure.calls += 1; + failure.fail_on.remove(&failure.calls) + } else { + false + } + }; if injected { emit(&format!("injected-failure-{point}")); } @@ -71,7 +100,7 @@ pub fn fail_once(point: &str) -> bool { } #[cfg(not(feature = "test_hooks"))] -pub fn fail_once(_point: &str) -> bool { +pub fn maybe_fail(_point: &str) -> bool { false } @@ -265,23 +294,56 @@ mod tests { let status = std::process::Command::new(std::env::current_exe().unwrap()) .args(["--exact", "test_hooks::tests::failure_points_from_environment"]) .env("SHPOOL_TEST__FAILURE_POINTS_CHILD", "1") - .env("SHPOOL_TEST__FAILURE_POINTS", "first, second,,first") + .env( + "SHPOOL_TEST__FAILURE_POINTS", + "first:4:2:4, second:1:3,,concurrent:2:4:7,other-concurrent:1:8", + ) .status() .unwrap(); assert!(status.success()); return; } - assert!(!super::fail_once("unknown")); - assert!(!super::fail_once("")); + assert!(!super::maybe_fail("unknown")); + assert!(!super::maybe_fail("")); + let enabled = cfg!(feature = "test_hooks"); + for call in 1..=6 { + assert_eq!(super::maybe_fail("first"), enabled && [2, 4].contains(&call)); + assert_eq!(super::maybe_fail("second"), enabled && [1, 3].contains(&call)); + } + let workers: Vec<_> = (0..8) - .map(|_| std::thread::spawn(|| (super::fail_once("first"), super::fail_once("second")))) + .map(|_| { + std::thread::spawn(|| { + (super::maybe_fail("concurrent"), super::maybe_fail("other-concurrent")) + }) + }) .collect(); let results: Vec<_> = workers.into_iter().map(|worker| worker.join().unwrap()).collect(); - let expected = usize::from(cfg!(feature = "test_hooks")); - assert_eq!(results.iter().filter(|(first, _)| *first).count(), expected); - assert_eq!(results.iter().filter(|(_, second)| *second).count(), expected); - assert!(!super::fail_once("first")); - assert!(!super::fail_once("second")); + assert_eq!(results.iter().filter(|(first, _)| *first).count(), if enabled { 3 } else { 0 }); + assert_eq!( + results.iter().filter(|(_, second)| *second).count(), + if enabled { 2 } else { 0 } + ); + assert!(!super::maybe_fail("concurrent")); + assert!(!super::maybe_fail("other-concurrent")); + } + + #[test] + fn failure_points_disabled_by_default() { + if std::env::var_os("SHPOOL_TEST__FAILURE_POINTS_CHILD").is_none() { + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "test_hooks::tests::failure_points_disabled_by_default"]) + .env("SHPOOL_TEST__FAILURE_POINTS_CHILD", "1") + .env_remove("SHPOOL_TEST__FAILURE_POINTS") + .status() + .unwrap(); + assert!(status.success()); + return; + } + + for _ in 0..3 { + assert!(!super::maybe_fail("heartbeat-write")); + } } } diff --git a/shpool/tests/regression.rs b/shpool/tests/regression.rs index a92de5b9..b0b92bcf 100644 --- a/shpool/tests/regression.rs +++ b/shpool/tests/regression.rs @@ -21,7 +21,7 @@ fn client_write_reset_preserves_session() -> anyhow::Result<()> { DaemonArgs { extra_env: vec![( String::from("SHPOOL_TEST__FAILURE_POINTS"), - String::from("maybe-switch-write"), + String::from("maybe-switch-write:1"), )], ..DaemonArgs::default() },