Retry Linux connections promptly after network recovery or resume

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"] }

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<E: std::fmt::Display> From<E> 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<String>
 /// Starts the adapter for an account from its part of the session.
 pub fn start(account: AccountKey, session: &str, hub: Arc<Hub>) -> Result<Adapter, String> {
     let stop = Arc::new(Halt::default());
+    recovery::watch(&stop);
     let (commands, inbox): (Sender<Command>, Receiver<Command>) = 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<Hub> {
         Arc::new(Hub::new(Store::open(std::path::Path::new(":memory:")).expect("a store in memory")))
     }

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<Mutex<Vec<Weak<Halt>>>> = 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<Halt>) {
+        {
+            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<Halt>) { linux::watch(stop); }
+
+#[cfg(not(target_os = "linux"))]
+pub(super) fn watch(_: &std::sync::Arc<Halt>) {}
+
+#[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();
+    }
+}

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-<test-hash> \
+  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).