diff --git a/docs/architecture.md b/docs/architecture.md index c06bc28..cc1c184 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -88,7 +88,8 @@ printf '{"cmd":"fetch","force":true}\n' | socat - UNIX-CONNECT:$XDG_RUNTIME_DIR/ `download_error`, `torrent_deferred`, `reaped`, `reap_done`, `scan_done`, `status`, `error`. `scan_done`, `reap_done` and `status` are terminal: a client that asked for work stops reading there. Commands run one at a time, in the order they arrive, except `status`: the socket answers it -straight away, so the Docker healthcheck is never left waiting behind a scan or a download. +straight away, so the Docker healthcheck is never left waiting behind a scan or a download, and +answers only the client that asked, since `status` would end any other client's session. Progress carries the enclosure id, without which a UI cannot tell one download from another and ends up animating every pending row. It is throttled to whole percents. The stream is a broadcast, diff --git a/docs/history.md b/docs/history.md index 61d97a7..4f587e0 100644 --- a/docs/history.md +++ b/docs/history.md @@ -22,6 +22,13 @@ database; a worker stuck on one job would still pass. Asking a daemon that downl podcasts to be idle within five seconds was never a fair test of whether it was alive. A test holds the queue full and checks `status` still comes back. +The first version broadcast the answer, as the queued one had been. Timing `status` during a forced +scan in production showed the catch: `status` is a terminal event, so the `ipx fetch` watching that +scan stopped reading at the first probe and printed the status line as its last, while the scan +carried on. When `status` waited behind the scan it could never arrive first, so this had never +shown. The answer now goes only to the client that asked, and the test checks that another client +hears nothing. + ## 2026-09-12 — Signing in through Authentik, for real Ray could not get Authentik's sign-in to reach ipx, following `docs/sso.md`, which had been written diff --git a/src/ipc.rs b/src/ipc.rs index a55977b..453f404 100644 --- a/src/ipc.rs +++ b/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; +/// 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 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::(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::(1); cmds.send(Command::Reap { dry_run: true }).await.unwrap(); let (events, _) = broadcast::channel::(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"); } } diff --git a/src/main.rs b/src/main.rs index 7e64d20..7d7a3d8 100644 --- a/src/main.rs +++ b/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));