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

Run torrents off the command worker

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AdXho5tTkjFLeUXKbEjKBh
This commit is contained in:
Ray Slakinski
2026-09-10 02:51:21 +00:00
parent e97f2b9c2f
commit 19309d609f
2 changed files with 115 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
src/main.rs
View File

@@ -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<torrent::Torrents>,
/// Caps how many torrents run at once when they are detached.
pub torrent_slots: Arc<tokio::sync::Semaphore>,
/// 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<Ctx>, 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<String>) -> Result<()> {
async fn daemon(
ctx: Arc<Ctx>,
config_path: PathBuf,
web_addr: Option<String>,
events: broadcast::Sender<Event>,
) -> 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::<Cmd>(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<String>,
@@ -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<Ctx>, 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<Ctx>,
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<Ctx>,
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<Ctx>, 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<Ctx>, 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<Ctx>,
feed_id: &str,
enclosure: i64,
url: &str,