diff --git a/crates/ipx/CHANGELOG.md b/crates/ipx/CHANGELOG.md index 0553d1b..13a0b06 100644 --- a/crates/ipx/CHANGELOG.md +++ b/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). diff --git a/crates/ipx/src/daemon.rs b/crates/ipx/src/daemon.rs index 93dc60d..e924170 100644 --- a/crates/ipx/src/daemon.rs +++ b/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::(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. diff --git a/crates/ipx/src/db.rs b/crates/ipx/src/db.rs index 9d88376..2f5c384 100644 --- a/crates/ipx/src/db.rs +++ b/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 { - 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> { + 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::("", "id")).collect::, _>>()?; + 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"); diff --git a/crates/ipx/src/scan.rs b/crates/ipx/src/scan.rs index f58c00f..92b6fb4 100644 --- a/crates/ipx/src/scan.rs +++ b/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; diff --git a/crates/ipx/tests/daemon.rs b/crates/ipx/tests/daemon.rs index a095dca..1cc5f8e 100644 --- a/crates/ipx/tests/daemon.rs +++ b/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.