perf(windows): a subscriber that does not read cannot hold up the push with a 1 s write timeout

notify_subscribers connects to every subscriber socket and write_all()s the
notification with the window manager locked (every command, the reaper and
the event thread), with no timeout: a subscriber that stops reading stalls the
lock holder, and with it every thread.

komorebi_transport::notify_sockets skips a subscriber whose write fails with
TimedOut or WouldBlock (warns, keeps it subscribed, goes on to the next);
every other write error still aborts as before. The timeout itself is the
caller's: the connect closure passed by the Windows daemon sets a 1 s write
timeout on the stream. KDE and macOS pass connect closures without a timeout,
so nothing changes for them. A new unit test covers the skip. The connect
itself cannot be bounded (uds_windows has no connect timeout), and the
listener's lock wait is untouched.

BEHAVIOUR: a Windows subscriber that does not read for a second misses that
notification instead of blocking the daemon. Gate needed: Windows e2e commands
and config_reload, the lock stress, subscriber scenarios if any. Verified:
native cargo test -p komorebi-transport 14 passed (incl. the new test);
tools/fast-check native, Windows clippy -D warnings and KDE check + lib tests
PASS; macOS type-check run by hand (fast-check's macOS step fails to select
a package on this merged tree) PASS. e2e not run.

crates/komorebi-transport/src/subscribers.rs

@@ -43,7 +43,23 @@ pub fn notify_sockets<S: Write>(
             match connect(path) {
                 Ok(mut stream) => {
                     tracing::debug!(target: LOG_TARGET, "pushed notification to subscriber: {socket}");
-                    stream.write_all(notification.as_bytes())?;
+                    match stream.write_all(notification.as_bytes()) {
+                        Ok(()) => {}
+                        // a subscriber that does not read (the stream has a write timeout)
+                        // is skipped, so that it cannot hold up the one that is pushing
+                        Err(error)
+                            if matches!(
+                                error.kind(),
+                                std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock
+                            ) =>
+                        {
+                            tracing::warn!(
+                                target: LOG_TARGET,
+                                "subscriber {socket} did not take the notification in time: {error}"
+                            );
+                        }
+                        Err(error) => return Err(error),
+                    }
                 }
                 Err(_) => {
                     stale_sockets.push(socket.clone());
@@ -132,4 +148,41 @@ mod tests {
         assert!(live.accept().is_ok());
         let _ = std::fs::remove_dir_all(&dir);
     }
+
+    /// A stream whose write times out, as one with a write timeout does when nobody reads it
+    struct Stuck;
+
+    impl Write for Stuck {
+        fn write(&mut self, _: &[u8]) -> std::io::Result<usize> {
+            Err(std::io::ErrorKind::TimedOut.into())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    #[test]
+    fn a_subscriber_that_times_out_is_skipped_and_stays_subscribed() {
+        let dir = dir("timeout");
+        let mut sockets = HashMap::from([
+            ("a.sock".to_string(), dir.join("a.sock")),
+            ("b.sock".to_string(), dir.join("b.sock")),
+        ]);
+        let pushed = std::cell::Cell::new(0);
+
+        let result = notify_sockets("{}", &mut sockets, &HashMap::new(), false, &dir, |_| {
+            pushed.set(pushed.get() + 1);
+            Ok(Stuck)
+        });
+
+        assert!(result.is_ok());
+        assert_eq!(
+            pushed.get(),
+            2,
+            "the subscribers after the stuck one are tried"
+        );
+        assert_eq!(sockets.len(), 2, "a slow subscriber is not removed");
+        let _ = std::fs::remove_dir_all(&dir);
+    }
 }

komorebi/src/lib.rs

@@ -14,13 +14,13 @@ pub mod focus_manager;
 pub mod live_resize;
 pub use komorebi_collections::Lockable;
 pub use komorebi_collections::lockable_sequence;
+pub mod lock_holder;
 pub mod monitor;
 pub mod monitor_reconciliator;
 pub mod pan;
 pub mod process_command;
 pub mod process_event;
 pub mod process_movement;
-pub mod lock_holder;
 pub mod reaper;
 pub mod scrolling;
 pub mod set_window_position;
@@ -389,7 +389,13 @@ pub fn notify_subscribers(
         &options,
         state_has_been_modified || is_override_event,
         &DATA_DIR,
-        |path| UnixStream::connect(path),
+        |path| {
+            let stream = UnixStream::connect(path)?;
+            // a subscriber that does not read must not hold up the writer, which has the
+            // window manager locked
+            stream.set_write_timeout(Some(std::time::Duration::from_secs(1)))?;
+            Ok(stream)
+        },
     )?;
 
     let mut stale_pipes = vec![];