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

Steps 7 and 8: torrents, OPML, and polish

Browse Source 
librqbit replaces the vendored BitTorrent 4.2.1 tree. Torrents download
in place because seeding serves the files it downloaded, so the planned
stage-then-move would have broken it. The stall budget now also covers
magnet metadata resolution, which otherwise never returns against a dead
swarm and wedged the scan.

Adds add/rm/import/export, tracing setup, systemd units and README.

A successful swarm download is unverified: no reachable peers here.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RPyeapneuXrCdojsaiXGbe
This commit is contained in:
Ray Slakinski
2026-09-09 21:02:22 +00:00
parent b12e0c46dd
commit 833c07240b
10 changed files with 651 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
PROGRESS.md
View File

@@ -13,8 +13,9 @@ The full design and step list live in the plan file at
- [x] **4. `download.rs`** — downloads, filters, dedupe.
- [x] **5. `retention.rs`** — oldest-first quota + age reaper.
- [x] **6. `ipc.rs` + daemon** — UDS JSON-lines server, TTL scheduler, CLI-proxies-to-daemon.
- [ ] **7. `torrent.rs`** — librqbit, seed to ratio/time, stall abort. *Done when:* smoke 5 passes.
- [ ] **8. OPML + polish** — import/export, add/rm/status, tracing setup, systemd unit, README.
- [x] **7. `torrent.rs`** — librqbit, seed to ratio/time, stall abort. (swarm download unverified —
see the step 7 entry)
- [x] **8. OPML + polish** — import/export, add/rm/status, tracing setup, systemd units, README.
## Smoke tests
@@ -29,6 +30,54 @@ The full design and step list live in the plan file at
---
## 2026-09-09 — Step 8: OPML and polish
`ipx add <url>` fetches the feed to name it from its own title (`Accidental Tech Podcast` ->
`accidental-tech-podcast`); a feed that cannot be reached is still added, named from its URL, rather
than refused. `ipx rm` leaves downloads and history alone, so re-adding a feed does not re-pull its
back catalogue. `ipx import`/`export` walk nested OPML folder outlines and skip URLs already
subscribed. `tracing` logs to stderr, `IPX_LOG` sets the filter. `contrib/` has a systemd user unit
for the daemon, plus a timer and one-shot service for the no-daemon style (with the caveat that
without a daemon there is no socket for a UI).
Verified: `cargo test` 27/27. Round-trip — exported 3 feeds with real titles, removed one,
re-imported: exactly the missing one came back, no duplicates, config still mode 0600. Quickstart
from a genuinely empty home with no env overrides: `list` on a missing config, `add`, `fetch`
(1 downloaded, 1 explicit skipped, 1 torrent 404), `list`; every path resolved under `$HOME`.
---
## 2026-09-09 — Step 7: torrent.rs
`src/torrent.rs`: a librqbit `Session` started lazily on first torrent (binding ports and starting a
DHT for a config that has never seen a torrent would be rude). `.torrent` URLs and magnets both go
through `AddTorrent::from_url`. Progress is polled once a second and reported through the same
`Progress` events HTTP downloads use.
Downloads go **straight into the feed folder** rather than staging and moving. The plan said move
then seed, which cannot work — seeding serves the files it downloaded, so moving them first breaks
it. In-place also removes a copy the original had to do.
Seeding stops at `seed_ratio` or `seed_time_mins`, whichever comes first, then the torrent is
released from the session (files kept).
**Bug the smoke test caught:** the stall budget only covered the download loop, but resolving a
magnet's metadata happens inside `add_torrent`, which against a dead swarm never returns — a
torrent nobody seeds wedged the scan indefinitely. `add_torrent` is now wrapped in the same budget.
Verified: `cargo test` 27/27 (ratio incl. the divide-by-zero case, stall_mins = 0 not meaning
"abort instantly"). Routing: with `enabled = false` a torrent enclosure is marked
`skipped/torrents disabled` and never attempted. Session startup works here. Stall abort measured
end to end: with `stall_mins = 1`, a dead magnet failed at 20:58:23 -> 20:59:24, exactly 61s, and
the row recorded `error / no metadata after 1 minutes, gave up`.
**Not verified: an actual successful swarm download.** This sandbox has no reachable peers, so
smoke 5's happy path — payload lands, seeding stops at the ratio — has not been run. The code paths
either side of it are tested; the swarm itself needs a real network. Worth running once against a
live torrent feed before trusting it.
---
## 2026-09-09 — Step 6: ipc.rs + daemon
`src/ipc.rs`: `Event` and `Command` as serde-tagged enums (`{"ev":...}` / `{"cmd":...}`), one JSON

