diff --git a/src/db.rs b/src/db.rs index 93b3beb..dee1bd1 100644 --- a/src/db.rs +++ b/src/db.rs @@ -559,39 +559,46 @@ impl Db { /// Returns true when this enclosure URL is new. False means we have downloaded it /// before, or deliberately reaped it -- either way it is not fetched again. - pub fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE enclosures SET state = 'done', path = ?2, bytes_done = ?3, - downloaded_at = ?4, last_error = NULL WHERE url = ?1", - rusqlite::params![url, path.to_string_lossy(), bytes as i64, now()], - )?; + pub async fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> { + self.exec( + "UPDATE enclosures SET state = 'done', path = $2, bytes_done = $3, + downloaded_at = $4, last_error = NULL WHERE url = $1", + vec![url.into(), path.to_string_lossy().into_owned().into(), (bytes as i64).into(), now().into()], + ) + .await?; Ok(()) } /// The row stays -- a failed URL is still a URL we have seen. `state` says why it has /// no file, and a retry is an explicit act rather than something a rescan does silently. - pub fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE enclosures SET state = ?2, last_error = ?3 WHERE url = ?1", - rusqlite::params![url, state, error], - )?; + pub async fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> { + self.exec( + "UPDATE enclosures SET state = $2, last_error = $3 WHERE url = $1", + vec![url.into(), state.into(), error.map(str::to_owned).into()], + ) + .await?; Ok(()) } - pub fn record_enclosure( + pub async fn record_enclosure( &self, feed_id: &str, guid: &str, enc: &crate::feed::Enclosure, ) -> Result { - let conn = self.conn.lock().unwrap(); - let inserted = conn.execute( - "INSERT OR IGNORE INTO enclosures (feed_id, guid, url, mime, length, state) - VALUES (?1, ?2, ?3, ?4, ?5, 'pending')", - rusqlite::params![feed_id, guid, enc.url, enc.mime, enc.length], - )?; + let inserted = self + .exec( + "INSERT INTO enclosures (feed_id, guid, url, mime, length, state) + VALUES ($1, $2, $3, $4, $5, 'pending') ON CONFLICT DO NOTHING", + vec![ + feed_id.into(), + guid.into(), + enc.url.clone().into(), + enc.mime.clone().into(), + enc.length.into(), + ], + ) + .await?; Ok(inserted == 1) } } @@ -608,23 +615,21 @@ pub struct Pending { impl Db { /// The download queue is the table, not the parse result: an enclosure held back by /// `max_new_per_check` is simply picked up by the next scan, in feed order. - pub fn pending(&self, feed_id: &str, limit: usize) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - // Newest first: a cap of 3 should mean the three latest episodes, not the - // three that happen to have been recorded first. + pub async fn pending(&self, feed_id: &str, limit: usize) -> Result> { + // Newest first: a cap of 3 should mean the three latest episodes, not the three that + // happen to have been recorded first. + self.rows( "SELECT x.id, x.url, x.mime FROM enclosures x JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid - WHERE x.feed_id = ?1 AND x.state = 'pending' + WHERE x.feed_id = $1 AND x.state = 'pending' ORDER BY coalesce(e.published, e.first_seen) DESC, x.id DESC - LIMIT ?2", - )?; - let rows = stmt - .query_map(rusqlite::params![feed_id, limit as i64], |r| { - Ok(Pending { id: r.get(0)?, url: r.get(1)?, mime: r.get(2)? }) - })? - .collect::>>()?; - Ok(rows) + LIMIT $2", + vec![feed_id.into(), (limit as i64).into()], + ) + .await? + .iter() + .map(|r| Ok(Pending { id: r.try_get("", "id")?, url: r.try_get("", "url")?, mime: r.try_get("", "mime")? })) + .collect() } } @@ -652,57 +657,53 @@ impl Db { /// /// (The Python intended `read = 1 AND flagged = 0` but never achieved it -- a missing /// plistlib import and an `EntreiesData` typo meant the filter always threw.) - pub fn reap_candidates(&self) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT e.id, e.url, e.path, e.bytes_done, coalesce(e.downloaded_at, 0), - CASE WHEN coalesce(readers.n, 0) >= coalesce(subs.n, 0) THEN 1 ELSE 0 END + pub async fn reap_candidates(&self) -> Result> { + // Yes/no as true and false, not 1 and 0: Postgres types a bare 1 as a 32-bit integer + // and will not hand it over as an i64. + self.rows( + "SELECT e.id, e.url, e.path, e.bytes_done, coalesce(e.downloaded_at, 0) AS age_key, + CASE WHEN coalesce(readers.n, 0) >= coalesce(subs.n, 0) THEN true ELSE false END AS read FROM enclosures e - LEFT JOIN (SELECT feed_id, count(*) n FROM subscriptions GROUP BY feed_id) subs + LEFT JOIN (SELECT feed_id, count(*) AS n FROM subscriptions GROUP BY feed_id) subs ON subs.feed_id = e.feed_id - LEFT JOIN (SELECT feed_id, guid, count(*) n FROM entry_state - WHERE read = 1 GROUP BY feed_id, guid) readers + LEFT JOIN (SELECT feed_id, guid, count(*) AS n FROM entry_state + WHERE read GROUP BY feed_id, guid) readers ON readers.feed_id = e.feed_id AND readers.guid = e.guid WHERE e.path IS NOT NULL AND NOT EXISTS (SELECT 1 FROM entry_state s - WHERE s.feed_id = e.feed_id AND s.guid = e.guid AND s.flagged = 1) + WHERE s.feed_id = e.feed_id AND s.guid = e.guid AND s.flagged) ORDER BY 6 DESC, coalesce(e.downloaded_at, 0) ASC, e.id ASC", - )?; - let rows = stmt - .query_map([], |r| { - Ok(Candidate { - id: r.get(0)?, - url: r.get(1)?, - path: r.get(2)?, - bytes: r.get(3)?, - age_key: r.get(4)?, - read: r.get::<_, i64>(5)? != 0, - }) - })? - .collect::>>()?; - Ok(rows) + vec![], + ) + .await? + .iter() + .map(|r| { + Ok(Candidate { + id: r.try_get("", "id")?, + url: r.try_get("", "url")?, + path: r.try_get("", "path")?, + bytes: r.try_get("", "bytes_done")?, + age_key: r.try_get("", "age_key")?, + read: r.try_get("", "read")?, + }) + }) + .collect() } /// The row survives the file: that is what stops a reaped episode being re-downloaded. - pub fn mark_reaped(&self, id: i64) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = ?1", - [id], - )?; + pub async fn mark_reaped(&self, id: i64) -> Result<()> { + self.exec("UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = $1", vec![id.into()]).await?; Ok(()) } /// Rows claiming a file that is no longer there (someone deleted it by hand). - pub fn missing_files(&self) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = - conn.prepare("SELECT id, path FROM enclosures WHERE path IS NOT NULL")?; - let rows = stmt - .query_map([], |r| Ok((r.get(0)?, r.get::<_, String>(1)?)))? - .collect::>>()?; - Ok(rows + pub async fn missing_files(&self) -> Result> { + Ok(enclosures::Entity::find() + .filter(enclosures::Column::Path.is_not_null()) + .all(&self.orm) + .await? .into_iter() + .filter_map(|e| Some((e.id, e.path?))) .filter(|(_, p)| !std::path::Path::new(p).exists()) .collect()) } @@ -1401,42 +1402,23 @@ impl Db { /// The next N enclosures with no file, newest entry first -- what "download latest" /// queues up. - pub fn undownloaded(&self, feed_id: &str, limit: i64) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( + pub async fn undownloaded(&self, feed_id: &str, limit: i64) -> Result> { + self.rows( "SELECT x.id FROM enclosures x JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid - WHERE x.feed_id = ?1 AND x.path IS NULL AND x.state != 'reaped' + WHERE x.feed_id = $1 AND x.path IS NULL AND x.state <> 'reaped' ORDER BY coalesce(e.published, e.first_seen) DESC - LIMIT ?2", - )?; - Ok(stmt - .query_map(rusqlite::params![feed_id, limit], |r| r.get(0))? - .collect::>>()?) + LIMIT $2", + vec![feed_id.into(), limit.into()], + ) + .await? + .iter() + .map(|r| Ok(r.try_get("", "id")?)) + .collect() } - pub fn enclosure(&self, id: i64) -> Result> { - let conn = self.conn.lock().unwrap(); - Ok(conn - .query_row( - "SELECT id, feed_id, guid, url, mime, length, path, state, last_error - FROM enclosures WHERE id = ?1", - [id], - |r| { - Ok(EncRow { - id: r.get(0)?, - feed_id: r.get(1)?, - guid: r.get(2)?, - url: r.get(3)?, - mime: r.get(4)?, - length: r.get(5)?, - path: r.get(6)?, - state: r.get(7)?, - last_error: r.get(8)?, - }) - }, - ) - .optional()?) + pub async fn enclosure(&self, id: i64) -> Result> { + Ok(enclosures::Entity::find_by_id(id).one(&self.orm).await?.map(EncRow::from)) } /// Read and kept, per person. The row is created on first touch. @@ -1466,13 +1448,12 @@ impl Db { /// How many files this feed has on disk. Decides whether a feed dropped from an OPML /// can be removed or must be kept. - pub fn downloaded_count(&self, feed_id: &str) -> Result { - let conn = self.conn.lock().unwrap(); - Ok(conn.query_row( - "SELECT count(*) FROM enclosures WHERE feed_id = ?1 AND path IS NOT NULL", - [feed_id], - |r| r.get(0), - )?) + pub async fn downloaded_count(&self, feed_id: &str) -> Result { + Ok(enclosures::Entity::find() + .filter(enclosures::Column::FeedId.eq(feed_id)) + .filter(enclosures::Column::Path.is_not_null()) + .count(&self.orm) + .await? as i64) } /// Names a feed without touching its conditional-GET validators. An OPML subscription @@ -1675,23 +1656,20 @@ 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 fn requeue_interrupted(&self) -> Result { - let conn = self.conn.lock().unwrap(); - Ok(conn.execute( - "UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL", - [], - )?) + 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) } /// 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<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE enclosures SET state = 'pending', last_error = NULL - WHERE id = ?1 AND path IS NULL", - [id], - )?; + pub async fn requeue(&self, id: i64) -> Result<()> { + self.exec( + "UPDATE enclosures SET state = 'pending', last_error = NULL WHERE id = $1 AND path IS NULL", + vec![id.into()], + ) + .await?; Ok(()) } } @@ -2085,7 +2063,7 @@ mod tests { (3,'f','new','u-new','pending');", ) .unwrap(); - let got: Vec = db.pending("f", 2).unwrap().into_iter().map(|p| p.url).collect(); + let got: Vec = db.pending("f", 2).await.unwrap().into_iter().map(|p| p.url).collect(); assert_eq!(got, vec!["u-new", "u-mid"], "newest first, oldest left for later"); } @@ -2100,7 +2078,7 @@ mod tests { (4,'f','d','u4','done','/tmp/x');", ) .unwrap(); - assert_eq!(db.requeue_interrupted().unwrap(), 1, "only the in-flight, fileless one"); + assert_eq!(db.requeue_interrupted().await.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() diff --git a/src/main.rs b/src/main.rs index 56321fb..6a227f0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -225,7 +225,7 @@ async fn main() -> Result<()> { Command::Add { url, folder, keywords } => { add(&ctx, &config_path, &url, folder, keywords).await } - Command::Rm { feed } => rm(&ctx, &config_path, &feed), + Command::Rm { feed } => rm(&ctx, &config_path, &feed).await, Command::User { cmd } => user_cmd(&ctx, cmd).await, Command::Import { file } => import(&ctx, &config_path, &file).await, Command::Export { file } => export(&ctx, &file), @@ -334,10 +334,10 @@ async fn run(ctx: &Arc, cmd: Cmd) -> Result<()> { match cmd { Cmd::Fetch { feed, force } => { // Make room before pulling more down, as the original did per download. - reap(ctx, false, false)?; + reap(ctx, false, false).await?; fetch(ctx, feed.as_deref(), force).await } - Cmd::Reap { dry_run } => reap(ctx, dry_run, true), + Cmd::Reap { dry_run } => reap(ctx, dry_run, true).await, Cmd::Download { enclosure } => download_one(ctx, enclosure).await, Cmd::Status => { ctx.out.emit(status(ctx).await); @@ -387,13 +387,13 @@ async fn daemon( } } - match ctx.db.requeue_interrupted() { + match ctx.db.requeue_interrupted().await { 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"), } - match retire_stranded(&ctx) { + match retire_stranded(&ctx).await { Ok(0) => {} Ok(n) => tracing::info!(feeds = n, "retired feeds whose OPML is no longer in config"), Err(e) => tracing::warn!(error = ?e, "could not retire feeds whose OPML is no longer in config"), @@ -651,7 +651,7 @@ fn url_stem(url: &str) -> String { .unwrap_or_else(|| url.to_owned()) } -fn rm(ctx: &Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> { +async fn rm(ctx: &Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> { let mut cfg = (*ctx.cfg()).clone(); if cfg.feeds.remove(feed).is_none() { // Derived from an OPML: drop it here, though the subscription will list it again @@ -663,7 +663,7 @@ fn rm(ctx: &Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> { cfg.save(config_path)?; // State and files stay: re-adding the feed should not re-download its back catalogue. println!("removed {feed}; downloads and history kept"); - retire_group(ctx, feed)?; + retire_group(ctx, feed).await?; Ok(()) } @@ -813,8 +813,8 @@ fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> { /// `standalone` false means this is the sweep that runs before a scan: it reports what it /// deleted, but must not emit the terminal ReapDone, or a client waiting on its `fetch` /// would stop reading before the scan had even started. -fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { - let r = retention::run(&ctx.cfg(), &ctx.db, dry_run)?; +async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { + let r = retention::run(&ctx.cfg(), &ctx.db, dry_run).await?; for c in r.aged_out.iter().chain(r.over_quota.iter()) { ctx.out.emit(Event::Reaped { path: c.path.clone(), @@ -992,14 +992,14 @@ pub fn subscriptions(ctx: &Ctx) -> Result> { /// itself is removed, since `subscriptions()` would otherwise keep scanning them under a /// fallback policy meant for a feed with no parent at all. A feed promoted to config is not /// derived any more, so it is only unmanaged. -pub fn retire_group(ctx: &Ctx, parent_id: &str) -> Result<()> { +pub async fn retire_group(ctx: &Ctx, parent_id: &str) -> Result<()> { let cfg = ctx.cfg(); for m in ctx.db.managed_feeds()?.into_iter().filter(|m| m.group_id == parent_id) { if cfg.feeds.contains_key(&m.id) { // Scanned from its config entry and still read. Dropped as derived, its stored // entries would go with it: davewiner's 11 were promoted without being unmanaged. ctx.db.unmanage(&m.id)?; - } else if ctx.db.downloaded_count(&m.id).unwrap_or(1) > 0 { + } else if ctx.db.downloaded_count(&m.id).await.unwrap_or(1) > 0 { ctx.db.set_orphaned(&m.id, true)?; } else { ctx.db.drop_managed(&m.id)?; @@ -1012,7 +1012,7 @@ pub fn retire_group(ctx: &Ctx, parent_id: &str) -> Result<()> { /// dropped or unmanaged. An OPML removed before `retire_group` existed left its feeds behind: /// davewiner's 922 were skipped by every scan and never cleared, and their stale errors were /// most of the ones stored. -fn retire_stranded(ctx: &Ctx) -> Result { +async fn retire_stranded(ctx: &Ctx) -> Result { let cfg = ctx.cfg(); let before = ctx.db.managed_feeds()?; let stranded: std::collections::BTreeSet<&str> = before @@ -1021,7 +1021,7 @@ fn retire_stranded(ctx: &Ctx) -> Result { .filter(|g| !cfg.feeds.contains_key(*g)) .collect(); for group in stranded { - retire_group(ctx, group)?; + retire_group(ctx, group).await?; } Ok(before.len() - ctx.db.managed_feeds()?.len()) } @@ -1159,7 +1159,7 @@ async fn scan_one( scan.new_entries += 1; } for enc in &entry.enclosures { - let was = if ctx.db.record_enclosure(id, &entry.guid, enc)? { + let was = if ctx.db.record_enclosure(id, &entry.guid, enc).await? { None } else if let Some(reason) = skipped.get(&enc.url) { Some(reason.as_str()) @@ -1169,8 +1169,8 @@ async fn scan_one( let now = reject(&ctx.cfg(), feed_cfg, &policy, entry, enc); if now != was { match now { - Some(reason) => ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason))?, - None => ctx.db.mark_enclosure(&enc.url, "pending", None)?, + Some(reason) => ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason)).await?, + None => ctx.db.mark_enclosure(&enc.url, "pending", None).await?, } } } @@ -1182,10 +1182,10 @@ async fn scan_one( let folder = download::folder_for(&cfg, id, feed_cfg, parsed.title.as_deref()); let dest_dir = cfg.general.download_dir.join(&folder); - for item in ctx.db.pending(id, budget)? { + for item in ctx.db.pending(id, budget).await? { if download::looks_like_torrent(&item.url, item.mime.as_deref()) { if !ctx.cfg().torrent.enabled { - ctx.db.mark_enclosure(&item.url, "skipped", Some("torrents disabled"))?; + ctx.db.mark_enclosure(&item.url, "skipped", Some("torrents disabled")).await?; ctx.out.emit(Event::TorrentDeferred { feed: id.to_string(), url: item.url.clone(), @@ -1195,14 +1195,14 @@ async fn scan_one( } if ctx.detach_torrents { // 'downloading' keeps the next scan from queueing it a second time. - ctx.db.mark_enclosure(&item.url, "downloading", None)?; + ctx.db.mark_enclosure(&item.url, "downloading", None).await?; spawn_torrent(ctx, id.to_string(), item.id, item.url.clone(), dest_dir.clone()); scan.torrents += 1; continue; } match torrent_one(ctx, id, item.id, &item.url, &dest_dir).await { Ok((path, bytes)) => { - ctx.db.mark_downloaded(&item.url, &path, bytes)?; + ctx.db.mark_downloaded(&item.url, &path, bytes).await?; ctx.out.emit(Event::DownloadDone { feed: id.to_string(), enclosure: item.id, @@ -1220,7 +1220,7 @@ async fn scan_one( url: item.url.clone(), msg: msg.clone(), }); - ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?; + ctx.db.mark_enclosure(&item.url, "error", Some(&msg)).await?; scan.failed += 1; } } @@ -1245,7 +1245,7 @@ async fn scan_one( url: item.url.clone(), msg: msg.clone(), }); - ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?; + ctx.db.mark_enclosure(&item.url, "error", Some(&msg)).await?; scan.failed += 1; } } @@ -1334,7 +1334,7 @@ async fn sync_group( if listed.iter().any(|(_, u)| u == &m.url) { continue; } - if ctx.db.downloaded_count(&m.id).unwrap_or(1) > 0 { + if ctx.db.downloaded_count(&m.id).await.unwrap_or(1) > 0 { // Never orphan a downloaded file: keep the feed and say why in the UI. ctx.db.set_orphaned(&m.id, true)?; kept += 1; @@ -1488,14 +1488,14 @@ async fn fetch_one( // as if it were an episode. let _ = tokio::fs::remove_file(&got.tmp).await; if !ctx.cfg().torrent.enabled { - ctx.db.mark_enclosure(url, "skipped", Some("torrents disabled"))?; + ctx.db.mark_enclosure(url, "skipped", Some("torrents disabled")).await?; anyhow::bail!("body is a torrent and torrents are disabled"); } return torrent_one(ctx, feed_id, enclosure, url, dest_dir).await; } let path = download::place(&got, dest_dir).await?; - ctx.db.mark_downloaded(url, &path, got.bytes)?; + ctx.db.mark_downloaded(url, &path, got.bytes).await?; Ok((path, got.bytes)) } @@ -1513,7 +1513,7 @@ fn spawn_torrent(ctx: &Arc, feed_id: String, enclosure: i64, url: String, d let db = &ctx.db; match outcome { Ok((path, bytes)) => { - if let Err(e) = db.mark_downloaded(&url, &path, bytes) { + if let Err(e) = db.mark_downloaded(&url, &path, bytes).await { tracing::warn!(error = ?e, "could not record the finished torrent"); } ctx.out.emit(Event::DownloadDone { @@ -1526,7 +1526,7 @@ fn spawn_torrent(ctx: &Arc, feed_id: String, enclosure: i64, url: String, d } Err(e) => { let msg = format!("{e:#}"); - let _ = db.mark_enclosure(&url, "error", Some(&msg)); + let _ = db.mark_enclosure(&url, "error", Some(&msg)).await; ctx.out.emit(Event::DownloadError { feed: feed_id, enclosure, url, msg }); } } @@ -1539,7 +1539,7 @@ async fn download_one(ctx: &Arc, id: i64) -> Result<()> { let cfg = ctx.cfg(); let enc = ctx .db - .enclosure(id)? + .enclosure(id).await? .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?; if enc.path.is_some() { return Ok(()); // Already here. @@ -1561,7 +1561,7 @@ async fn download_one(ctx: &Arc, id: i64) -> Result<()> { ctx.out.emit(Event::FeedStart { feed: enc.feed_id.clone() }); let is_torrent = download::looks_like_torrent(&enc.url, enc.mime.as_deref()); if is_torrent && cfg.torrent.enabled && ctx.detach_torrents { - ctx.db.mark_enclosure(&enc.url, "downloading", None)?; + ctx.db.mark_enclosure(&enc.url, "downloading", None).await?; spawn_torrent(ctx, enc.feed_id.clone(), enc.id, enc.url.clone(), dest_dir); return Ok(()); } @@ -1577,7 +1577,7 @@ async fn download_one(ctx: &Arc, id: i64) -> Result<()> { match result { Ok((path, bytes)) => { - ctx.db.mark_downloaded(&enc.url, &path, bytes)?; + ctx.db.mark_downloaded(&enc.url, &path, bytes).await?; ctx.out.emit(Event::DownloadDone { feed: enc.feed_id.clone(), enclosure: enc.id, @@ -1588,7 +1588,7 @@ async fn download_one(ctx: &Arc, id: i64) -> Result<()> { } Err(e) => { let msg = format!("{e:#}"); - ctx.db.mark_enclosure(&enc.url, "error", Some(&msg))?; + ctx.db.mark_enclosure(&enc.url, "error", Some(&msg)).await?; ctx.out.emit(Event::DownloadError { feed: enc.feed_id.clone(), enclosure: enc.id, @@ -1756,10 +1756,10 @@ mod tests { ctx.db.upsert_managed("empty", "http://x/empty.xml", "Empty", "parent").unwrap(); ctx.db.upsert_managed("has-file", "http://x/has-file.xml", "Has File", "parent").unwrap(); let enc = feed::Enclosure { url: "http://x/ep.mp3".into(), mime: None, length: None }; - ctx.db.record_enclosure("has-file", "g1", &enc).unwrap(); - ctx.db.mark_downloaded(&enc.url, std::path::Path::new("/downloads/ep.mp3"), 1).unwrap(); + ctx.db.record_enclosure("has-file", "g1", &enc).await.unwrap(); + ctx.db.mark_downloaded(&enc.url, std::path::Path::new("/downloads/ep.mp3"), 1).await.unwrap(); - retire_group(&ctx, "parent").unwrap(); + retire_group(&ctx, "parent").await.unwrap(); let managed = ctx.db.managed_feeds().unwrap(); assert!(!managed.iter().any(|m| m.id == "empty"), "nothing downloaded, so it is forgotten"); @@ -1780,7 +1780,7 @@ mod tests { ctx.db.upsert_managed("listed", "http://x/l.xml", "Listed", "live-opml").unwrap(); ctx.db.record_entry("promoted", &feed::Entry { guid: "g1".into(), ..Default::default() }).unwrap(); - assert_eq!(retire_stranded(&ctx).unwrap(), 2, "empty dropped, promoted unmanaged"); + assert_eq!(retire_stranded(&ctx).await.unwrap(), 2, "empty dropped, promoted unmanaged"); let managed: Vec = ctx.db.managed_feeds().unwrap().into_iter().map(|m| m.id).collect(); assert_eq!(managed, ["listed"], "a group still in config is left alone"); diff --git a/src/retention.rs b/src/retention.rs index 66ecf5b..a6e81e4 100644 --- a/src/retention.rs +++ b/src/retention.rs @@ -45,25 +45,25 @@ pub fn aged(candidates: &[Candidate], cutoff: i64) -> Vec { .collect() } -pub fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result { +pub async fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result { let mut report = Report::default(); // Someone may have deleted a file by hand; the row must stop claiming it exists. - for (id, path) in db.missing_files()? { + for (id, path) in db.missing_files().await? { if !dry_run { - db.mark_reaped(id)?; + db.mark_reaped(id).await?; } tracing::debug!(path, "file gone, row reaped"); report.reconciled += 1; } - let candidates = db.reap_candidates()?; + let candidates = db.reap_candidates().await?; if cfg.general.max_age_days > 0 { let cutoff = now() - (cfg.general.max_age_days * 86_400) as i64; report.aged_out = aged(&candidates, cutoff); for c in &report.aged_out { - report.bytes_freed += remove(db, c, dry_run)?; + report.bytes_freed += remove(db, c, dry_run).await?; } if !dry_run { report.entries_pruned = db.prune_entries(cutoff)?; @@ -81,14 +81,14 @@ pub fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result { let total: u64 = remaining.iter().map(|c| c.bytes.max(0) as u64).sum(); report.over_quota = pick(&remaining, total, limit); for c in &report.over_quota { - report.bytes_freed += remove(db, c, dry_run)?; + report.bytes_freed += remove(db, c, dry_run).await?; } } Ok(report) } -fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result { +async fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result { if dry_run { return Ok(c.bytes.max(0) as u64); } @@ -100,7 +100,7 @@ fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result { tracing::warn!(path = c.path, error = %e, "could not delete"); return Ok(0); } - db.mark_reaped(c.id)?; + db.mark_reaped(c.id).await?; Ok(size) } @@ -176,7 +176,7 @@ mod tests { ) .unwrap(); - let got: Vec = db.reap_candidates().unwrap().iter().map(|c| c.id).collect(); + let got: Vec = db.reap_candidates().await.unwrap().iter().map(|c| c.id).collect(); assert_eq!( got, vec![4, 2, 3], diff --git a/src/web.rs b/src/web.rs index 67a36e8..1c897d5 100644 --- a/src/web.rs +++ b/src/web.rs @@ -1422,7 +1422,7 @@ async fn remove_feed( } cfg.save(&state.config_path)?; state.ctx.reload_cfg(&state.config_path)?; - crate::retire_group(&state.ctx, &id)?; + crate::retire_group(&state.ctx, &id).await?; Ok(StatusCode::NO_CONTENT) } @@ -1458,12 +1458,12 @@ async fn download_now( let enc = state .ctx .db - .enclosure(id)? + .enclosure(id).await? .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?; if enc.path.is_some() { return Ok(StatusCode::NO_CONTENT); // Already here. } - state.ctx.db.requeue(id)?; + state.ctx.db.requeue(id).await?; state .cmds .send(Command::Download { enclosure: id }) @@ -1487,7 +1487,7 @@ async fn delete_file( let enc = state .ctx .db - .enclosure(id)? + .enclosure(id).await? .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?; // There is one copy of the file: deleting it deletes everyone's. Say so before doing @@ -1519,7 +1519,7 @@ async fn delete_file( return Err(e.into()); } // The row survives as 'reaped', which is what stops the next scan re-downloading it. - state.ctx.db.mark_reaped(id)?; + state.ctx.db.mark_reaped(id).await?; state.events.send(Event::Reaped { path: enc.path.unwrap_or_default(), bytes: enc.length.unwrap_or(0).max(0) as u64, @@ -1573,7 +1573,7 @@ async fn media( Path(id): Path, req: Request, ) -> Response { - let Ok(Some(enc)) = state.ctx.db.enclosure(id) else { + let Ok(Some(enc)) = state.ctx.db.enclosure(id).await else { return (StatusCode::NOT_FOUND, "no such enclosure").into_response(); }; let Some(path) = enc.path else { @@ -1648,9 +1648,9 @@ async fn download_latest( Path(id): Path, Json(body): Json, ) -> Result, ApiError> { - let ids = state.ctx.db.undownloaded(&id, body.count.clamp(1, 100))?; + let ids = state.ctx.db.undownloaded(&id, body.count.clamp(1, 100)).await?; for enc in &ids { - state.ctx.db.requeue(*enc)?; + state.ctx.db.requeue(*enc).await?; state .cmds .send(Command::Download { enclosure: *enc })