Answer status on the socket instead of queuing it behind the worker

The worker runs one job at a time, and status was one of its jobs, so the
Docker healthcheck waited behind the startup scan (54 seconds of it after
the last deploy) and timed out at 5. Any scan or download longer than
three probes would have had a working daemon marked unhealthy. The socket
now answers status straight away; everything else still queues. A test
fills the queue and checks status comes back anyway.

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

src/ipc.rs

@@ -172,11 +172,17 @@ pub async fn daemon_is_live(path: &Path) -> bool {
     UnixStream::connect(path).await.is_ok()
 }
 
+/// 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>;
+
 /// Accepts connections, feeding commands to `cmds` and events from `events` back out.
 pub async fn serve(
     path: PathBuf,
     events: broadcast::Sender<Event>,
     cmds: mpsc::Sender<Command>,
+    status: StatusFn,
 ) -> Result<()> {
     // A socket file left by a crashed daemon would block the bind; a live one was already
     // rejected by the caller's daemon_is_live() check.
@@ -195,8 +201,9 @@ pub async fn serve(
         let (stream, _) = listener.accept().await?;
         let rx = events.subscribe();
         let cmds = cmds.clone();
+        let status = status.clone();
         tokio::spawn(async move {
-            if let Err(e) = handle(stream, rx, cmds).await {
+            if let Err(e) = handle(stream, rx, cmds, status).await {
                 tracing::debug!(error = %e, "client gone");
             }
         });
@@ -207,6 +214,7 @@ async fn handle(
     stream: UnixStream,
     mut rx: broadcast::Receiver<Event>,
     cmds: mpsc::Sender<Command>,
+    status: StatusFn,
 ) -> Result<()> {
     let (read, mut write) = stream.into_split();
 
@@ -229,6 +237,11 @@ async fn handle(
             continue;
         }
         match serde_json::from_str::<Command>(line) {
+            // Answered here, not queued behind whatever the worker is on: see StatusFn.
+            Ok(Command::Status) => {
+                tracing::info!(target: "ipx::io", "-> {line}");
+                status();
+            }
             Ok(cmd) => {
                 if cmds.send(cmd).await.is_err() {
                     break; // Worker is gone; so are we.
@@ -329,4 +342,28 @@ mod tests {
         }
         .is_terminal());
     }
+
+    #[tokio::test]
+    async fn status_is_answered_while_the_worker_is_busy() {
+        // The queue is full and nobody drains it, as when the worker is deep in a long download:
+        // anything sent to it would wait for ever.
+        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 });
+        });
+        let (client, server) = UnixStream::pair().unwrap();
+        tokio::spawn(handle(server, events.subscribe(), cmds, status));
+
+        let (read, mut write) = client.into_split();
+        write.write_all(b"{\"cmd\":\"status\"}\n").await.unwrap();
+        let line = tokio::time::timeout(std::time::Duration::from_secs(2), BufReader::new(read).lines().next_line())
+            .await
+            .expect("status waited behind the worker")
+            .unwrap()
+            .unwrap();
+        assert!(line.contains(r#""ev":"status""#), "{line}");
+    }
 }

src/main.rs

@@ -339,14 +339,24 @@ async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> {
         Cmd::Reap { dry_run } => reap(ctx, dry_run, true),
         Cmd::Download { enclosure } => download_one(ctx, enclosure).await,
         Cmd::Status => {
-            let (pending, downloaded) = ctx.db.counts()?;
-            let feeds = subscriptions(ctx).map(|s| s.len()).unwrap_or(0);
-            ctx.out.emit(Event::Status { feeds, pending, downloaded });
+            ctx.out.emit(status(ctx));
             Ok(())
         }
     }
 }
 
+/// The counts `ipx status` prints. A running daemon's socket answers with this directly rather
+/// than through the job queue.
+fn status(ctx: &Ctx) -> Event {
+    match ctx.db.counts() {
+        Ok((pending, downloaded)) => {
+            let feeds = subscriptions(ctx).map(|s| s.len()).unwrap_or(0);
+            Event::Status { feeds, pending, downloaded }
+        }
+        Err(e) => Event::Error { msg: format!("{e:#}") },
+    }
+}
+
 async fn daemon(
     ctx: Arc<Ctx>,
     config_path: PathBuf,
@@ -385,7 +395,12 @@ async fn daemon(
     let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);
 
     let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?;
-    let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd));
+    // 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)))
+    };
+    let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd, answer));
 
     // One command at a time: the queue is what keeps two scans from overlapping.
     let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60));