89
README.md
View File

@@ -11,21 +11,100 @@ Python 2 engine behind **iPodderX** (2004-2008, Ray Slakinski & August Trometer)
open-sourced under the MIT License in 2010.
What carries over: the feed scan and TTL handling, GUID/URL dedupe, per-feed and per-date download
folders, keyword filters, the explicit-content filter, torrent enclosures, and "SmartSpace" — the
folders, keyword filters, the explicit-content filter, torrent enclosures, and "SmartSpace" -- the
oldest-first disk quota reaper.
What does not: iTunes and iPhoto export via AppleScript, text-to-speech enclosures, the Windows
WMP/COM paths, XML plists and Python pickles for state, the `directory.iPodderX.com` survey ping,
3DES-encrypted preferences, and the `printMSG` stdout protocol (replaced by a JSON-lines socket).
## Status
## Quickstart
Under construction. See [PROGRESS.md](PROGRESS.md) for what works today.
```sh
cargo install --path .
ipx add https://atp.fm/rss # names the feed from its own title
ipx list
ipx fetch # scan now
ipx daemon # or run continuously, honouring each feed's <ttl>
```
Config lives at `~/.config/ipx/config.toml` (mode 0600, since it may hold feed passwords);
state at `~/.local/share/ipx/state.db`. Override with `IPX_CONFIG` and `IPX_DATA_DIR`.
Set `IPX_LOG=ipx=debug` for verbose logging on stderr.
## Commands
| command | what it does |
|---|---|
| `ipx add <url> [--folder X] [--keywords a,b]` | subscribe; the id comes from the feed title |
| `ipx rm <feed>` | unsubscribe; downloads and history are kept |
| `ipx list` / `ipx status` | subscriptions and their state |
| `ipx fetch [FEED] [--force]` | scan; `--force` ignores the TTL |
| `ipx reap [--dry-run]` | run retention now |
| `ipx import/export <file.opml>` | move subscriptions in or out |
| `ipx daemon` | scheduler plus the control socket |
Any command with a wire form probes the socket first: if a daemon is running it does the work,
and the CLI just renders the events it streams back. `--local` forces in-process execution.
## Configuration
`~/.config/ipx/config.toml`, state in `~/.local/share/ipx/state.db`. See PROGRESS.md until the
usage section lands.
```toml
[general]
download_dir = "~/Podcasts"
socket = "/run/user/1000/ipx.sock" # default: $XDG_RUNTIME_DIR/ipx.sock
interval_mins = 60 # default poll; a feed's own <ttl> wins when longer
organize = "feed" # "feed" | "date"
max_total_gb = 50 # 0 = unlimited
max_age_days = 30 # 0 = keep forever
[torrent]
enabled = true
seed_ratio = 1.0 # stop seeding at this ratio ...
seed_time_mins = 60 # ... or after this long, whichever comes first
port_range = "6881-6889"
stall_mins = 30 # give up on a torrent making no progress
[feeds.atp]
url = "https://atp.fm/rss"
folder = "Accidental Tech Podcast" # default: the feed title
keywords = ["deep dive"] # OR across keywords, AND within one
allow_explicit = false
auto_download = true
max_new_per_check = 3 # the rest wait for the next scan
username = "ray" # optional HTTP basic auth
password_env = "IPX_ATP_PASS" # or a literal `password`
```
Retention keeps files that are `flagged` in the database, and deletes read episodes before unread
ones, oldest first.
## Socket protocol
Newline-delimited JSON over a Unix socket, both directions.
```sh
$ printf '{"cmd":"fetch","force":true}\n' | socat - UNIX-CONNECT:$XDG_RUNTIME_DIR/ipx.sock
{"ev":"feed_start","feed":"atp"}
{"ev":"progress","feed":"atp","url":"...","file":"ep1.mp3","done":8192,"total":3000000}
{"ev":"download_done","feed":"atp","url":"...","path":"...","bytes":3000000}
{"ev":"feed_done","feed":"atp","new":1,"downloaded":1,"failed":0,"torrents":0}
{"ev":"scan_done","feeds":1}
```
Commands: `fetch` (optional `feed`, `force`), `reap` (optional `dry_run`), `status`.
Events: `feed_start`, `feed_skip`, `feed_done`, `feed_error`, `progress`, `download_done`,
`download_error`, `torrent_deferred`, `reaped`, `reap_done`, `scan_done`, `status`, `error`.
`scan_done`, `reap_done` and `status` are terminal -- a client that asked for work stops there.
Progress is throttled to whole percents. The stream is a broadcast, so a client attached to a busy
daemon also sees that daemon's other work.
## Running it as a service
`contrib/` has a systemd user unit for the daemon, and a timer plus one-shot service if you would
rather run periodic scans with no daemon (in which case there is no socket for a UI to attach to).
## License

