From 19309d609f6c4e8bcaa16327a05c3a5318998037 Mon Sep 17 00:00:00 2001 From: rays Date: Thu, 10 Sep 2026 02:51:21 +0000 Subject: [PATCH] Run torrents off the command worker A torrent ran inline on the single sequential worker, so it blocked every feed scan, HTTP download and status command behind it -- for up to stall_mins waiting on metadata, and for up to seed_time_mins seeding after finishing. A live daemon was wedged with 47 pending torrents and would not answer a status command for 15s. spawn_torrent detaches the job behind a 2-permit semaphore and marks the row 'downloading' so a rescan cannot queue it twice. One-shot CLI runs stay inline, or the process would exit mid-download. Torrents themselves verified working: a 755 MB Debian netinst downloaded to completion against a real swarm. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01AdXho5tTkjFLeUXKbEjKBh --- PROGRESS.md | 35 ++++++++++++++++++ src/main.rs | 100 +++++++++++++++++++++++++++++++++++++++++----------- 2 files changed, 115 insertions(+), 20 deletions(-) diff --git a/PROGRESS.md b/PROGRESS.md index a2e0087..56a3a7a 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -56,6 +56,41 @@ and until now nothing set them. --- +## 2026-09-10 — Torrents: they work, and one was freezing the whole daemon + +Asked "is torrents working?". The honest answer had been "unverified since step 7" — so it got tested +properly. + +**They work.** End to end against a real swarm: Debian 13.6.0 netinst, 791,674,880 bytes, completed, +`file` confirms a bootable ISO, row marked `done`. Metadata resolution, transfer, placement and +completion all good. + +**But the daemon was wedged.** Ray's CT-log feed had 47 pending torrents and nothing was progressing. +The worker was blocked: a `status` command over the control socket got no reply in 15s. Torrents ran +*inline on the single command worker*, so one torrent stopped every feed scan, every HTTP download +and every status command behind it — for up to `stall_mins` (30 default) waiting on metadata, and +worse, `Torrents::fetch` blocks on seeding for up to `seed_time_mins` (60 default) **after** +finishing. One successful torrent could have frozen podcast fetching for an hour. + +Torrents are now detached: `spawn_torrent` runs the job on its own task behind a 2-permit semaphore, +the row is marked `downloading` so a rescan does not queue it twice, and the worker moves straight +on. A one-shot CLI run still runs them inline, or the process would exit mid-download. Verified: the +worker now answers `status` in 0.0s. + +A diagnostic misstep worth recording: `/api/settings` returning 200 was taken as proof the worker was +alive. It is not — the web server is a separate task, so HTTP stays responsive while the queue is +completely stuck. Probing the control socket is the real test. + +**Those CT-log torrents are ~557 GB each, 47 of them.** The RSS advertises `length="0"`, so nothing +warns you; the size only appears once metadata resolves. `auto_download` on that feed is set to +false as a hold, since restarting would otherwise have begun a half-terabyte transfer immediately +(the scheduler's first tick fires at startup). Flip it back in the feed's Settings when wanted. + +Also noted: an aborted torrent leaves librqbit's pre-created file placeholders behind in the +destination folder (zero-length). Not cleaned up yet. + +--- + ## 2026-09-10 — Schedule pickers, and progress painting every row **Pickers.** "Check feeds every" is a number input plus a unit dropdown (minutes/hours/days/weeks) diff --git a/src/main.rs b/src/main.rs index 32cdee3..35d52c2 100644 --- a/src/main.rs +++ b/src/main.rs @@ -86,6 +86,13 @@ pub struct Ctx { /// 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. pub torrents: tokio::sync::OnceCell, + /// Caps how many torrents run at once when they are detached. + pub torrent_slots: Arc, + /// Run torrents off the command worker. A torrent takes minutes to fetch metadata and + /// then seeds for up to an hour, and the worker is sequential -- inline, one torrent + /// stops every feed scan, every HTTP download and every status command behind it. + /// A one-shot CLI run keeps them inline, or the process would exit mid-download. + pub detach_torrents: bool, } impl Ctx { @@ -145,30 +152,34 @@ async fn main() -> Result<()> { return ipc::proxy(&cfg.general.socket, cmd).await; } - let ctx = Ctx { + let is_daemon = matches!(cli.command, Command::Daemon { .. }); + let (events, _) = broadcast::channel(1024); + let ctx = Arc::new(Ctx { cfg: std::sync::RwLock::new(std::sync::Arc::new(cfg)), db, client: reqwest::Client::builder() .user_agent(concat!("ipx/", env!("CARGO_PKG_VERSION"))) .build()?, - out: Emitter::terminal(), + out: if is_daemon { Emitter::socket(events.clone(), false) } else { Emitter::terminal() }, torrents: tokio::sync::OnceCell::new(), - }; + torrent_slots: Arc::new(tokio::sync::Semaphore::new(2)), + detach_torrents: is_daemon, + }); match cli.command { Command::List => list(&ctx, &config_path), - Command::Daemon { web } => daemon(ctx, config_path, web).await, + Command::Daemon { web } => daemon(ctx, config_path, web, events).await, 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::Import { file } => import(ctx, &config_path, &file).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, } } -async fn run(ctx: &Ctx, cmd: Cmd) -> Result<()> { +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. @@ -185,15 +196,18 @@ async fn run(ctx: &Ctx, cmd: Cmd) -> Result<()> { } } -async fn daemon(ctx: Ctx, config_path: PathBuf, web_addr: Option) -> Result<()> { +async fn daemon( + ctx: Arc, + config_path: PathBuf, + web_addr: Option, + events: broadcast::Sender, +) -> Result<()> { let socket = ctx.cfg().general.socket.clone(); if ipc::daemon_is_live(&socket).await { anyhow::bail!("a daemon is already listening on {}", socket.display()); } - let (events, _) = broadcast::channel(1024); let (tx_cmd, mut rx_cmd) = mpsc::channel::(64); - let ctx = Arc::new(Ctx { out: Emitter::socket(events.clone(), false), ..ctx }); let web = start_web(&ctx, &config_path, web_addr, &tx_cmd, &events).await?; let server = tokio::spawn(ipc::serve(socket.clone(), events.clone(), tx_cmd)); @@ -327,7 +341,7 @@ async fn shutdown() { /// Subscribes to one feed, naming it from its own title. async fn add( - ctx: Ctx, + ctx: &Ctx, config_path: &std::path::Path, url: &str, folder: Option, @@ -337,7 +351,7 @@ async fn add( if let Some((id, _)) = cfg.feeds.iter().find(|(_, f)| f.url == url) { anyhow::bail!("already subscribed as {id:?}"); } - let id = add_one(&ctx, &mut cfg, url, folder, keywords).await?; + let id = add_one(ctx, &mut cfg, url, folder, keywords).await?; cfg.save(config_path)?; println!("added {id}"); Ok(()) @@ -389,7 +403,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<()> { +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() { anyhow::bail!("no feed with id {feed:?}"); @@ -400,7 +414,7 @@ fn rm(ctx: Ctx, config_path: &std::path::Path, feed: &str) -> Result<()> { Ok(()) } -async fn import(ctx: Ctx, config_path: &std::path::Path, file: &std::path::Path) -> Result<()> { +async fn import(ctx: &Ctx, config_path: &std::path::Path, file: &std::path::Path) -> Result<()> { let mut cfg = (*ctx.cfg()).clone(); let text = std::fs::read_to_string(file) .with_context(|| format!("reading {}", file.display()))?; @@ -510,7 +524,7 @@ fn reap(ctx: &Ctx, dry_run: bool, standalone: bool) -> Result<()> { Ok(()) } -async fn fetch(ctx: &Ctx, only: Option<&str>, force: bool) -> Result<()> { +async fn fetch(ctx: &Arc, only: Option<&str>, force: bool) -> Result<()> { let cfg = ctx.cfg(); if let Some(id) = only && !cfg.feeds.contains_key(id) @@ -586,7 +600,7 @@ struct Scan { /// Ok(None) means 304. async fn scan_one( - ctx: &Ctx, + ctx: &Arc, id: &str, feed_cfg: &config::Feed, state: &db::HttpState, @@ -652,6 +666,13 @@ async fn scan_one( scan.torrents += 1; continue; } + if ctx.detach_torrents { + // 'downloading' keeps the next scan from queueing it a second time. + ctx.db.mark_enclosure(&item.url, "downloading", None)?; + 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)?; @@ -729,7 +750,7 @@ fn reject(feed_cfg: &config::Feed, entry: &feed::Entry, url: &str) -> Option<&'s } async fn fetch_one( - ctx: &Ctx, + ctx: &Arc, feed_id: &str, enclosure: i64, feed_cfg: &config::Feed, @@ -774,9 +795,43 @@ async fn fetch_one( Ok((path, got.bytes)) } +/// Runs a torrent off the command worker, so feed scans and HTTP downloads keep moving +/// while it fetches metadata, transfers and then seeds. +fn spawn_torrent(ctx: &Arc, feed_id: String, enclosure: i64, url: String, dest_dir: PathBuf) { + let ctx = ctx.clone(); + tokio::spawn(async move { + // Held for the whole job, so a feed full of torrents cannot open hundreds at once. + let _permit = match ctx.torrent_slots.clone().acquire_owned().await { + Ok(p) => p, + Err(_) => return, + }; + let outcome = torrent_one(&ctx, &feed_id, enclosure, &url, &dest_dir).await; + let db = &ctx.db; + match outcome { + Ok((path, bytes)) => { + if let Err(e) = db.mark_downloaded(&url, &path, bytes) { + tracing::warn!(error = ?e, "could not record the finished torrent"); + } + ctx.out.emit(Event::DownloadDone { + feed: feed_id, + enclosure, + url, + path: path.display().to_string(), + bytes, + }); + } + Err(e) => { + let msg = format!("{e:#}"); + let _ = db.mark_enclosure(&url, "error", Some(&msg)); + ctx.out.emit(Event::DownloadError { feed: feed_id, enclosure, url, msg }); + } + } + }); +} + /// Downloads one specific enclosure immediately, whatever the per-scan cap says and /// wherever it sits in the queue. -async fn download_one(ctx: &Ctx, id: i64) -> Result<()> { +async fn download_one(ctx: &Arc, id: i64) -> Result<()> { let cfg = ctx.cfg(); let enc = ctx .db @@ -796,6 +851,11 @@ async fn download_one(ctx: &Ctx, 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)?; + spawn_torrent(ctx, enc.feed_id.clone(), enc.id, enc.url.clone(), dest_dir); + return Ok(()); + } let result = if is_torrent { if !cfg.torrent.enabled { Err(anyhow::anyhow!("torrents are disabled")) @@ -836,7 +896,7 @@ async fn download_one(ctx: &Ctx, id: i64) -> Result<()> { /// 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, + ctx: &Arc, feed_id: &str, enclosure: i64, url: &str,