Add a log view, and Docker packaging

The Log button shows the running daemon live: feed scans, downloads,
torrents and every HTTP request. It reads a ring buffer filled by a
tracing layer rather than tailing a file, so it works under Docker where
logs go to stdout. The access-log middleware skips /api/logs, or the
panel's poll would log itself forever.

Detached torrents could leave a row stuck in 'downloading' across a
restart, where nothing would ever revisit it; those are requeued at
startup.

Dockerfile, entrypoint and compose: 114 MB runtime, config bound to
0.0.0.0 on first run since container loopback is unreachable, drops to
PUID:PGID for Unraid, and a healthcheck that goes through the control
socket so a wedged worker reads as unhealthy.

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

src/db.rs

@@ -703,6 +703,17 @@ impl Db {
         Ok(())
     }
 
+    /// Nothing can be in flight the moment the daemon starts, so any row still marked
+    /// `downloading` is a leftover from a restart or a crash. Left alone it would sit
+    /// there forever: the pending queue skips it and nothing else ever revisits it.
+    pub fn requeue_interrupted(&self) -> Result<usize> {
+        let conn = self.conn.lock().unwrap();
+        Ok(conn.execute(
+            "UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL",
+            [],
+        )?)
+    }
+
     /// Puts an enclosure back in the queue so the next scan picks it up. This is how a
     /// `skipped` verdict (from a filter that has since been changed) gets revisited.
     pub fn requeue(&self, id: i64) -> Result<()> {
@@ -782,6 +793,27 @@ mod tests {
             "search is case-insensitive and covers the description");
     }
 