7
contrib/ipx-scan.service Normal file
View File

@@ -0,0 +1,7 @@
[Unit]
Description=ipx feed scan (one shot)
[Service]
Type=oneshot
ExecStart=%h/.cargo/bin/ipx fetch
Environment=IPX_LOG=ipx=info

18
contrib/ipx.service Normal file
View File

@@ -0,0 +1,18 @@
# User unit: install to ~/.config/systemd/user/ipx.service, then
# systemctl --user enable --now ipx
# The socket lands in $XDG_RUNTIME_DIR/ipx.sock by default, so a UI running as the
# same user can attach without extra configuration.
[Unit]
Description=ipx podcatcher
After=network-online.target
Wants=network-online.target
[Service]
Type=simple
ExecStart=%h/.cargo/bin/ipx daemon
Restart=on-failure
RestartSec=30
Environment=IPX_LOG=ipx=info
[Install]
WantedBy=default.target

16
contrib/ipx.timer Normal file
View File

@@ -0,0 +1,16 @@
# Alternative to the daemon: a periodic one-shot scan, closer to how the original
# iPodderX agent was driven. Use this OR ipx.service, not both -- with no daemon
# running there is no socket, so a UI cannot attach.
#
# Install ipx-scan.service and ipx.timer to ~/.config/systemd/user/, then
# systemctl --user enable --now ipx.timer
[Unit]
Description=Periodic ipx feed scan
[Timer]
OnBootSec=5min
OnUnitActiveSec=1h
Persistent=true
[Install]
WantedBy=timers.target

45
src/config.rs
View File

