rays/ipodderx-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SeaORM: enclosures, downloads and the reaper

Browse Source 
Twelve enclosure functions move to SeaORM: recording, the download queue,
marking done or failed, requeueing, and what the reaper may delete. INSERT OR
IGNORE becomes ON CONFLICT DO NOTHING; the reaper's read verdict is true or
false rather than 1 or 0, which Postgres would type as a 32-bit integer and
refuse to read as an i64; `read = 1` and `flagged = 1` test the booleans
themselves. retention::run and its callers (reap, rm, retire_group,
retire_stranded) become async.

The reaper deletes files, so it was checked on a copy of production against the
old SQL on the same file: all 2,195 candidates, identical and in the same order.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Ray Slakinski
2026-09-18 18:38:07 +00:00
parent 6bf1ad6b31
commit a68bfb179b
4 changed files with 154 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
src/db.rs
View File

@@ -559,39 +559,46 @@ impl Db {
/// Returns true when this enclosure URL is new. False means we have downloaded it /// 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. /// 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<()> { pub async fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> {
let conn = self.conn.lock().unwrap(); self.exec(
conn.execute( "UPDATE enclosures SET state = 'done', path = $2, bytes_done = $3,
"UPDATE enclosures SET state = 'done', path = ?2, bytes_done = ?3, downloaded_at = $4, last_error = NULL WHERE url = $1",
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()],
rusqlite::params![url, path.to_string_lossy(), bytes as i64, now()], )
)?; .await?;
Ok(()) Ok(())
} }
/// The row stays -- a failed URL is still a URL we have seen. `state` says why it has /// 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. /// 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<()> { pub async fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> {
let conn = self.conn.lock().unwrap(); self.exec(
conn.execute( "UPDATE enclosures SET state = $2, last_error = $3 WHERE url = $1",
"UPDATE enclosures SET state = ?2, last_error = ?3 WHERE url = ?1", vec![url.into(), state.into(), error.map(str::to_owned).into()],
rusqlite::params![url, state, error], )
)?; .await?;
Ok(()) Ok(())
} }
pub fn record_enclosure( pub async fn record_enclosure(
&self, &self,
feed_id: &str, feed_id: &str,
guid: &str, guid: &str,
enc: &crate::feed::Enclosure, enc: &crate::feed::Enclosure,
) -> Result<bool> { ) -> Result<bool> {
let conn = self.conn.lock().unwrap(); let inserted = self
let inserted = conn.execute( .exec(
"INSERT OR IGNORE INTO enclosures (feed_id, guid, url, mime, length, state) "INSERT INTO enclosures (feed_id, guid, url, mime, length, state)
VALUES (?1, ?2, ?3, ?4, ?5, 'pending')", VALUES ($1, $2, $3, $4, $5, 'pending') ON CONFLICT DO NOTHING",
rusqlite::params![feed_id, guid, enc.url, enc.mime, enc.length], vec![
)?; feed_id.into(),
guid.into(),
enc.url.clone().into(),
enc.mime.clone().into(),
enc.length.into(),
],
)
.await?;
Ok(inserted == 1) Ok(inserted == 1)
} }
} }
@@ -608,23 +615,21 @@ pub struct Pending {
impl Db { impl Db {
/// The download queue is the table, not the parse result: an enclosure held back by /// 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. /// `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<Vec<Pending>> { pub async fn pending(&self, feed_id: &str, limit: usize) -> Result<Vec<Pending>> {
let conn = self.conn.lock().unwrap(); // Newest first: a cap of 3 should mean the three latest episodes, not the three that
let mut stmt = conn.prepare( // happen to have been recorded first.
// Newest first: a cap of 3 should mean the three latest episodes, not the self.rows(
// three that happen to have been recorded first.
"SELECT x.id, x.url, x.mime FROM enclosures x "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 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 ORDER BY coalesce(e.published, e.first_seen) DESC, x.id DESC
LIMIT ?2", LIMIT $2",
)?; vec![feed_id.into(), (limit as i64).into()],
let rows = stmt )
.query_map(rusqlite::params![feed_id, limit as i64], |r| { .await?
Ok(Pending { id: r.get(0)?, url: r.get(1)?, mime: r.get(2)? }) .iter()
})? .map(|r| Ok(Pending { id: r.try_get("", "id")?, url: r.try_get("", "url")?, mime: r.try_get("", "mime")? }))
.collect::<rusqlite::Result<Vec<_>>>()?; .collect()
Ok(rows)
} }
} }
@@ -652,57 +657,53 @@ impl Db {
/// ///
/// (The Python intended `read = 1 AND flagged = 0` but never achieved it -- a missing /// (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.) /// plistlib import and an `EntreiesData` typo meant the filter always threw.)
pub fn reap_candidates(&self) -> Result<Vec<Candidate>> { pub async fn reap_candidates(&self) -> Result<Vec<Candidate>> {
let conn = self.conn.lock().unwrap(); // Yes/no as true and false, not 1 and 0: Postgres types a bare 1 as a 32-bit integer
let mut stmt = conn.prepare( // and will not hand it over as an i64.
"SELECT e.id, e.url, e.path, e.bytes_done, coalesce(e.downloaded_at, 0), self.rows(
CASE WHEN coalesce(readers.n, 0) >= coalesce(subs.n, 0) THEN 1 ELSE 0 END "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 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 ON subs.feed_id = e.feed_id
LEFT JOIN (SELECT feed_id, guid, count(*) n FROM entry_state LEFT JOIN (SELECT feed_id, guid, count(*) AS n FROM entry_state
WHERE read = 1 GROUP BY feed_id, guid) readers WHERE read GROUP BY feed_id, guid) readers
ON readers.feed_id = e.feed_id AND readers.guid = e.guid ON readers.feed_id = e.feed_id AND readers.guid = e.guid
WHERE e.path IS NOT NULL WHERE e.path IS NOT NULL
AND NOT EXISTS (SELECT 1 FROM entry_state s 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", ORDER BY 6 DESC, coalesce(e.downloaded_at, 0) ASC, e.id ASC",
)?; vec![],
let rows = stmt )
.query_map([], |r| { .await?
Ok(Candidate { .iter()
id: r.get(0)?, .map(|r| {
url: r.get(1)?, Ok(Candidate {
path: r.get(2)?, id: r.try_get("", "id")?,
bytes: r.get(3)?, url: r.try_get("", "url")?,
age_key: r.get(4)?, path: r.try_get("", "path")?,
read: r.get::<_, i64>(5)? != 0, bytes: r.try_get("", "bytes_done")?,
}) age_key: r.try_get("", "age_key")?,
})? read: r.try_get("", "read")?,
.collect::<rusqlite::Result<Vec<_>>>()?; })
Ok(rows) })
.collect()
} }
/// The row survives the file: that is what stops a reaped episode being re-downloaded. /// The row survives the file: that is what stops a reaped episode being re-downloaded.
pub fn mark_reaped(&self, id: i64) -> Result<()> { pub async fn mark_reaped(&self, id: i64) -> Result<()> {
let conn = self.conn.lock().unwrap(); self.exec("UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = $1", vec![id.into()]).await?;
conn.execute(
"UPDATE enclosures SET state = 'reaped', path = NULL WHERE id = ?1",
[id],
)?;
Ok(()) Ok(())
} }
/// Rows claiming a file that is no longer there (someone deleted it by hand). /// Rows claiming a file that is no longer there (someone deleted it by hand).
pub fn missing_files(&self) -> Result<Vec<(i64, String)>> { pub async fn missing_files(&self) -> Result<Vec<(i64, String)>> {
let conn = self.conn.lock().unwrap(); Ok(enclosures::Entity::find()
let mut stmt = .filter(enclosures::Column::Path.is_not_null())
conn.prepare("SELECT id, path FROM enclosures WHERE path IS NOT NULL")?; .all(&self.orm)
let rows = stmt .await?
.query_map([], |r| Ok((r.get(0)?, r.get::<_, String>(1)?)))?
.collect::<rusqlite::Result<Vec<_>>>()?;
Ok(rows
.into_iter() .into_iter()
.filter_map(|e| Some((e.id, e.path?)))
.filter(|(_, p)| !std::path::Path::new(p).exists()) .filter(|(_, p)| !std::path::Path::new(p).exists())
.collect()) .collect())
} }
@@ -1401,42 +1402,23 @@ impl Db {
/// The next N enclosures with no file, newest entry first -- what "download latest" /// The next N enclosures with no file, newest entry first -- what "download latest"
/// queues up. /// queues up.
pub fn undownloaded(&self, feed_id: &str, limit: i64) -> Result<Vec<i64>> { pub async fn undownloaded(&self, feed_id: &str, limit: i64) -> Result<Vec<i64>> {
let conn = self.conn.lock().unwrap(); self.rows(
let mut stmt = conn.prepare(
"SELECT x.id FROM enclosures x "SELECT x.id FROM enclosures x
JOIN entries e ON e.feed_id = x.feed_id AND e.guid = x.guid 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 ORDER BY coalesce(e.published, e.first_seen) DESC
LIMIT ?2", LIMIT $2",
)?; vec![feed_id.into(), limit.into()],
Ok(stmt )
.query_map(rusqlite::params![feed_id, limit], |r| r.get(0))? .await?
.collect::<rusqlite::Result<Vec<_>>>()?) .iter()
.map(|r| Ok(r.try_get("", "id")?))
.collect()
} }
pub fn enclosure(&self, id: i64) -> Result<Option<EncRow>> { pub async fn enclosure(&self, id: i64) -> Result<Option<EncRow>> {
let conn = self.conn.lock().unwrap(); Ok(enclosures::Entity::find_by_id(id).one(&self.orm).await?.map(EncRow::from))
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()?)
} }
/// Read and kept, per person. The row is created on first touch. /// 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 /// How many files this feed has on disk. Decides whether a feed dropped from an OPML
/// can be removed or must be kept. /// can be removed or must be kept.
pub fn downloaded_count(&self, feed_id: &str) -> Result<i64> { pub async fn downloaded_count(&self, feed_id: &str) -> Result<i64> {
let conn = self.conn.lock().unwrap(); Ok(enclosures::Entity::find()
Ok(conn.query_row( .filter(enclosures::Column::FeedId.eq(feed_id))
"SELECT count(*) FROM enclosures WHERE feed_id = ?1 AND path IS NOT NULL", .filter(enclosures::Column::Path.is_not_null())
[feed_id], .count(&self.orm)
|r| r.get(0), .await? as i64)
)?)
} }
/// Names a feed without touching its conditional-GET validators. An OPML subscription /// 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 /// 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 /// `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. /// there forever: the pending queue skips it and nothing else ever revisits it.
pub fn requeue_interrupted(&self) -> Result<usize> { pub async fn requeue_interrupted(&self) -> Result<usize> {
let conn = self.conn.lock().unwrap(); Ok(self
Ok(conn.execute( .exec("UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL", vec![])
"UPDATE enclosures SET state = 'pending' WHERE state = 'downloading' AND path IS NULL", .await? as usize)
[],
)?)
} }
/// Puts an enclosure back in the queue so the next scan picks it up. This is how a /// 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. /// `skipped` verdict (from a filter that has since been changed) gets revisited.
pub fn requeue(&self, id: i64) -> Result<()> { pub async fn requeue(&self, id: i64) -> Result<()> {
let conn = self.conn.lock().unwrap(); self.exec(
conn.execute( "UPDATE enclosures SET state = 'pending', last_error = NULL WHERE id = $1 AND path IS NULL",
"UPDATE enclosures SET state = 'pending', last_error = NULL vec![id.into()],
WHERE id = ?1 AND path IS NULL", )
[id], .await?;
)?;
Ok(()) Ok(())
} }
} }
@@ -2085,7 +2063,7 @@ mod tests {
(3,'f','new','u-new','pending');", (3,'f','new','u-new','pending');",
) )
.unwrap(); .unwrap();
let got: Vec<String> = db.pending("f", 2).unwrap().into_iter().map(|p| p.url).collect(); let got: Vec<String> = 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"); 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');", (4,'f','d','u4','done','/tmp/x');",
) )
.unwrap(); .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 conn = db.conn.lock().unwrap();
let state = |id: i64| -> String { let state = |id: i64| -> String {
conn.query_row("SELECT state FROM enclosures WHERE id = ?1", [id], |r| r.get(0)).unwrap() conn.query_row("SELECT state FROM enclosures WHERE id = ?1", [id], |r| r.get(0)).unwrap()

70
src/main.rs
View File

@@ -225,7 +225,7 @@ async fn main() -> Result<()> {
Command::Add { url, folder, keywords } => { Command::Add { url, folder, keywords } => {
add(&ctx, &config_path, &url, folder, keywords).await 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::User { cmd } => user_cmd(&ctx, cmd).await,
Command::Import { file } => import(&ctx, &config_path, &file).await, Command::Import { file } => import(&ctx, &config_path, &file).await,
Command::Export { file } => export(&ctx, &file), Command::Export { file } => export(&ctx, &file),
@@ -334,10 +334,10 @@ async fn run(ctx: &Arc<Ctx>, cmd: Cmd) -> Result<()> {
match cmd { match cmd {
Cmd::Fetch { feed, force } => { Cmd::Fetch { feed, force } => {
// Make room before pulling more down, as the original did per download. // 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 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::Download { enclosure } => download_one(ctx, enclosure).await,
Cmd::Status => { Cmd::Status => {
ctx.out.emit(status(ctx).await); 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(n) if n > 0 => tracing::info!(count = n, "requeued downloads interrupted by a restart"),
Ok(_) => {} Ok(_) => {}
Err(e) => tracing::warn!(error = ?e, "could not requeue interrupted downloads"), Err(e) => tracing::warn!(error = ?e, "could not requeue interrupted downloads"),
} }
match retire_stranded(&ctx) { match retire_stranded(&ctx).await {
Ok(0) => {} Ok(0) => {}
Ok(n) => tracing::info!(feeds = n, "retired feeds whose OPML is no longer in config"), 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"), 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()) .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(); let mut cfg = (*ctx.cfg()).clone();
if cfg.feeds.remove(feed).is_none() { if cfg.feeds.remove(feed).is_none() {
// Derived from an OPML: drop it here, though the subscription will list it again // 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)?; cfg.save(config_path)?;
// State and files stay: re-adding the feed should not re-download its back catalogue. // State and files stay: re-adding the feed should not re-download its back catalogue.
println!("removed {feed}; downloads and history kept"); println!("removed {feed}; downloads and history kept");
retire_group(ctx, feed)?; retire_group(ctx, feed).await?;
Ok(()) 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 /// `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` /// deleted, but must not emit the terminal ReapDone, or a client waiting on its `fetch`
/// would stop reading before the scan had even started. /// would stop reading before the scan had even started.
fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { async fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> {
let r = retention::run(&ctx.cfg(), &ctx.db, dry_run)?; let r = retention::run(&ctx.cfg(), &ctx.db, dry_run).await?;
for c in r.aged_out.iter().chain(r.over_quota.iter()) { for c in r.aged_out.iter().chain(r.over_quota.iter()) {
ctx.out.emit(Event::Reaped { ctx.out.emit(Event::Reaped {
path: c.path.clone(), path: c.path.clone(),
@@ -992,14 +992,14 @@ pub fn subscriptions(ctx: &Ctx) -> Result<Vec<Sub>> {
/// itself is removed, since `subscriptions()` would otherwise keep scanning them under a /// 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 /// 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. /// 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(); let cfg = ctx.cfg();
for m in ctx.db.managed_feeds()?.into_iter().filter(|m| m.group_id == parent_id) { for m in ctx.db.managed_feeds()?.into_iter().filter(|m| m.group_id == parent_id) {
if cfg.feeds.contains_key(&m.id) { if cfg.feeds.contains_key(&m.id) {
// Scanned from its config entry and still read. Dropped as derived, its stored // 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. // entries would go with it: davewiner's 11 were promoted without being unmanaged.
ctx.db.unmanage(&m.id)?; 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)?; ctx.db.set_orphaned(&m.id, true)?;
} else { } else {
ctx.db.drop_managed(&m.id)?; 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: /// 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 /// davewiner's 922 were skipped by every scan and never cleared, and their stale errors were
/// most of the ones stored. /// most of the ones stored.
fn retire_stranded(ctx: &Ctx) -> Result<usize> { async fn retire_stranded(ctx: &Ctx) -> Result<usize> {
let cfg = ctx.cfg(); let cfg = ctx.cfg();
let before = ctx.db.managed_feeds()?; let before = ctx.db.managed_feeds()?;
let stranded: std::collections::BTreeSet<&str> = before let stranded: std::collections::BTreeSet<&str> = before
@@ -1021,7 +1021,7 @@ fn retire_stranded(ctx: &Ctx) -> Result<usize> {
.filter(|g| !cfg.feeds.contains_key(*g)) .filter(|g| !cfg.feeds.contains_key(*g))
.collect(); .collect();
for group in stranded { for group in stranded {
retire_group(ctx, group)?; retire_group(ctx, group).await?;
} }
Ok(before.len() - ctx.db.managed_feeds()?.len()) Ok(before.len() - ctx.db.managed_feeds()?.len())
} }
@@ -1159,7 +1159,7 @@ async fn scan_one(
scan.new_entries += 1; scan.new_entries += 1;
} }
for enc in &entry.enclosures { 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 None
} else if let Some(reason) = skipped.get(&enc.url) { } else if let Some(reason) = skipped.get(&enc.url) {
Some(reason.as_str()) Some(reason.as_str())
@@ -1169,8 +1169,8 @@ async fn scan_one(
let now = reject(&ctx.cfg(), feed_cfg, &policy, entry, enc); let now = reject(&ctx.cfg(), feed_cfg, &policy, entry, enc);
if now != was { if now != was {
match now { match now {
Some(reason) => ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason))?, Some(reason) => ctx.db.mark_enclosure(&enc.url, "skipped", Some(reason)).await?,
None => ctx.db.mark_enclosure(&enc.url, "pending", None)?, 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 folder = download::folder_for(&cfg, id, feed_cfg, parsed.title.as_deref());
let dest_dir = cfg.general.download_dir.join(&folder); 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 download::looks_like_torrent(&item.url, item.mime.as_deref()) {
if !ctx.cfg().torrent.enabled { 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 { ctx.out.emit(Event::TorrentDeferred {
feed: id.to_string(), feed: id.to_string(),
url: item.url.clone(), url: item.url.clone(),
@@ -1195,14 +1195,14 @@ async fn scan_one(
} }
if ctx.detach_torrents { if ctx.detach_torrents {
// 'downloading' keeps the next scan from queueing it a second time. // '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()); spawn_torrent(ctx, id.to_string(), item.id, item.url.clone(), dest_dir.clone());
scan.torrents += 1; scan.torrents += 1;
continue; continue;
} }
match torrent_one(ctx, id, item.id, &item.url, &dest_dir).await { match torrent_one(ctx, id, item.id, &item.url, &dest_dir).await {
Ok((path, bytes)) => { 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 { ctx.out.emit(Event::DownloadDone {
feed: id.to_string(), feed: id.to_string(),
enclosure: item.id, enclosure: item.id,
@@ -1220,7 +1220,7 @@ async fn scan_one(
url: item.url.clone(), url: item.url.clone(),
msg: msg.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; scan.failed += 1;
} }
} }
@@ -1245,7 +1245,7 @@ async fn scan_one(
url: item.url.clone(), url: item.url.clone(),
msg: msg.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; scan.failed += 1;
} }
} }
@@ -1334,7 +1334,7 @@ async fn sync_group(
if listed.iter().any(|(_, u)| u == &m.url) { if listed.iter().any(|(_, u)| u == &m.url) {
continue; 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. // Never orphan a downloaded file: keep the feed and say why in the UI.
ctx.db.set_orphaned(&m.id, true)?; ctx.db.set_orphaned(&m.id, true)?;
kept += 1; kept += 1;
@@ -1488,14 +1488,14 @@ async fn fetch_one(
// as if it were an episode. // as if it were an episode.
let _ = tokio::fs::remove_file(&got.tmp).await; let _ = tokio::fs::remove_file(&got.tmp).await;
if !ctx.cfg().torrent.enabled { 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"); anyhow::bail!("body is a torrent and torrents are disabled");
} }
return torrent_one(ctx, feed_id, enclosure, url, dest_dir).await; return torrent_one(ctx, feed_id, enclosure, url, dest_dir).await;
} }
let path = download::place(&got, 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)) Ok((path, got.bytes))
} }
@@ -1513,7 +1513,7 @@ fn spawn_torrent(ctx: &Arc<Ctx>, feed_id: String, enclosure: i64, url: String, d
let db = &ctx.db; let db = &ctx.db;
match outcome { match outcome {
Ok((path, bytes)) => { 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"); tracing::warn!(error = ?e, "could not record the finished torrent");
} }
ctx.out.emit(Event::DownloadDone { ctx.out.emit(Event::DownloadDone {
@@ -1526,7 +1526,7 @@ fn spawn_torrent(ctx: &Arc<Ctx>, feed_id: String, enclosure: i64, url: String, d
} }
Err(e) => { Err(e) => {
let msg = format!("{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 }); ctx.out.emit(Event::DownloadError { feed: feed_id, enclosure, url, msg });
} }
} }
@@ -1539,7 +1539,7 @@ async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> {
let cfg = ctx.cfg(); let cfg = ctx.cfg();
let enc = ctx let enc = ctx
.db .db
.enclosure(id)? .enclosure(id).await?
.ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?; .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?;
if enc.path.is_some() { if enc.path.is_some() {
return Ok(()); // Already here. return Ok(()); // Already here.
@@ -1561,7 +1561,7 @@ async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> {
ctx.out.emit(Event::FeedStart { feed: enc.feed_id.clone() }); ctx.out.emit(Event::FeedStart { feed: enc.feed_id.clone() });
let is_torrent = download::looks_like_torrent(&enc.url, enc.mime.as_deref()); let is_torrent = download::looks_like_torrent(&enc.url, enc.mime.as_deref());
if is_torrent && cfg.torrent.enabled && ctx.detach_torrents { 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); spawn_torrent(ctx, enc.feed_id.clone(), enc.id, enc.url.clone(), dest_dir);
return Ok(()); return Ok(());
} }
@@ -1577,7 +1577,7 @@ async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> {
match result { match result {
Ok((path, bytes)) => { 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 { ctx.out.emit(Event::DownloadDone {
feed: enc.feed_id.clone(), feed: enc.feed_id.clone(),
enclosure: enc.id, enclosure: enc.id,
@@ -1588,7 +1588,7 @@ async fn download_one(ctx: &Arc<Ctx>, id: i64) -> Result<()> {
} }
Err(e) => { Err(e) => {
let msg = format!("{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 { ctx.out.emit(Event::DownloadError {
feed: enc.feed_id.clone(), feed: enc.feed_id.clone(),
enclosure: enc.id, 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("empty", "http://x/empty.xml", "Empty", "parent").unwrap();
ctx.db.upsert_managed("has-file", "http://x/has-file.xml", "Has File", "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 }; 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.record_enclosure("has-file", "g1", &enc).await.unwrap();
ctx.db.mark_downloaded(&enc.url, std::path::Path::new("/downloads/ep.mp3"), 1).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(); let managed = ctx.db.managed_feeds().unwrap();
assert!(!managed.iter().any(|m| m.id == "empty"), "nothing downloaded, so it is forgotten"); 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.upsert_managed("listed", "http://x/l.xml", "Listed", "live-opml").unwrap();
ctx.db.record_entry("promoted", &feed::Entry { guid: "g1".into(), ..Default::default() }).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<String> = ctx.db.managed_feeds().unwrap().into_iter().map(|m| m.id).collect(); let managed: Vec<String> = ctx.db.managed_feeds().unwrap().into_iter().map(|m| m.id).collect();
assert_eq!(managed, ["listed"], "a group still in config is left alone"); assert_eq!(managed, ["listed"], "a group still in config is left alone");

18
src/retention.rs
View File

@@ -45,25 +45,25 @@ pub fn aged(candidates: &[Candidate], cutoff: i64) -> Vec<Candidate> {
.collect() .collect()
} }
pub fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> { pub async fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> {
let mut report = Report::default(); let mut report = Report::default();
// Someone may have deleted a file by hand; the row must stop claiming it exists. // 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 { if !dry_run {
db.mark_reaped(id)?; db.mark_reaped(id).await?;
} }
tracing::debug!(path, "file gone, row reaped"); tracing::debug!(path, "file gone, row reaped");
report.reconciled += 1; report.reconciled += 1;
} }
let candidates = db.reap_candidates()?; let candidates = db.reap_candidates().await?;
if cfg.general.max_age_days > 0 { if cfg.general.max_age_days > 0 {
let cutoff = now() - (cfg.general.max_age_days * 86_400) as i64; let cutoff = now() - (cfg.general.max_age_days * 86_400) as i64;
report.aged_out = aged(&candidates, cutoff); report.aged_out = aged(&candidates, cutoff);
for c in &report.aged_out { 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 { if !dry_run {
report.entries_pruned = db.prune_entries(cutoff)?; report.entries_pruned = db.prune_entries(cutoff)?;
@@ -81,14 +81,14 @@ pub fn run(cfg: &Config, db: &Db, dry_run: bool) -> Result<Report> {
let total: u64 = remaining.iter().map(|c| c.bytes.max(0) as u64).sum(); let total: u64 = remaining.iter().map(|c| c.bytes.max(0) as u64).sum();
report.over_quota = pick(&remaining, total, limit); report.over_quota = pick(&remaining, total, limit);
for c in &report.over_quota { 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) Ok(report)
} }
fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result<u64> { async fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result<u64> {
if dry_run { if dry_run {
return Ok(c.bytes.max(0) as u64); return Ok(c.bytes.max(0) as u64);
} }
@@ -100,7 +100,7 @@ fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result<u64> {
tracing::warn!(path = c.path, error = %e, "could not delete"); tracing::warn!(path = c.path, error = %e, "could not delete");
return Ok(0); return Ok(0);
} }
db.mark_reaped(c.id)?; db.mark_reaped(c.id).await?;
Ok(size) Ok(size)
} }
@@ -176,7 +176,7 @@ mod tests {
) )
.unwrap(); .unwrap();
let got: Vec<i64> = db.reap_candidates().unwrap().iter().map(|c| c.id).collect(); let got: Vec<i64> = db.reap_candidates().await.unwrap().iter().map(|c| c.id).collect();
assert_eq!( assert_eq!(
got, got,
vec![4, 2, 3], vec![4, 2, 3],

16
src/web.rs
View File

@@ -1422,7 +1422,7 @@ async fn remove_feed(
} }
cfg.save(&state.config_path)?; cfg.save(&state.config_path)?;
state.ctx.reload_cfg(&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) Ok(StatusCode::NO_CONTENT)
} }
@@ -1458,12 +1458,12 @@ async fn download_now(
let enc = state let enc = state
.ctx .ctx
.db .db
.enclosure(id)? .enclosure(id).await?
.ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?; .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?;
if enc.path.is_some() { if enc.path.is_some() {
return Ok(StatusCode::NO_CONTENT); // Already here. return Ok(StatusCode::NO_CONTENT); // Already here.
} }
state.ctx.db.requeue(id)?; state.ctx.db.requeue(id).await?;
state state
.cmds .cmds
.send(Command::Download { enclosure: id }) .send(Command::Download { enclosure: id })
@@ -1487,7 +1487,7 @@ async fn delete_file(
let enc = state let enc = state
.ctx .ctx
.db .db
.enclosure(id)? .enclosure(id).await?
.ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?; .ok_or_else(|| anyhow::anyhow!("no enclosure {id}"))?;
// There is one copy of the file: deleting it deletes everyone's. Say so before doing // 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()); return Err(e.into());
} }
// The row survives as 'reaped', which is what stops the next scan re-downloading it. // 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 { state.events.send(Event::Reaped {
path: enc.path.unwrap_or_default(), path: enc.path.unwrap_or_default(),
bytes: enc.length.unwrap_or(0).max(0) as u64, bytes: enc.length.unwrap_or(0).max(0) as u64,
@@ -1573,7 +1573,7 @@ async fn media(
Path(id): Path<i64>, Path(id): Path<i64>,
req: Request, req: Request,
) -> Response { ) -> 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(); return (StatusCode::NOT_FOUND, "no such enclosure").into_response();
}; };
let Some(path) = enc.path else { let Some(path) = enc.path else {
@@ -1648,9 +1648,9 @@ async fn download_latest(
Path(id): Path<String>, Path(id): Path<String>,
Json(body): Json<HowMany>, Json(body): Json<HowMany>,
) -> Result<Json<serde_json::Value>, ApiError> { ) -> Result<Json<serde_json::Value>, 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 { for enc in &ids {
state.ctx.db.requeue(*enc)?; state.ctx.db.requeue(*enc).await?;
state state
.cmds .cmds
.send(Command::Download { enclosure: *enc }) .send(Command::Download { enclosure: *enc })