+    #[test]
+    fn a_restart_requeues_interrupted_downloads() {
+        let db = Db::memory().unwrap();
+        db.exec_for_test(
+            "INSERT INTO enclosures (id, feed_id, guid, url, state, path) VALUES
+                (1,'f','a','u1','downloading',NULL),
+                (2,'f','b','u2','pending',NULL),
+                (3,'f','c','u3','downloading','/tmp/already-here'),
+                (4,'f','d','u4','done','/tmp/x');",
+        )
+        .unwrap();
+        assert_eq!(db.requeue_interrupted().unwrap(), 1, "only the in-flight, fileless one");
+        let conn = db.conn.lock().unwrap();
+        let state = |id: i64| -> String {
+            conn.query_row("SELECT state FROM enclosures WHERE id = ?1", [id], |r| r.get(0)).unwrap()
+        };
+        assert_eq!(state(1), "pending");
+        assert_eq!(state(3), "downloading", "it has a file; leave it alone");
+        assert_eq!(state(4), "done");
+    }
+
     #[test]
     fn enclosure_url_is_the_dedupe_key() {
         let db = Db::memory().unwrap();

src/logbuf.rs

@@ -0,0 +1,154 @@
+//! In-process ring buffer of log lines, so the UI can show what the daemon is doing.
+//!
+//! Tailing a file would not survive Docker, where logs go to stdout and there is no file
+//! to read. Capturing inside the tracing pipeline works the same either way.
+
+use std::collections::VecDeque;
+use std::sync::{LazyLock, Mutex};
+use tracing::field::{Field, Visit};
+use tracing_subscriber::Layer;
+use tracing_subscriber::layer::Context;
+
+/// Kept small enough to be cheap to hold and to serialise in one response.
+const CAPACITY: usize = 2000;
+
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct LogLine {
+    /// Monotonic, so a client can ask for "everything after N" without duplicates.
+    pub seq: u64,
+    pub ts: i64,
+    pub level: String,
+    pub target: String,
+    pub msg: String,
+}
+
+struct Ring {
+    lines: VecDeque<LogLine>,
+    next_seq: u64,
+}
+
+static BUF: LazyLock<Mutex<Ring>> = LazyLock::new(|| {
+    Mutex::new(Ring { lines: VecDeque::with_capacity(CAPACITY), next_seq: 1 })
+});
+
+pub fn push(level: &str, target: &str, msg: String) {
+    let mut ring = match BUF.lock() {
+        Ok(r) => r,
+        Err(p) => p.into_inner(), // a poisoned log buffer must not take the process down
+    };
+    let seq = ring.next_seq;
+    ring.next_seq += 1;
+    if ring.lines.len() == CAPACITY {
+        ring.lines.pop_front();
+    }
+    ring.lines.push_back(LogLine {
+        seq,
+        ts: crate::db::now(),
+        level: level.to_owned(),
+        target: target.to_owned(),
+        msg,
+    });
+}
+
+/// Lines newer than `after`, oldest first, plus the highest seq now held.
+pub fn since(after: u64, limit: usize) -> (Vec<LogLine>, u64) {
+    let ring = match BUF.lock() {
+        Ok(r) => r,
+        Err(p) => p.into_inner(),
+    };
+    let latest = ring.next_seq.saturating_sub(1);
+    let mut out: Vec<LogLine> = ring
+        .lines
+        .iter()
+        .filter(|l| l.seq > after)
+        .cloned()
+        .collect();
+    // On a first load (after = 0) the tail is what matters, not the head.
+    if out.len() > limit {
+        out.drain(..out.len() - limit);
+    }
+    (out, latest)
+}
+
+/// A tracing layer that mirrors every event into the ring.
+pub struct RingLayer;
+
+impl<S: tracing::Subscriber> Layer<S> for RingLayer {
+    fn on_event(&self, event: &tracing::Event<'_>, _ctx: Context<'_, S>) {
+        let mut v = Collect::default();
+        event.record(&mut v);
+        let meta = event.metadata();
+        push(meta.level().as_str(), meta.target(), v.finish());
+    }
+}
+
+#[derive(Default)]
+struct Collect {
+    message: String,
+    fields: Vec<String>,
+}
+
+impl Collect {
+    fn finish(self) -> String {
+        if self.fields.is_empty() {
+            self.message
+        } else if self.message.is_empty() {
+            self.fields.join(" ")
+        } else {
+            format!("{} {}", self.message, self.fields.join(" "))
+        }
+    }
+    fn add(&mut self, field: &Field, value: String) {
+        if field.name() == "message" {
+            self.message = value;
+        } else {
+            self.fields.push(format!("{}={}", field.name(), value));
+        }
+    }
+}
+
+impl Visit for Collect {
+    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+        self.add(field, format!("{value:?}"));
+    }
+    fn record_str(&mut self, field: &Field, value: &str) {
+        self.add(field, value.to_owned());
+    }
+    fn record_i64(&mut self, field: &Field, value: i64) {
+        self.add(field, value.to_string());
+    }
+    fn record_u64(&mut self, field: &Field, value: u64) {
+        self.add(field, value.to_string());
+    }
+    fn record_bool(&mut self, field: &Field, value: bool) {
+        self.add(field, value.to_string());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn the_ring_drops_oldest_and_keeps_sequence_stable() {
+        for i in 0..(CAPACITY + 50) {
+            push("INFO", "t", format!("line {i}"));
+        }
+        let (all, latest) = since(0, CAPACITY * 2);
+        assert_eq!(all.len(), CAPACITY, "bounded");
+        assert!(latest >= (CAPACITY + 50) as u64);
+        assert!(
+            all.first().unwrap().seq < all.last().unwrap().seq,
+            "oldest first"
+        );
+
+        // "everything after the last one I saw" must return nothing new.
+        let (none, _) = since(latest, 100);
+        assert!(none.is_empty());
+
+        // A first load takes the tail, not the head.
+        let (tail, _) = since(0, 5);
+        assert_eq!(tail.len(), 5);
+        assert_eq!(tail.last().unwrap().seq, latest);
+    }
+}

src/main.rs

@@ -3,6 +3,7 @@ mod db;
 mod download;
 mod feed;
 mod ipc;
+mod logbuf;
 mod retention;
 mod torrent;
 mod web;
@@ -119,13 +120,19 @@ impl Ctx {
 #[tokio::main]
 async fn main() -> Result<()> {
     let cli = Cli::parse();
-    tracing_subscriber::fmt()
-        .with_env_filter(
-            tracing_subscriber::EnvFilter::try_from_env("IPX_LOG")
-                .unwrap_or_else(|_| "ipx=info".into()),
-        )
-        .with_writer(std::io::stderr)
-        .init();
+    // Everything goes to stderr as before, and is mirrored into a ring the UI can read.
+    {
+        use tracing_subscriber::layer::SubscriberExt;
+        use tracing_subscriber::util::SubscriberInitExt;
+        tracing_subscriber::registry()
+            .with(
+                tracing_subscriber::EnvFilter::try_from_env("IPX_LOG")
+                    .unwrap_or_else(|_| "ipx=info".into()),
+            )
+            .with(tracing_subscriber::fmt::layer().with_writer(std::io::stderr))
+            .with(logbuf::RingLayer)
+            .init();
+    }
 
     let config_path = cli.config.clone().unwrap_or_else(config::config_path);
     let cfg = config::Config::load(&config_path)?;
@@ -207,6 +214,12 @@ async fn daemon(
         anyhow::bail!("a daemon is already listening on {}", socket.display());
     }
 
+    match ctx.db.requeue_interrupted() {
+        Ok(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"),
+        Ok(_) => {}
+        Err(e) => tracing::warn!(error = ?e, "could not requeue interrupted downloads"),
+    }
+
     let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(64);
 
     let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?;

src/web.rs

@@ -72,9 +72,11 @@ pub fn router(state: WebState) -> Router {
         .route("/api/fetch", post(fetch_now))
         .route("/api/opml", get(export_opml).post(import_opml))
         .route("/api/settings", get(get_settings).patch(patch_settings))
+        .route("/api/logs", get(logs))
         .route("/api/events", get(events))
         .route("/media/{id}", get(media))
         .layer(middleware::from_fn_with_state(state.clone(), auth))
+        .layer(middleware::from_fn(access_log))
         .with_state(state)
 }
 
@@ -790,3 +792,49 @@ async fn patch_settings(
     state.ctx.reload_cfg(&state.config_path)?;
     Ok(StatusCode::NO_CONTENT)
 }
+
+#[derive(Deserialize)]
+struct LogQuery {
+    /// Highest seq the client already has; 0 means "give me the tail".
+    #[serde(default)]
+    after: u64,
+    #[serde(default = "two_hundred")]
+    limit: usize,
+}
+
+fn two_hundred() -> usize {
+    200
+}
+
+#[derive(Serialize)]
+struct LogPage {
+    lines: Vec<crate::logbuf::LogLine>,
+    latest: u64,
+}
+
+async fn logs(Query(q): Query<LogQuery>) -> Json<LogPage> {
+    let (lines, latest) = crate::logbuf::since(q.after, q.limit.clamp(1, 2000));
+    Json(LogPage { lines, latest })
+}
+
+/// One line per HTTP request, so the web side shows up in the same log as the daemon.
+///
+/// The log view polls `/api/logs`, so logging that path would generate a line per poll
+/// forever -- a feed of nothing but its own requests.
+async fn access_log(req: Request, next: Next) -> Response {
+    let path = req.uri().path().to_owned();
+    let method = req.method().clone();
+    let quiet = path.starts_with("/api/logs");
+    let started = std::time::Instant::now();
+    let resp = next.run(req).await;
+    if !quiet {
+        let ms = started.elapsed().as_millis();
+        let status = resp.status().as_u16();
+        if resp.status().is_success() || resp.status().is_redirection() {
+            tracing::info!(target: "ipx::http", "{method} {path} -> {status} in {ms}ms");
+        } else {
+            tracing::warn!(target: "ipx::http", "{method} {path} -> {status} in {ms}ms");
+        }
+    }
+    resp
+}