Answer status to the client that asked, not everyone

status is a terminal event. Broadcast, the healthcheck's answer ended any
ipx fetch that was watching a scan, which stopped reading at the next probe
while the scan carried on. It could not happen while status waited behind
the scan; answering it at once made it happen every 30 seconds. Each
connection's writer now takes private replies beside the broadcast, and
the test checks another client hears nothing.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TAC7sLVqfKmY6rsTLXzNgk

src/ipc.rs

@@ -174,8 +174,9 @@ pub async fn daemon_is_live(path: &Path) -> bool {
 
 /// Answers `status` for the socket, without the worker. The worker runs one job at a time, and a
 /// healthcheck left waiting behind a scan or a long download timed out and called a busy daemon
-/// dead.
-pub type StatusFn = std::sync::Arc<dyn Fn() + Send + Sync>;
+/// dead. The answer goes to the client that asked and no one else: broadcast, it ended any
+/// `ipx fetch` that was watching a scan, since `status` is a terminal event.
+pub type StatusFn = std::sync::Arc<dyn Fn() -> Event + Send + Sync>;
 
 /// Accepts connections, feeding commands to `cmds` and events from `events` back out.
 pub async fn serve(
@@ -218,9 +219,17 @@ async fn handle(
 ) -> Result<()> {
     let (read, mut write) = stream.into_split();
 
-    // Events out.
+    // Events out: everything broadcast, and the answers meant for this client alone.
+    let (reply, mut replies) = mpsc::channel::<Event>(4);
     let writer = tokio::spawn(async move {
-        while let Ok(ev) = rx.recv().await {
+        loop {
+            let ev = tokio::select! {
+                Some(ev) = replies.recv() => ev,
+                got = rx.recv() => match got {
+                    Ok(ev) => ev,
+                    Err(_) => break,
+                },
+            };
             let mut line = serde_json::to_string(&ev).unwrap_or_default();
             line.push('\n');
             if write.write_all(line.as_bytes()).await.is_err() {
@@ -240,7 +249,11 @@ async fn handle(
             // Answered here, not queued behind whatever the worker is on: see StatusFn.
             Ok(Command::Status) => {
                 tracing::info!(target: "ipx::io", "-> {line}");
-                status();
+                let ev = status();
+                if let Ok(json) = serde_json::to_string(&ev) {
+                    tracing::info!(target: "ipx::io", "<- {json}");
+                }
+                let _ = reply.send(ev).await;
             }
             Ok(cmd) => {
                 if cmds.send(cmd).await.is_err() {
@@ -350,10 +363,10 @@ mod tests {
         let (cmds, _worker) = mpsc::channel::<Command>(1);
         cmds.send(Command::Reap { dry_run: true }).await.unwrap();
         let (events, _) = broadcast::channel::<Event>(8);
-        let tx = events.clone();
-        let status: StatusFn = std::sync::Arc::new(move || {
-            let _ = tx.send(Event::Status { feeds: 1, pending: 2, downloaded: 3 });
-        });
+        // Another client, watching a scan: it must not be handed someone else's answer, which
+        // would end its session.
+        let mut watcher = events.subscribe();
+        let status: StatusFn = std::sync::Arc::new(|| Event::Status { feeds: 1, pending: 2, downloaded: 3 });
         let (client, server) = UnixStream::pair().unwrap();
         tokio::spawn(handle(server, events.subscribe(), cmds, status));
 
@@ -365,5 +378,6 @@ mod tests {
             .unwrap()
             .unwrap();
         assert!(line.contains(r#""ev":"status""#), "{line}");
+        assert!(watcher.try_recv().is_err(), "the answer went to every client, not just the one asking");
     }
 }

src/main.rs

@@ -398,7 +398,7 @@ async fn daemon(
     // status is answered by the socket itself; everything else waits its turn in the queue.
     let answer: ipc::StatusFn = {
         let ctx = ctx.clone();
-        Arc::new(move || ctx.out.emit(status(&ctx)))
+        Arc::new(move || status(&ctx))
     };
     let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd, answer));