diff --git a/Cargo.lock b/Cargo.lock index d9cf7ea..b2a15a8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1028,6 +1028,7 @@ dependencies = [ "dbus", "futures", "getrandom 0.3.4", + "gio", "glass_pumpkin", "grammers-client", "grammers-crypto", diff --git a/crates/chatd/Cargo.toml b/crates/chatd/Cargo.toml index de65df4..c96daa9 100644 --- a/crates/chatd/Cargo.toml +++ b/crates/chatd/Cargo.toml @@ -79,5 +79,7 @@ windows-sys = { version = "0.59", features = ["Win32_Networking_WinSock"] } windows-sys = { version = "0.59", features = ["Win32_Foundation", "Win32_System_Diagnostics_ToolHelp", "Win32_System_ProcessStatus", "Win32_System_Threading"] } [target.'cfg(target_os = "linux")'.dependencies] +# OS connectivity hints, including the netlink backend when NetworkManager is absent. +gio = "0.18" # Already used by keyring. Inspect structured timeout errors without matching messages. dbus = { version = "0.9", features = ["vendored"] } diff --git a/crates/chatd/src/adapters/mod.rs b/crates/chatd/src/adapters/mod.rs index 77c84bd..f852895 100644 --- a/crates/chatd/src/adapters/mod.rs +++ b/crates/chatd/src/adapters/mod.rs @@ -16,6 +16,7 @@ //! message edited, deleted or reacted to only by the command the shell's request became. mod discord; +mod recovery; pub mod net; mod script; pub mod messages; @@ -38,7 +39,7 @@ use protocol::{ }; use std::collections::{HashMap, VecDeque}; use std::panic::{catch_unwind, AssertUnwindSafe}; -use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::sync::mpsc::{channel, Receiver, Sender}; use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; @@ -385,6 +386,7 @@ impl From for Failure { #[derive(Default)] struct Halt { stopped: AtomicBool, + recovery: AtomicU64, /// Panics caught on this adapter's threads. The first is recovered from; the second stops /// the adapter and the account says why. panics: AtomicU32, @@ -417,6 +419,16 @@ impl Context { self.stopped() } + fn reconnect_pause(&self, time: Duration, generation: u64) -> ReconnectWait { + let waiting = lock(&self.stop.waiting); + drop(self.stop.wake.wait_timeout_while(waiting, time, |_| { + !self.stopped() && self.stop.recovery.load(Ordering::Relaxed) == generation + })); + if self.stopped() { ReconnectWait::Stopped } + else if self.stop.recovery.load(Ordering::Relaxed) != generation { ReconnectWait::Recovered } + else { ReconnectWait::Elapsed } + } + pub fn key(&self, id: &str) -> Key { Key::new(&self.account, id) } @@ -982,10 +994,14 @@ impl Drops { /// A panic inside `connect` is caught here, at the thread's edge: the account goes offline /// with [`RECOVERING`] and connects again after [`PANIC_PAUSE`]; a second panic on this /// adapter stops it with [`BROKEN`]. -fn keep_connected(ctx: &Context, mut connect: impl FnMut() -> Failure) { - let mut pause = Duration::from_secs(1); +fn keep_connected(ctx: &Context, connect: impl FnMut() -> Failure) { + keep_connected_with_backoff(ctx, connect, Duration::from_secs(1)); +} + +fn keep_connected_with_backoff(ctx: &Context, mut connect: impl FnMut() -> Failure, mut pause: Duration) { let mut drops = Drops::default(); while !ctx.stopped() { + let generation = ctx.stop.recovery.load(Ordering::Relaxed); let began = Instant::now(); let failure = match catch_unwind(AssertUnwindSafe(&mut connect)) { Ok(failure) => failure, @@ -1022,12 +1038,27 @@ fn keep_connected(ctx: &Context, mut connect: impl FnMut() -> Failure) { ctx.hub.set_offline(&ctx.account, pause <= RECONNECTING); // Not on the second: many clients retrying in step is what a network watches for. let jitter = pause.mul_f64(0.25 * f64::from(now().unsigned_abs() as u32 % 1000) / 1000.); - if ctx.pause(pause + jitter) { - return; + match ctx.reconnect_pause(pause + jitter, generation) { + ReconnectWait::Stopped => return, + ReconnectWait::Elapsed => {}, + ReconnectWait::Recovered => { + // A changed route or resume is a reason to try, not proof that + // the remote service is reachable. Failures still back off. + drops = Drops::default(); + pause = Duration::from_secs(1); + ctx.hub.set_offline(&ctx.account, true); + eprintln!("chatd: {}: network changed or system resumed; retrying connection", ctx.account.network.label()); + // Consume bursts on the next iteration and never attempt more + // often than once per second, even if hints keep arriving. + if ctx.pause(Duration::from_secs(1).saturating_sub(began.elapsed())) { return; } + }, } } } +#[derive(Debug, PartialEq, Eq)] +enum ReconnectWait { Elapsed, Recovered, Stopped } + /// A retry this close is "reconnecting"; one further off is plain offline. const RECONNECTING: Duration = Duration::from_secs(30); @@ -1155,6 +1186,7 @@ pub fn without(network: Network, session: &str, account: &str) -> Option /// Starts the adapter for an account from its part of the session. pub fn start(account: AccountKey, session: &str, hub: Arc) -> Result { let stop = Arc::new(Halt::default()); + recovery::watch(&stop); let (commands, inbox): (Sender, Receiver) = channel(); let ctx = Context { account: account.clone(), @@ -1199,6 +1231,71 @@ mod tests { use super::*; use crate::store::Store; + #[test] + fn recovery_interrupts_a_fifteen_minute_reconnect_wait_without_rapid_attempts() { + let hub = hub(); + let account = AccountKey { network: Network::Signal, account: "recovery".into() }; + let (_adapter, ctx, _inbox) = local(account.clone(), hub.clone()); + ctx.link(Link::Connected); + let (attempted, attempts) = channel(); + let thread_ctx = ctx.clone(); + let thread = std::thread::spawn(move || { + let mut count = 0; + keep_connected_with_backoff(&thread_ctx, || { + count += 1; + attempted.send(Instant::now()).unwrap(); + if count == 1 { Failure::Dropped("offline".into()) } + else { Failure::SignedOut("test finished".into()) } + }, Duration::from_secs(900)); + }); + let first = attempts.recv_timeout(Duration::from_secs(2)).unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + while linked(&hub, &account).0 != Link::Offline { + assert!(Instant::now() < deadline); + std::thread::sleep(Duration::from_millis(1)); + } + for _ in 0..100 { recovery::hint(&ctx.stop); } + let next = attempts.recv_timeout(Duration::from_secs(2)).expect("recovery must not wait out the 900 s delay"); + assert!(next.duration_since(first) >= Duration::from_secs(1), "hints cannot create a rapid retry loop"); + thread.join().unwrap(); + assert_eq!(linked(&hub, &account).0, Link::Expired); + recovery::hint(&ctx.stop); + assert!(attempts.try_recv().is_err(), "an expired session is never retried by a hint"); + } + + #[test] + fn recovery_during_a_failed_connect_is_not_lost_before_the_wait() { + let (_adapter, ctx, _inbox) = local(AccountKey { network: Network::Signal, account: "during".into() }, hub()); + let generation = ctx.stop.recovery.load(Ordering::Relaxed); + recovery::hint(&ctx.stop); + assert_eq!(ctx.reconnect_pause(Duration::from_secs(900), generation), ReconnectWait::Recovered); + let generation = ctx.stop.recovery.load(Ordering::Relaxed); + assert_eq!(ctx.reconnect_pause(Duration::ZERO, generation), ReconnectWait::Elapsed, "a consumed hint does not wake every later wait"); + } + + #[test] + fn stopping_an_adapter_interrupts_its_reconnect_wait() { + let (adapter, ctx, _inbox) = local(AccountKey { network: Network::Signal, account: "stop".into() }, hub()); + let thread = std::thread::spawn(move || ctx.reconnect_pause(Duration::from_secs(900), 0)); + drop(adapter); + assert_eq!(thread.join().unwrap(), ReconnectWait::Stopped); + } + + #[test] + fn recovery_hints_do_not_bypass_http_or_command_pacing() { + let (_adapter, ctx, _inbox) = local(AccountKey { network: Network::Signal, account: "pacing".into() }, hub()); + let thread_ctx = ctx.clone(); + let (done, finished) = channel(); + let thread = std::thread::spawn(move || { + assert!(!thread_ctx.pause(Duration::from_millis(150))); + done.send(()).unwrap(); + }); + recovery::hint(&ctx.stop); + assert!(finished.recv_timeout(Duration::from_millis(50)).is_err()); + finished.recv_timeout(Duration::from_secs(2)).unwrap(); + thread.join().unwrap(); + } + fn hub() -> Arc { Arc::new(Hub::new(Store::open(std::path::Path::new(":memory:")).expect("a store in memory"))) } diff --git a/crates/chatd/src/adapters/recovery.rs b/crates/chatd/src/adapters/recovery.rs new file mode 100644 index 0000000..65800a1 --- /dev/null +++ b/crates/chatd/src/adapters/recovery.rs @@ -0,0 +1,120 @@ +//! OS hints wake reconnect backoff; they never declare an account connected. +//! Linux uses GIO's network monitor (netlink without NetworkManager) and login1 +//! resume signals. One event loop serves all accounts without periodic probes. + +use super::Halt; +#[cfg(any(target_os = "linux", test))] +use crate::hub::lock; +#[cfg(any(target_os = "linux", test))] +use std::sync::atomic::Ordering; + +#[cfg(any(target_os = "linux", test))] +pub(super) fn hint(stop: &Halt) { + // Share the waiter's mutex so a notification cannot fall between checking + // the generation and entering the condvar wait. + let _waiting = lock(&stop.waiting); + stop.recovery.fetch_add(1, Ordering::Relaxed); + stop.wake.notify_all(); +} + +#[cfg(target_os = "linux")] +mod linux { + use super::*; + use gio::prelude::*; + use std::sync::{Arc, Mutex, Once, OnceLock, Weak}; + + static ACCOUNTS: OnceLock>>> = OnceLock::new(); + static START: Once = Once::new(); + + fn changed() { + let mut accounts = lock(ACCOUNTS.get_or_init(Mutex::default)); + accounts.retain(|account| { + let Some(account) = account.upgrade() else { return false; }; + hint(&account); + true + }); + } + + pub(super) fn watch(stop: &Arc) { + { + let mut accounts = lock(ACCOUNTS.get_or_init(Mutex::default)); + accounts.retain(|account| account.strong_count() != 0); + accounts.push(Arc::downgrade(stop)); + } + START.call_once(|| { + if let Err(err) = std::thread::Builder::new().name("network-recovery".into()).spawn(run) { + eprintln!("chatd: could not watch network recovery: {err}"); + } + }); + } + + fn run() { + let context = gio::glib::MainContext::new(); + let result = context.with_thread_default(|| { + let monitor = gio::NetworkMonitor::default(); + let handler = monitor.connect_network_changed(|_, available| { + if available { changed(); } + }); + // Keep the system bus and its subscription alive with this loop. + // Network recovery still works when login1/system D-Bus is absent. + let bus = gio::bus_get_sync(gio::BusType::System, gio::Cancellable::NONE).ok(); + let subscription = bus.as_ref().map(|bus| bus.signal_subscribe( + Some("org.freedesktop.login1"), Some("org.freedesktop.login1.Manager"), + Some("PrepareForSleep"), Some("/org/freedesktop/login1"), None, + gio::DBusSignalFlags::NONE, + |_, _, _, _, _, args| { + if args.get::<(bool,)>() == Some((false,)) { changed(); } + }, + )); + gio::glib::MainLoop::new(Some(&context), false).run(); + if let (Some(bus), Some(subscription)) = (bus, subscription) { bus.signal_unsubscribe(subscription); } + monitor.disconnect(handler); + }); + if let Err(err) = result { eprintln!("chatd: could not watch network recovery: {err}"); } + } +} + +#[cfg(target_os = "linux")] +pub(super) fn watch(stop: &std::sync::Arc) { linux::watch(stop); } + +#[cfg(not(target_os = "linux"))] +pub(super) fn watch(_: &std::sync::Arc) {} + +#[cfg(all(test, target_os = "linux"))] +mod tests { + use super::*; + use gio::prelude::*; + use std::sync::Arc; + use std::time::{Duration, Instant}; + + #[test] + #[ignore = "requires an isolated network namespace; see docs/testing-chatd.md"] + fn kernel_route_recovery_without_networkmanager_wakes_reconnect() { + assert_eq!(std::env::var("GIO_USE_NETWORK_MONITOR").as_deref(), Ok("netlink")); + assert_ne!(std::fs::read_link("/proc/self/ns/net").unwrap(), std::fs::read_link("/proc/1/ns/net").unwrap(), + "never modify the host network: run this test under unshare --net"); + let context = gio::glib::MainContext::new(); + context.with_thread_default(|| { + let monitor = gio::NetworkMonitor::default(); + assert_eq!(monitor.type_().name(), "GNetworkMonitorNetlink"); + let stop = Arc::new(Halt::default()); + let recovery = stop.clone(); + let handler = monitor.connect_network_changed(move |_, available| { + if available { hint(&recovery); } + }); + for args in [vec!["link", "add", "testnet0", "type", "dummy"], + vec!["link", "set", "testnet0", "up"], + vec!["addr", "add", "192.0.2.2/24", "dev", "testnet0"], + vec!["route", "add", "default", "dev", "testnet0"]] { + assert!(std::process::Command::new("ip").args(args).status().unwrap().success()); + } + let deadline = Instant::now() + Duration::from_secs(3); + while stop.recovery.load(Ordering::Relaxed) == 0 { + while context.pending() { context.iteration(false); } + assert!(Instant::now() < deadline, "kernel route notification never reached the reconnect wakeup"); + std::thread::sleep(Duration::from_millis(1)); + } + monitor.disconnect(handler); + }).unwrap(); + } +} diff --git a/docs/reconnecting.md b/docs/reconnecting.md new file mode 100644 index 0000000..a24a757 --- /dev/null +++ b/docs/reconnecting.md @@ -0,0 +1,61 @@ +# Reconnecting after a network outage + +The shared connection loop used by Signal, Discord, Slack, Teams and Messages +retries temporary failures with exponential backoff and jitter, up to 15 minutes. +An expired or explicitly revoked session stops retrying and keeps its cache. + +On Linux, chatd also listens for OS recovery hints. GIO's `GNetworkMonitor` +reports network configuration changes with an available route; its Linux +backends include netlink, so NetworkManager is not required. A system D-Bus +subscription to login1's `PrepareForSleep(false)` covers suspend/hibernate resume. +One GLib event thread serves all accounts and performs no periodic probes. + +A hint interrupts an adapter's reconnect wait and resets its accumulated delay. +Attempts stay at least one second apart, including notification bursts. A hint +received during an unsuccessful connection attempt is retained for the following +wait. Ordinary HTTP/command pacing and panic recovery are not interrupted. +Healthy connections are not closed, and hints never mark an account Connected: +only its adapter's successful connection does that. Subsequent failures still +back off, and expired sessions are never revived by a recovery hint. + +This detects local route changes and resume, not remote service restoration or +DNS recovery with no OS network event. Missing login1/system D-Bus leaves the +network monitor available. Other platforms retain timed reconnection until they +have a native recovery-event implementation. GIO is a Linux-only dependency; +the Linux GUI packages already supply its runtime libraries. + +Recovery produces a journal entry such as: + +```text +chatd: Signal: network changed or system resumed; retrying connection +``` + +After updating chatd, restart the user service once to load the new code: + +```sh +systemctl --user restart tinyclient-chatd.service +``` + +## Validation + +The daemon unit suite covers interrupting a simulated 900-second reconnect wait, +a burst of recovery hints, a hint arriving before the wait starts, normal pacing, +stop notification and expired-session behavior. + +A Linux integration check exercises actual kernel route notification through the +netlink backend in a disposable network namespace. It explicitly refuses to run +in the same network namespace as PID 1, so it cannot modify host networking: + +```sh +cargo test --frozen --release -p chatd --lib --no-run +sudo unshare --net env GIO_USE_NETWORK_MONITOR=netlink \ + target/release/deps/chatd- \ + adapters::recovery::tests::kernel_route_recovery_without_networkmanager_wakes_reconnect \ + --exact --ignored --nocapture +``` + +Use the test executable path printed by Cargo. The manual test requires `ip` +and permission to create a network namespace; ordinary unit tests do not. + +References: [GIO network monitoring](https://docs.gtk.org/gio/class.NetworkMonitor.html) +and [login1 resume signals](https://www.freedesktop.org/software/systemd/man/latest/org.freedesktop.login1.html).