@@ -188,6 +188,33 @@ fn default_socket() -> PathBuf {
}
}
/// Feed ids are the TOML table key, so they must be readable and punctuation-free.
pub fn slug(text: &str) -> String {
let mut out = String::new();
for c in text.chars() {
if c.is_ascii_alphanumeric() {
out.push(c.to_ascii_lowercase());
} else if c.is_alphanumeric() {
out.push(c); // Keep non-ASCII letters; TOML bare keys are stricter, but we quote.
} else if !out.ends_with('-') {
out.push('-');
}
}
let out = out.trim_matches('-').to_owned();
let out: String = out.chars().take(40).collect();
let out = out.trim_matches('-').to_owned();
if out.is_empty() { "feed".into() } else { out }
}
/// `slug`, with a numeric suffix if that id is taken.
pub fn unique_slug(text: &str, taken: &BTreeMap<String, Feed>) -> String {
let base = slug(text);
if !taken.contains_key(&base) {
return base;
}
(2..).map(|n| format!("{base}-{n}")).find(|s| !taken.contains_key(s)).unwrap()
}
fn home() -> PathBuf {
dirs::home_dir().unwrap_or_else(|| PathBuf::from("."))
}
@@ -241,6 +268,24 @@ mod tests {
assert_eq!(t.ports(), (6881, 6889), "reversed range is not a range");
}
#[test]
fn slugs_are_readable_and_unique() {
assert_eq!(slug("Accidental Tech Podcast"), "accidental-tech-podcast");
assert_eq!(slug(" The Daily!! "), "the-daily");
assert_eq!(slug("99% Invisible"), "99-invisible");
assert_eq!(slug("///"), "feed");
assert_eq!(slug("").len(), 4);
assert!(slug(&"x".repeat(100)).len() <= 40);
let mut taken = BTreeMap::new();
taken.insert("the-daily".to_string(), Feed {
url: "u".into(), folder: None, keywords: vec![], allow_explicit: false,
auto_download: true, max_new_per_check: None, username: None,
password: None, password_env: None,
});
assert_eq!(unique_slug("The Daily", &taken), "the-daily-2");
}
#[test]
fn password_env_wins_over_literal() {
let mut f = Feed {

14
src/download.rs
View File

@@ -258,20 +258,6 @@ pub fn matches_keywords(keywords: &[String], haystacks: &[&str]) -> bool {
})
}
/// Formats a progress line, throttled by the caller to ~1% steps as the original's
/// lastDLStepSize guard did.
pub fn progress_line(name: &str, done: u64, total: Option<u64>) -> String {
match total {
Some(t) if t > 0 => format!(
" {name}: {:.1}% ({:.1}/{:.1} MB)",
done as f64 / t as f64 * 100.0,
done as f64 / 1_048_576.0,
t as f64 / 1_048_576.0
),
_ => format!(" {name}: {:.1} MB", done as f64 / 1_048_576.0),
}
}
#[cfg(test)]
mod tests {
use super::*;

18
src/feed.rs
View File

@@ -9,10 +9,7 @@ use crate::config::Feed as FeedCfg;
#[derive(Debug, Default)]
pub struct ParsedFeed {
pub title: Option<String>,
pub link: Option<String>,
pub ttl_mins: Option<u64>,
/// Feed-level explicit flag; per the original, it overrides the entry level.
pub explicit: bool,
pub entries: Vec<Entry>,
}
@@ -148,9 +145,7 @@ fn from_rss(ch: rss::Channel) -> ParsedFeed {
ParsedFeed {
title: non_empty(Some(ch.title())),
link: non_empty(Some(ch.link())),
ttl_mins: ch.ttl().and_then(|t| t.trim().parse().ok()),
explicit,
entries,
}
}
@@ -205,13 +200,7 @@ fn from_atom(feed: atom_syndication::Feed) -> ParsedFeed {
ParsedFeed {
title: non_empty(Some(feed.title().as_str())),
link: feed
.links()
.iter()
.find(|l| l.rel() == "alternate")
.map(|l| l.href().to_owned()),
ttl_mins: None,
explicit: false,
entries,
}
}
@@ -261,7 +250,6 @@ mod tests {
assert_eq!(feed.title.as_deref(), Some("Test Cast"));
assert_eq!(feed.ttl_mins, Some(45));
assert!(!feed.explicit, "feed-level explicit is 'no'");
assert_eq!(feed.entries.len(), 3);
let ep = &feed.entries[0];
@@ -296,8 +284,10 @@ mod tests {
<enclosure url="https://x/a.mp3" length="1" type="audio/mpeg"/></item>
</channel></rss>"#;
let feed = parse(xml).unwrap();
assert!(feed.explicit);
assert!(feed.entries[0].explicit, "feed level must win");
assert!(
feed.entries[0].explicit,
"the entry says nothing; the feed-level flag must still mark it explicit"
);
}
#[test]

252
src/main.rs
View File

@@ -4,8 +4,9 @@ mod download;
mod feed;
mod ipc;
mod retention;
mod torrent;
use anyhow::Result;
use anyhow::{Context, Result};
use clap::{Parser, Subcommand};
use ipc::{Command as Cmd, Emitter, Event};
use std::path::PathBuf;
@@ -46,6 +47,22 @@ enum Command {
},
/// Counts of feeds, pending and downloaded enclosures
Status,
/// Subscribe to a feed
Add {
url: String,
/// Download folder name (default: the feed title)
#[arg(long)]
folder: Option<String>,
/// Only take enclosures matching these keywords
#[arg(long, value_delimiter = ',')]
keywords: Vec<String>,
},
/// Unsubscribe. Downloads and history are left alone.
Rm { feed: String },
/// Add every feed in an OPML file
Import { file: PathBuf },
/// Write subscriptions out as OPML
Export { file: PathBuf },
/// Run the scheduler and serve the control socket
Daemon,
}
@@ -56,6 +73,17 @@ struct Ctx {
db: db::Db,
client: reqwest::Client,
out: Emitter,
/// Started on first use: a BitTorrent session binds ports and starts a DHT, which is
/// rude to do for a config that has never seen a torrent.
torrents: tokio::sync::OnceCell<torrent::Torrents>,
}
impl Ctx {
async fn torrents(&self) -> Result<&torrent::Torrents> {
self.torrents
.get_or_try_init(|| torrent::Torrents::new(&self.cfg))
.await
}
}
#[tokio::main]
@@ -80,7 +108,12 @@ async fn main() -> Result<()> {
}
Command::Reap { dry_run } => Some(Cmd::Reap { dry_run: *dry_run }),
Command::Status => Some(Cmd::Status),
Command::List | Command::Daemon => None,
Command::List
| Command::Daemon
| Command::Add { .. }
| Command::Rm { .. }
| Command::Import { .. }
| Command::Export { .. } => None,
};
if let Some(cmd) = &wire_cmd
&& !cli.local
@@ -96,11 +129,18 @@ async fn main() -> Result<()> {
.user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION")))
.build()?,
out: Emitter::terminal(),
torrents: tokio::sync::OnceCell::new(),
};
match cli.command {
Command::List => list(&ctx, &config_path),
Command::Daemon => daemon(ctx).await,
Command::Add { url, folder, keywords } => {
add(ctx, &config_path, &url, folder, keywords).await
}
Command::Rm { feed } => rm(ctx, &config_path, &feed),
Command::Import { file } => import(ctx, &config_path, &file).await,
Command::Export { file } => export(&ctx, &file),
_ => run(&ctx, wire_cmd.expect("only List and Daemon have no wire form")).await,
}
}
@@ -174,6 +214,146 @@ async fn shutdown() {
}
}
/// Subscribes to one feed, naming it from its own title.
async fn add(
mut ctx: Ctx,
config_path: &std::path::Path,
url: &str,
folder: Option<String>,
keywords: Vec<String>,
) -> Result<()> {
if let Some((id, _)) = ctx.cfg.feeds.iter().find(|(_, f)| f.url == url) {
anyhow::bail!("already subscribed as {id:?}");
}
let id = add_one(&mut ctx, url, folder, keywords).await?;
ctx.cfg.save(config_path)?;
println!("added {id}");
Ok(())
}
/// Returns the new feed id. The title needs a fetch, so a feed that cannot be reached is
/// still added -- under a slug derived from its URL -- rather than refused.
async fn add_one(
ctx: &mut Ctx,
url: &str,
folder: Option<String>,
keywords: Vec<String>,
) -> Result<String> {
let probe = config::Feed {
url: url.to_owned(),
folder: folder.clone(),
keywords: keywords.clone(),
allow_explicit: false,
auto_download: true,
max_new_per_check: None,
username: None,
password: None,
password_env: None,
};
let title = match feed::fetch(&ctx.client, &probe, None, None).await {
Ok(feed::Fetched::Body { bytes, .. }) => feed::parse(&bytes)
.ok()
.and_then(|f| f.title)
.unwrap_or_else(|| url_stem(url)),
_ => {
tracing::warn!(url, "could not read the feed; naming it from its URL");
url_stem(url)
}
};
let id = config::unique_slug(&title, &ctx.cfg.feeds);
ctx.cfg.feeds.insert(id.clone(), probe);
Ok(id)
}
/// Host plus last path segment, for naming a feed we could not read.
fn url_stem(url: &str) -> String {
url::Url::parse(url)
.ok()
.and_then(|u| u.host_str().map(str::to_owned))
.unwrap_or_else(|| url.to_owned())
}
fn rm(mut ctx: Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> {
if ctx.cfg.feeds.remove(feed).is_none() {
anyhow::bail!("no feed with id {feed:?}");
}
ctx.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");
Ok(())
}
async fn import(mut ctx: Ctx, config_path: &std::path::Path, file: &std::path::Path) -> Result<()> {
let text = std::fs::read_to_string(file)
.with_context(|| format!("reading {}", file.display()))?;
let doc = opml::OPML::from_str(&text).map_err(|e| anyhow::anyhow!("parsing OPML: {e}"))?;
let mut found = vec![];
collect_outlines(&doc.body.outlines, &mut found);
let mut added = 0;
for (title, url) in found {
if ctx.cfg.feeds.values().any(|f| f.url == url) {
continue;
}
// Name it from the OPML title rather than refetching every feed.
let id = config::unique_slug(&title, &ctx.cfg.feeds);
ctx.cfg.feeds.insert(
id.clone(),
config::Feed {
url,
folder: None,
keywords: vec![],
allow_explicit: false,
auto_download: true,
max_new_per_check: None,
username: None,
password: None,
password_env: None,
},
);
println!("added {id}");
added += 1;
}
ctx.cfg.save(config_path)?;
println!("{added} feed(s) imported");
Ok(())
}
/// OPML nests feeds inside folder outlines, so this walks the whole tree.
fn collect_outlines(outlines: &[opml::Outline], out: &mut Vec<(String, String)>) {
for o in outlines {
if let Some(url) = &o.xml_url {
let title = o.title.clone().unwrap_or_else(|| o.text.clone());
out.push((title, url.clone()));
}
collect_outlines(&o.outlines, out);
}
}
fn export(ctx: &Ctx, file: &std::path::Path) -> Result<()> {
let mut doc = opml::OPML::default();
doc.head = Some(opml::Head {
title: Some("ipx subscriptions".into()),
..Default::default()
});
for (id, feed) in &ctx.cfg.feeds {
let title = ctx
.db
.feed_summary(id)
.ok()
.and_then(|s| s.title)
.unwrap_or_else(|| id.clone());
doc.add_feed(&title, &feed.url);
}
let xml = doc.to_string().map_err(|e| anyhow::anyhow!("writing OPML: {e}"))?;
std::fs::write(file, xml).with_context(|| format!("writing {}", file.display()))?;
println!("exported {} feed(s) to {}", ctx.cfg.feeds.len(), file.display());
Ok(())
}
fn list(ctx: &Ctx, config_path: &std::path::Path) -> Result<()> {
if ctx.cfg.feeds.is_empty() {
println!("No feeds configured in {}", config_path.display());
@@ -332,8 +512,8 @@ async fn scan_one(
for item in ctx.db.pending(id, budget)? {
if download::looks_like_torrent(&item.url, item.mime.as_deref()) {
// Step 7 owns these; recorded so nothing re-queues them meanwhile.
ctx.db.mark_enclosure(&item.url, "torrent", None)?;
if !ctx.cfg.torrent.enabled {
ctx.db.mark_enclosure(&item.url, "skipped", Some("torrents disabled"))?;
ctx.out.emit(Event::TorrentDeferred {
feed: id.to_string(),
url: item.url.clone(),
@@ -341,6 +521,30 @@ async fn scan_one(
scan.torrents += 1;
continue;
}
match torrent_one(ctx, id, &item.url, &dest_dir).await {
Ok((path, bytes)) => {
ctx.db.mark_downloaded(&item.url, &path, bytes)?;
ctx.out.emit(Event::DownloadDone {
feed: id.to_string(),
url: item.url.clone(),
path: path.display().to_string(),
bytes,
});
scan.downloaded += 1;
}
Err(e) => {
let msg = format!("{e:#}");
ctx.out.emit(Event::DownloadError {
feed: id.to_string(),
url: item.url.clone(),
msg: msg.clone(),
});
ctx.db.mark_enclosure(&item.url, "error", Some(&msg))?;
scan.failed += 1;
}
}
continue;
}
match fetch_one(ctx, id, feed_cfg, &item.url, &dest_dir).await {
Ok((path, bytes)) => {
ctx.out.emit(Event::DownloadDone {
@@ -417,10 +621,14 @@ async fn fetch_one(
.await?;
if matches!(got.kind, download::Kind::Torrent) {
// The MIME lied. Don't file a .torrent as an episode.
// The MIME lied. Hand the URL to the torrent session instead of filing a .torrent
// as if it were an episode.
let _ = tokio::fs::remove_file(&got.tmp).await;
ctx.db.mark_enclosure(url, "torrent", None)?;
anyhow::bail!("body is a torrent, deferred to the torrent downloader");
if !ctx.cfg.torrent.enabled {
ctx.db.mark_enclosure(url, "skipped", Some("torrents disabled"))?;
anyhow::bail!("body is a torrent and torrents are disabled");
}
return torrent_one(ctx, feed_id, url, dest_dir).await;
}
let path = download::place(&got, dest_dir).await?;
@@ -428,6 +636,36 @@ async fn fetch_one(
Ok((path, got.bytes))
}
/// Torrent progress is reported the same way an HTTP download's is, throttled to whole
/// percents so a UI is not flooded.
async fn torrent_one(
ctx: &Ctx,
feed_id: &str,
url: &str,
dest_dir: &std::path::Path,
) -> Result<(PathBuf, u64)> {
let name = download::filename_for(url, None);
let mut last_pct = -1i64;
ctx.torrents()
.await?
.fetch(&ctx.cfg, url, dest_dir, |done, total| {
if total > 0 {
let pct = (done * 100 / total) as i64;
if pct > last_pct {
last_pct = pct;
ctx.out.emit(Event::Progress {
feed: feed_id.to_string(),
url: url.to_string(),
file: name.clone(),
done,
total: Some(total),
});
}
}
})
.await
}
fn ago(t: Option<i64>) -> String {
let Some(t) = t else { return "never".into() };
format!("{} ago", duration((db::now() - t).max(0) as u64))

176
src/torrent.rs Normal file
View File

@@ -0,0 +1,176 @@
//! Torrent enclosures via librqbit. Replaces iPXDownloader.getTorrent and the vendored
//! BitTorrent 4.2.1 + khashmir tree.
use anyhow::{Context, Result, bail};
use librqbit::{AddTorrent, AddTorrentOptions, AddTorrentResponse, Session, SessionOptions};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
use crate::config::Config;
pub struct Torrents {
session: Arc<Session>,
}
/// How long a torrent may make no progress before we give up on it. The original called
/// this torrentMaxBeatTime and counted display ticks; a wall clock is easier to reason about.
fn stall_limit(cfg: &Config) -> Duration {
Duration::from_secs(cfg.torrent.stall_mins.max(1) * 60)
}
impl Torrents {
pub async fn new(cfg: &Config) -> Result<Self> {
let (lo, _hi) = cfg.torrent.ports();
let listen_addr: SocketAddr = format!("0.0.0.0:{lo}").parse()?;
let opts = SessionOptions {
listen: Some(librqbit::ListenerOptions {
listen_addr,
..Default::default()
}),
client_name_and_version: Some(concat!("ipx ", env!("CARGO_PKG_VERSION")).into()),
..Default::default()
};
// The session root is only a fallback; every torrent names its own output folder.
let session = Session::new_with_opts(cfg.general.download_dir.clone(), opts)
.await
.context("starting the torrent session")?;
Ok(Self { session })
}
/// Downloads a torrent (a .torrent URL or a magnet link) straight into `dest_dir`,
/// seeds it up to the configured limit, then stops.
///
/// Downloading in place rather than into a staging dir is deliberate: seeding serves
/// the files it downloaded, so moving them first would break it.
pub async fn fetch(
&self,
cfg: &Config,
url: &str,
dest_dir: &Path,
mut on_progress: impl FnMut(u64, u64),
) -> Result<(PathBuf, u64)> {
tokio::fs::create_dir_all(dest_dir).await?;
let limit = stall_limit(cfg);
// Resolving a magnet's metadata happens inside add_torrent, and against a dead
// swarm it never returns. The stall budget has to cover this phase too, or a
// torrent nobody is seeding wedges the scan forever.
let added = tokio::time::timeout(
limit,
self.session.add_torrent(
AddTorrent::from_url(url),
Some(AddTorrentOptions {
output_folder: Some(dest_dir.to_string_lossy().into_owned()),
overwrite: true,
..Default::default()
}),
),
)
.await
.map_err(|_| {
anyhow::anyhow!(
"no metadata after {} minutes, gave up",
limit.as_secs() / 60
)
})?
.context("adding the torrent")?;
let (id, handle) = match added {
AddTorrentResponse::Added(id, h) | AddTorrentResponse::AlreadyManaged(id, h) => (id, h),
AddTorrentResponse::ListOnly(_) => bail!("torrent added in list-only mode"),
};
let mut last_progress = 0u64;
let mut last_move = Instant::now();
loop {
tokio::time::sleep(Duration::from_secs(1)).await;
let stats = handle.stats();
if let Some(err) = &stats.error {
let _ = self.session.delete(id.into(), false).await;
bail!("torrent failed: {err}");
}
on_progress(stats.progress_bytes, stats.total_bytes);
if stats.finished {
break;
}
if stats.progress_bytes > last_progress {
last_progress = stats.progress_bytes;
last_move = Instant::now();
} else if last_move.elapsed() > limit {
// Leave the partial files behind; the row records the failure.
let _ = self.session.delete(id.into(), false).await;
bail!("no progress for {} minutes, gave up", limit.as_secs() / 60);
}
}
let name = handle.name().unwrap_or_else(|| "torrent".to_owned());
let path = handle.output_folder().join(&name);
let size = handle.stats().total_bytes;
self.seed(cfg, &handle).await;
self.session
.delete(id.into(), false)
.await
.context("releasing the torrent")?;
Ok((path, size))
}
/// Seeds until the ratio or the time limit is reached, whichever comes first.
async fn seed(&self, cfg: &Config, handle: &Arc<librqbit::ManagedTorrent>) {
if cfg.torrent.seed_ratio <= 0.0 || cfg.torrent.seed_time_mins == 0 {
return;
}
let deadline = Instant::now() + Duration::from_secs(cfg.torrent.seed_time_mins * 60);
loop {
let stats = handle.stats();
if ratio(stats.uploaded_bytes, stats.total_bytes) >= cfg.torrent.seed_ratio {
tracing::info!(ratio = cfg.torrent.seed_ratio, "seed ratio reached");
return;
}
if Instant::now() >= deadline {
tracing::info!(mins = cfg.torrent.seed_time_mins, "seed time reached");
return;
}
tokio::time::sleep(Duration::from_secs(5)).await;
}
}
}
/// Uploaded over total. A zero-byte torrent counts as fully seeded rather than dividing by zero.
pub fn ratio(uploaded: u64, total: u64) -> f64 {
if total == 0 {
return f64::INFINITY;
}
uploaded as f64 / total as f64
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ratio_handles_the_empty_torrent() {
assert_eq!(ratio(0, 100), 0.0);
assert_eq!(ratio(50, 100), 0.5);
assert_eq!(ratio(200, 100), 2.0);
assert!(ratio(0, 0).is_infinite(), "must not divide by zero or seed forever");
}
#[test]
fn stall_limit_is_never_zero() {
let mut cfg = Config::default();
cfg.torrent.stall_mins = 0;
assert_eq!(stall_limit(&cfg), Duration::from_secs(60), "0 would abort instantly");
cfg.torrent.stall_mins = 30;
assert_eq!(stall_limit(&cfg), Duration::from_secs(1800));
}
}