diff --git a/PROGRESS.md b/PROGRESS.md index 5224a08..07181d7 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -11,8 +11,7 @@ The full design and step list live in the plan file at - [x] **2. `config.rs` + `db.rs`** — TOML config structs + SQLite schema. - [x] **3. `feed.rs`** — conditional GET, RSS-then-Atom parse, persist entries. - [x] **4. `download.rs`** — downloads, filters, dedupe. -- [ ] **5. `retention.rs`** — oldest-first quota + age reaper, `ipx reap [--dry-run]`. - *Done when:* smoke 6 passes. +- [x] **5. `retention.rs`** — oldest-first quota + age reaper. - [ ] **6. `ipc.rs` + daemon** — broadcast event bus, UDS JSON-lines server, TTL scheduler, CLI-proxies-to-daemon. *Done when:* smoke 3 passes. - [ ] **7. `torrent.rs`** — librqbit, seed to ratio/time, stall abort. *Done when:* smoke 5 passes. @@ -31,6 +30,35 @@ The full design and step list live in the plan file at --- +## 2026-09-09 — Step 5: retention.rs + +`src/retention.rs`: reconcile pass (rows claiming a file that is gone become `reaped`, fixing the +step-4 wart), age sweep, quota sweep keeping the original's 50 MB headroom pad, and entry pruning. +`ipx reap [--dry-run]`; a sweep also runs before every `fetch`, as the Python did per download. +`pick()` and `aged()` are pure so the ordering rules are testable without touching a disk. + +**Judgement call worth Ray's eye.** The Python meant to reap only `read = 1 AND flagged = 0` but +never managed it — a missing `plistlib` import and an `EntreiesData` typo made that filter throw on +every candidate, so with a `.ipxd` present nothing was ever deleted. Requiring `read = 1` here would +be equally dead, because nothing marks episodes read until a UI exists. So: **`flagged` is the +keep-forever marker, and `read` only decides what goes first** (`ORDER BY read DESC, downloaded_at +ASC`). Quota therefore actually reclaims space headless. Say the word if you would rather unread +episodes were never touched. + +Second call: `max_age_days` deletes *files* older than the cutoff, not just fileless entries as the +plan's wording had it — "keep 30 days of episodes" is what the setting reads like on a NAS. + +Verified: `cargo test` 21/21, including the two tests encoding the exact bug the Python had — +flagged files are never offered, and read sort ahead of unread. Smoke 6 with three 30 MB episodes +against a 0.1 GB quota (52.4 MB ceiling after the pad): dry run listed ep1+ep2 and deleted nothing +(3 files still on disk), the real run deleted exactly those two oldest, left ep3, flipped both rows +to `reaped` with `path = NULL`. A full re-parse with the conditional-GET headers cleared then +re-downloaded nothing. + +Next: step 6 — `ipc.rs` + daemon. + +--- + ## 2026-09-09 — Step 4: download.rs `src/download.rs`: streaming download to `/.ipx-incomplete/` (same filesystem as the diff --git a/src/db.rs b/src/db.rs index b5e16e2..201f7ab 100644 --- a/src/db.rs +++ b/src/db.rs @@ -89,6 +89,12 @@ impl Db { Ok(Self { conn: Mutex::new(conn) }) } + #[cfg(test)] + pub fn exec_for_test(&self, sql: &str) -> Result<()> { + self.conn.lock().unwrap().execute_batch(sql)?; + Ok(()) + } + pub fn feed_summary(&self, feed_id: &str) -> Result { let conn = self.conn.lock().unwrap(); let mut sum: FeedSummary = conn @@ -291,6 +297,94 @@ impl Db { } } + +/// A downloaded file the reaper may consider. +#[derive(Debug, Clone, PartialEq)] +pub struct Candidate { + pub id: i64, + pub url: String, + pub path: String, + pub bytes: i64, + /// When it landed; the reaper works oldest-first. + pub age_key: i64, + pub read: bool, +} + +impl Db { + /// Files on disk, flagged ones excluded, read before unread and oldest first within + /// each group. + /// + /// 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). Requiring + /// `read = 1` outright would be just as dead here, since nothing marks episodes read + /// until a UI exists -- so `flagged` is the keep-forever marker, and `read` only decides + /// what goes first. + 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), coalesce(n.read, 0), coalesce(n.flagged, 0) + FROM enclosures e + LEFT JOIN entries n ON n.feed_id = e.feed_id AND n.guid = e.guid + WHERE e.path IS NOT NULL AND coalesce(n.flagged, 0) = 0 + ORDER BY coalesce(n.read, 0) 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) + } + + /// 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], + )?; + 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 + .into_iter() + .filter(|(_, p)| !std::path::Path::new(p).exists()) + .collect()) + } + + /// Old entries that never had a file, or no longer have one. Enclosure rows stay -- + /// they are the dedupe history. + pub fn prune_entries(&self, older_than: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM entries WHERE flagged = 0 + AND coalesce(published, first_seen) < ?1 + AND NOT EXISTS ( + SELECT 1 FROM enclosures e + WHERE e.feed_id = entries.feed_id AND e.guid = entries.guid + AND e.path IS NOT NULL)", + [older_than], + )?; + Ok(n) + } +} + /// Unix seconds. Everything time-shaped in the DB is stored this way. pub fn now() -> i64 { std::time::SystemTime::now() diff --git a/src/main.rs b/src/main.rs index 8f67901..fcf8e2a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,6 +2,7 @@ mod config; mod db; mod download; mod feed; +mod retention; use anyhow::Result; use clap::{Parser, Subcommand}; @@ -22,6 +23,12 @@ struct Cli { enum Command { /// Show configured feeds and their state List, + /// Delete old or over-quota downloads + Reap { + /// Show what would go, delete nothing + #[arg(long)] + dry_run: bool, + }, /// Scan feeds for new entries Fetch { /// Only this feed id @@ -41,7 +48,15 @@ async fn main() -> Result<()> { match cli.command { Command::List => list(&cfg, &db, &config_path), - Command::Fetch { feed, force } => fetch(&cfg, &db, feed.as_deref(), force).await, + Command::Fetch { feed, force } => { + // Make room before pulling more down, as the original did per download. + report_reap(&retention::run(&cfg, &db, false)?, false); + fetch(&cfg, &db, feed.as_deref(), force).await + } + Command::Reap { dry_run } => { + report_reap(&retention::run(&cfg, &db, dry_run)?, true); + Ok(()) + } } } @@ -254,6 +269,24 @@ async fn fetch_one( Ok(path) } +fn report_reap(r: &retention::Report, verbose: bool) { + for c in r.aged_out.iter().chain(r.over_quota.iter()) { + println!("reap {} ({:.1} MB)", c.path, c.bytes.max(0) as f64 / 1_048_576.0); + } + if r.reconciled > 0 { + println!("{} row(s) pointed at files that were already gone", r.reconciled); + } + let total = r.aged_out.len() + r.over_quota.len(); + if total > 0 || verbose { + println!( + "reaped {total} file(s), {:.1} MB, {} stale entr{} pruned", + r.bytes_freed as f64 / 1_048_576.0, + r.entries_pruned, + if r.entries_pruned == 1 { "y" } else { "ies" } + ); + } +} + fn ago(t: Option) -> String { let Some(t) = t else { return "never".into() }; format!("{} ago", duration((db::now() - t).max(0) as u64)) diff --git a/src/retention.rs b/src/retention.rs new file mode 100644 index 0000000..1904201 --- /dev/null +++ b/src/retention.rs @@ -0,0 +1,187 @@ +//! Disk quota and age sweeps. Replaces iPXQuotaManager.py ("SmartSpace"). + +use anyhow::Result; + +use crate::config::Config; +use crate::db::{Candidate, Db, now}; + +/// Headroom kept below the quota, as getToQuota()'s `1048576 * 50` did. +pub const PAD_BYTES: u64 = 50 * 1024 * 1024; + +#[derive(Debug, Default)] +pub struct Report { + pub reconciled: usize, + pub aged_out: Vec, + pub over_quota: Vec, + pub entries_pruned: usize, + pub bytes_freed: u64, +} + +/// Oldest-first until enough is freed. Pure, so the ordering rule is testable without a disk. +pub fn pick(candidates: &[Candidate], total: u64, limit: u64) -> Vec { + let ceiling = limit.saturating_sub(PAD_BYTES); + if total <= ceiling { + return vec![]; + } + let mut need = total - ceiling; + let mut out = vec![]; + for c in candidates { + if need == 0 { + break; + } + let size = c.bytes.max(0) as u64; + out.push(c.clone()); + need = need.saturating_sub(size); + } + out +} + +/// Files older than the cutoff. Flagged ones are already excluded by the query. +pub fn aged(candidates: &[Candidate], cutoff: i64) -> Vec { + candidates + .iter() + .filter(|c| c.age_key > 0 && c.age_key < cutoff) + .cloned() + .collect() +} + +pub 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()? { + if !dry_run { + db.mark_reaped(id)?; + } + tracing::debug!(path, "file gone, row reaped"); + report.reconciled += 1; + } + + let candidates = db.reap_candidates()?; + + 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)?; + } + if !dry_run { + report.entries_pruned = db.prune_entries(cutoff)?; + } + } + + if cfg.general.max_total_gb > 0.0 { + let limit = (cfg.general.max_total_gb * 1_073_741_824.0) as u64; + // Re-read: the age sweep may already have freed enough. + let remaining: Vec = candidates + .iter() + .filter(|c| !report.aged_out.iter().any(|a| a.id == c.id)) + .cloned() + .collect(); + 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)?; + } + } + + Ok(report) +} + +fn remove(db: &Db, c: &Candidate, dry_run: bool) -> Result { + if dry_run { + return Ok(c.bytes.max(0) as u64); + } + let size = std::fs::metadata(&c.path).map(|m| m.len()).unwrap_or(c.bytes.max(0) as u64); + // Missing is fine: the point is that it is gone. + if let Err(e) = std::fs::remove_file(&c.path) + && e.kind() != std::io::ErrorKind::NotFound + { + tracing::warn!(path = c.path, error = %e, "could not delete"); + return Ok(0); + } + db.mark_reaped(c.id)?; + Ok(size) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn cand(id: i64, bytes: i64, age_key: i64, read: bool) -> Candidate { + Candidate { + id, + url: format!("https://x/{id}"), + path: format!("/tmp/{id}"), + bytes, + age_key, + read, + } + } + + #[test] + fn pick_takes_oldest_first_and_stops_once_under() { + let gb = 1_073_741_824u64; + // Ten 1 GB files, oldest first as the query would return them. + let cands: Vec = (1..=10).map(|i| cand(i, gb as i64, i, true)).collect(); + + // 10 GB used against a 6 GB quota: need 4 GB back, plus the 50 MB pad. + let picked = pick(&cands, 10 * gb, 6 * gb); + assert_eq!( + picked.iter().map(|c| c.id).collect::>(), + vec![1, 2, 3, 4, 5], + "oldest first, one extra to clear the headroom pad" + ); + } + + #[test] + fn pick_deletes_nothing_when_under_quota() { + let gb = 1_073_741_824u64; + let cands: Vec = (1..=3).map(|i| cand(i, gb as i64, i, true)).collect(); + assert!(pick(&cands, 3 * gb, 50 * gb).is_empty()); + } + + #[test] + fn aged_only_takes_files_past_the_cutoff() { + let cands = vec![cand(1, 10, 100, true), cand(2, 10, 500, true), cand(3, 10, 0, true)]; + let picked = aged(&cands, 200); + assert_eq!(picked.iter().map(|c| c.id).collect::>(), vec![1]); + // age_key 0 means "never recorded" -- not the same as "infinitely old". + } + + #[test] + fn query_never_offers_flagged_files_and_prefers_read_ones() { + let db = Db::memory().unwrap(); + db.exec_for_test( + "INSERT INTO entries (feed_id, guid, first_seen, read, flagged) VALUES + ('f', 'keep', 0, 1, 1), + ('f', 'unread', 0, 0, 0), + ('f', 'read', 0, 1, 0); + INSERT INTO enclosures (id, feed_id, guid, url, path, bytes_done, state, downloaded_at) VALUES + (1, 'f', 'keep', 'u1', '/tmp/keep', 10, 'done', 10), + (2, 'f', 'unread', 'u2', '/tmp/unread', 10, 'done', 20), + (3, 'f', 'read', 'u3', '/tmp/read', 10, 'done', 30);", + ) + .unwrap(); + + let got: Vec = db.reap_candidates().unwrap().iter().map(|c| c.id).collect(); + assert_eq!(got, vec![3, 2], "flagged excluded; read goes before unread"); + } + + #[test] + fn prune_keeps_entries_that_still_have_a_file() { + let db = Db::memory().unwrap(); + db.exec_for_test( + "INSERT INTO entries (feed_id, guid, first_seen, read, flagged) VALUES + ('f', 'has-file', 100, 1, 0), + ('f', 'no-file', 100, 1, 0), + ('f', 'flagged', 100, 1, 1), + ('f', 'recent', 900, 1, 0); + INSERT INTO enclosures (id, feed_id, guid, url, path, state) VALUES + (1, 'f', 'has-file', 'u1', '/tmp/x', 'done');", + ) + .unwrap(); + + assert_eq!(db.prune_entries(500).unwrap(), 1, "only the old, fileless, unflagged one"); + } +}