A download a restart stopped is taken up again at the start (#410)

A file download was never marked 'downloading', only torrents were, so
one under way when the daemon stopped stayed 'pending' and
requeue_interrupted had nothing to do. The first pass after a start
scans only the feeds that are due, and a pending file is fetched only
in a scan of its feed, so after a deploy a half-downloaded episode
waited for its feed's next check: an hour on the default schedule, a
day for a daily feed. Found by the daemon test of #400, where the
episode was not asked for again in a minute after a restart.

fetch_one now marks the file 'downloading' while it runs, as a
torrent's is; every way out of it already records 'done' or 'error',
and the scan's queue takes only 'pending', so a running download is
not queued twice. requeue_interrupted gives back the ids it put back,
and the daemon puts each on the worker's queue before it serves, where
download_one takes it whatever its feed's schedule says. The daemon
test now checks the restarted daemon asks for the episode again.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

crates/ipx/CHANGELOG.md

@@ -26,6 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- A download a restart stopped, a deploy's for one, is taken up again as the daemon starts, rather than waiting for its feed's next check, an hour or a day away (#410).
 - A YouTube channel no longer misses most days' videos: YouTube answers some feed requests with a 404 or 500 it does not mean, and ipx now asks again, twice, a second apart, before calling the feed failing (#382), and then every hour for a day where it would have waited a day, since YouTube's failures run for hours; and its YouTube feeds are read one at a time, two seconds apart, since YouTube refuses many asked together (#386).
 - Syncing an OPML or Patreon list reads who subscribes to it once, not once for every feed it lists: a list of 922 was thousands of the same queries a sync (#377).
 - A feed you already have, added again by the same address over http rather than https or the other way round, is the one you have, not a second subscription with every item twice. Changing a feed's address to another feed's over the other scheme is refused the same way (#362).

crates/ipx/src/daemon.rs

@@ -59,7 +59,7 @@ pub(crate) async fn daemon(
 
     // The rest of start-up, in the trace main began (#374): how long a deploy took to answer, and
     // what held it.
-    async {
+    let interrupted = async {
         // A database with nobody in it cannot be signed into.
         if ctx.db.users().instrument(db::step("users")).await?.is_empty() {
             let dir = ctx.config_path.parent().unwrap_or(Path::new("."));
@@ -84,11 +84,13 @@ pub(crate) async fn daemon(
             }
         }
 
-        match ctx.db.requeue_interrupted().instrument(db::step("requeue_interrupted")).await {
-            Ok(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"),
-            Ok(_) => {}
-            Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not requeue interrupted downloads"),
+        let interrupted = match ctx.db.requeue_interrupted().instrument(db::step("requeue_interrupted")).await {
+            Ok(ids) => ids,
+            Err(e) => {
+                tracing::warn!(error = %format!("{e:#}"), "could not requeue interrupted downloads");
+                vec![]
             }
+        };
 
         match catalogue::retire_stranded(&ctx).instrument(db::step("retire_stranded")).await {
             Ok(0) => {}
@@ -102,7 +104,7 @@ pub(crate) async fn daemon(
                 tracing::warn!(error = %format!("{e:#}"), "could not take feeds of their own out of their old list's group")
             }
         }
-        anyhow::Ok(())
+        anyhow::Ok(interrupted)
     }
     .instrument(start.clone())
     .await?;
@@ -112,6 +114,16 @@ pub(crate) async fn daemon(
     // every other request that queues work waited behind them (#154).
     let (tx_cmd, mut rx_cmd) = mpsc::channel::<Cmd>(1024);
 
+    // Taken again first thing: left for their feeds' next check, a download a deploy stopped
+    // waited an hour, or a day for a daily feed (#410).
+    if !interrupted.is_empty() {
+        tracing::info!(count = interrupted.len(), "taking up downloads interrupted by a restart");
+    }
+    for enclosure in interrupted {
+        if tx_cmd.try_send(Cmd::Download { enclosure }).is_err() {
+            break; // The queue is full; the rest wait for their feeds' next check, as before.
+        }
+    }
     let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).instrument(start.clone()).await?;
     // status and reload are answered by the socket itself; everything else waits its turn in the
     // queue.

crates/ipx/src/db.rs

@@ -2635,11 +2635,14 @@ impl Db {
 
     /// 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 async fn requeue_interrupted(&self) -> Result<usize> {
-        Ok(self
-            .exec("UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL", vec![])
-            .await? as usize)
+    /// there forever: the pending queue skips it and nothing else ever revisits it. Gives the
+    /// ones put back, for the daemon to take again at once rather than at their feed's next check.
+    pub async fn requeue_interrupted(&self) -> Result<Vec<i64>> {
+        const LEFT: &str = "state = 'downloading' AND path IS NULL";
+        let rows = self.rows(&format!("SELECT id FROM enclosures WHERE {LEFT} ORDER BY id"), vec![]).await?;
+        let ids = rows.iter().map(|r| r.try_get::<i64>("", "id")).collect::<Result<Vec<_>, _>>()?;
+        self.exec(&format!("UPDATE enclosures SET state = 'pending' WHERE {LEFT}"), vec![]).await?;
+        Ok(ids)
     }
 
     /// Puts an enclosure back in the queue so the next scan picks it up. This is how a
@@ -4114,7 +4117,7 @@ mod tests {
         )
         .await
         .unwrap();
-        assert_eq!(db.requeue_interrupted().await.unwrap(), 1, "only the in-flight, fileless one");
+        assert_eq!(db.requeue_interrupted().await.unwrap(), [1], "only the in-flight, fileless one");
         let state = async |id: i64| db.strings_for_test(&format!("SELECT state FROM enclosures WHERE id = {id}")).await;
         assert_eq!(state(1).await, ["pending"]);
         assert_eq!(state(3).await, ["downloading"], "it has a file; leave it alone");

crates/ipx/src/scan.rs

@@ -717,6 +717,9 @@ pub(crate) async fn fetch_one(
     let mut last_pct = -1i64;
     let name = download::filename_for(url, None);
     let cfg = ctx.cfg();
+    // Said while it runs, as a torrent's is: a stop mid-file leaves it 'downloading', which the
+    // next start takes up at once (#410). Every way out records 'done' or 'error'.
+    ctx.db.mark_enclosure(url, "downloading", None).await?;
     let got = download::download(&ctx.client, &cfg, feed_cfg, url, |done, total| {
         if let Some(t) = total.filter(|t| *t > 0) {
             let pct = (done * 100 / t) as i64;

crates/ipx/tests/daemon.rs

@@ -9,6 +9,12 @@ use std::process::{Child, Command, Output, Stdio};
 use std::time::{Duration, Instant};
 
 const EPISODE_BYTES: u64 = 50_000_000;
+/// Times the episode has been asked for.
+static EPISODE_GETS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+
+fn episode_gets() -> usize {
+    EPISODE_GETS.load(std::sync::atomic::Ordering::SeqCst)
+}
 
 /// A feed of one episode, and the episode, sent 4 KB every 50 ms: a download that takes minutes.
 fn serve_feed() -> u16 {
@@ -28,6 +34,7 @@ fn answer(mut sock: TcpStream, port: u16) {
         return;
     }
     if line.contains("/ep.mp3") {
+        EPISODE_GETS.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
         let head = format!("HTTP/1.1 200 OK\r\nContent-Type: audio/mpeg\r\nContent-Length: {EPISODE_BYTES}\r\n\r\n");
         // An MP3's frame sync first, so the file is taken for audio.
         let mut chunk = vec![0u8; 4096];
@@ -203,12 +210,12 @@ fn the_daemon_starts_answers_refuses_a_second_and_stops_on_sigterm_mid_download(
     assert!(!s.socket().exists(), "the socket is removed");
     assert!(s.log("first.log").contains("abandoning the job in progress"), "{}", s.log("first.log"));
 
-    // Never recorded, the abandoned file is still queued, not lost or marked failed. The next start
-    // takes it at its feed's next check, not at once (#410).
-    let queued = s.run(&["--local", "status"]);
-    assert!(text(&queued).contains("1 queued to download, 0 downloaded"), "{}", text(&queued));
+    // The next start takes the abandoned download up again at once, not at its feed's next check,
+    // which is an hour away (#410).
+    assert_eq!(episode_gets(), 1);
     let mut again = s.daemon("again.log");
-    until("the restarted daemon", Duration::from_secs(60), || s.log("again.log").contains("daemon started"));
+    until("the episode asked for again", Duration::from_secs(30), || episode_gets() == 2);
+    assert!(s.log("again.log").contains("taking up downloads interrupted by a restart"), "{}", s.log("again.log"));
     terminate(&mut again);
 
     // No daemon: the command does the work itself.