diff --git a/.gitignore b/.gitignore index 6f4273a..22b7c49 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,8 @@ skills-lock.json # Local agent and planning notes AGENT.md AGENTS.md +TORRENT_FEATURES.md +torrent_features.md CLAUDE.md GEMINI.md implementation_plan.md diff --git a/TORRENT_FEATURES.md b/TORRENT_FEATURES.md deleted file mode 100644 index c977946..0000000 --- a/TORRENT_FEATURES.md +++ /dev/null @@ -1,230 +0,0 @@ -# Firelink Torrent feature matrix - -This document is the source of truth for Firelink's BitTorrent scope, current -implementation status, and next work. It compares Firelink with the -BitTorrent-specific surface of the bundled Aria2 1.37.0 engine. Aria2's -generic HTTP/FTP/SFTP/Metalink options, arbitrary shell hooks, and daemon -administration RPCs are intentionally separate unless they affect Torrent -ownership or safety. - -Reference: [Aria2 1.37.0 manual](https://aria2.github.io/manual/en/html/aria2c.html). - -## Audit basis - -- Audited on 2026-08-03 at Firelink `32034e9` (`main`) plus the current working - tree, with the cumulative - Torrent work reviewed from `edc76a7`. -- Source of truth: `src-tauri/src/torrent.rs`, `torrent_probe.rs`, `queue.rs`, - `lib.rs`, `settings.rs`, `download_ownership.rs`, `db.rs`, the IPC bindings, - frontend stores/components, and `scripts/smoke-torrent.js`. -- Reliability claims require a source postcondition or a test/harness - assertion. A passing local macOS check does not prove Windows/Linux native - behavior, public tracker/DHT reachability, or packaged-app behavior. -- The requested Agy and OpenCode review was bounded to the cumulative Torrent - diff and relevant paths. Their advice was used only after this source audit - and was verified against the live tree. - -## Implemented - -### Intake, metadata, and file selection - -- Local `.torrent` files, magnet links, and remote HTTP(S) `.torrent` metadata. - Remote metadata is bounded, redirect/SSRF checked, credential-free, parsed, - and cached before enqueue. -- Strict bencode parsing, sorted-key validation, size/depth bounds, UTF-8 - validation, canonical info-hash verification, safe output components, and - managed metadata retention/rekeying. -- Selected-file preview and validated `select-file` handling. Firelink derives - the Torrent output contract with Aria2 `index-out`; it does not use the - generic `out` option for Torrent files. -- Torrent metadata probing uses Aria2 `bt-metadata-only` and `bt-save-metadata` - internally, validates the returned hash, and conservatively cleans probe - directories. It is not exposed as a separate metadata-only download mode. -- Validated metadata is also stored under a canonical lowercase hexadecimal - info-hash key. Plain magnets containing only `xt` and optional `dn` reuse - that cache before probing when the cached file has no tracker, web-seed, or - other source-specific outer metadata; tracker, web-seed, source, and unknown - query parameters conservatively force a fresh probe. Cache hits are - revalidated against bencode and the exact hash, copied into the current - draft ID, and therefore remain compatible with Add-window rekeying. -- Canonical metadata writes use a same-directory temporary file and rename; - invalid entries and abandoned canonical temporary files are removed safely. - Canonical files use a separate `.info-.torrent` namespace from - draft/final IDs, and reads are bounded before parsing. Startup retention - keeps canonical files referenced by persisted Torrent records' - `torrentInfoHash`, as well as draft/final ID-keyed files. -- `addTorrent` passes validated web-seed/mirror URIs when supplied through the - existing download input. There is no separate Torrent web-seed manager. - -### Queue and lifecycle ownership - -- Torrents use the existing Firelink queue admission, global/per-queue permits, - pause/resume, cancellation, retry/GID replacement, restart recovery, and - terminal reconciliation. -- A Torrent's Aria2 GID is paired with the Firelink download ID and lifecycle - epoch. Late RPC results and stale terminal events cannot revive a removed or - newer lifecycle. -- Exactly one queue permit remains parked for the complete Aria2 lifecycle, - including seeding, and release is idempotent. -- Aria2 `getFiles` reconciliation establishes output ownership for Torrent - files. Ownership and optional unselected-file removal reservations are - canonicalized, persisted, collision-checked, and kept separate. -- Generic `addUri` explicitly sets both `follow-torrent=false` and - `follow-metalink=false`. This prevents an HTTP download from creating an - unmanaged child GID outside Firelink's queue, ownership, cancellation, retry, - and restart model. - -### Transfer, seeding, and integrity controls - -- Optional `seed-time` and/or `seed-ratio` policies, including ratio-only and - unlimited-ratio semantics; upload progress and seeding status are reflected - in the UI. -- Per-Torrent upload limit through Aria2 `max-upload-limit`, with a live, - lifecycle-fenced update path. -- Global Aria2 aggregate upload limit through - `max-overall-upload-limit`. It is persisted, validated, applied at daemon - startup, and changeable through `aria2.changeGlobalOption`; in Firelink it - primarily controls Torrent seeding traffic, and blank means Aria2's - unlimited value (`0`). -- Per-Torrent maximum peers (`bt-max-peers`) and low-speed peer expansion - threshold (`bt-request-peer-speed-limit`), including live updates. -- Optional piece-integrity verification through `check-integrity` and a safe - `bt-hash-check-seed`/`bt-seed-unverified=false` policy. Firelink does not - silently seed unverified data when the user requests verification. -- Optional `bt-stop-timeout` stall policy, persisted per Torrent and reapplied - on start/retry. -- Optional `bt-prioritize-piece` head/tail preview policy, normalized and - reapplied on start/retry. -- Validated encryption policies mapped consistently to - `bt-force-encryption`, `bt-require-crypto`, and `bt-min-crypto-level`. -- Optional `bt-remove-unselected-file` cleanup after successful completion, - only with an explicit partial selection and confirmation. The selected-file - ownership and unselected-file reservation are committed atomically; Firelink - clears the reservation only after Aria2's reserved paths are absent, - including when a transfer fails. Restart recovery preserves queued/paused - and orphaned reservations while reclaiming only observed failed or completed - cleanup. Disabling the option after a detach clears the reservation before - the edited item is persisted. - -### Trackers, peers, and network identity - -- Additional `bt-tracker` URLs and `bt-exclude-tracker`, including the explicit - `*` wildcard. URLs are bounded, normalized, credential-free, and limited to - HTTP(S)/UDP schemes. -- `bt-tracker-connect-timeout`, `bt-tracker-timeout`, and - `bt-tracker-interval`, persisted per Torrent and reapplied on start/retry. -- Bounded read-only `aria2.getPeers` diagnostics. Firelink discards peer IPs, - ports, IDs, and bitfields at the native boundary and exposes only bounded - operational speeds and choking/seeder flags. -- Global DHT, IPv6 DHT, PEX, and LPD toggles. Private-Torrent behavior remains - Aria2-controlled. -- Launch-scoped TCP/UDP listen-port ranges, external BitTorrent IP, IPv4/IPv6 - DHT entry points, IPv6 DHT listen address, and LPD interface. Settings are - validated, persisted, and applied only after Firelink restart. -- Optional bounded peer-ID prefix and peer-agent overrides. They are disabled - by default and carry identity/privacy/compatibility warnings. -- Global `bt-max-open-files`, bounded to 1–4096, applied at startup and - updateable for newly added Torrents through `aria2.changeGlobalOption`. - -### Evidence already present in the tree - -- Rust unit coverage for bencode/hash/path validation, option normalization, - queue ownership, lifecycle fencing, persistence sanitization, atomic Torrent - removal reservations, conservative restart recovery, host case-insensitive - path identity, and native startup argument construction. -- `src-tauri/tests/torrent_rpc.rs` covers the production authenticated JSON-RPC - HTTP boundary in a Windows-compatible integration-test target. -- `npm run smoke:torrent` and `npm run smoke:torrent:failure-paths` cover - deterministic local seeding, magnet metadata resolution, selected output, - pause/resume, ownership, cancellation/removal, unavailable trackers, - daemon failure, integrity, encryption, tracker/piece policies, open-file and - aggregate-upload limits, and stall-timeout behavior. - -## Aria2 comparison: available but not exposed or only partially represented - -| Aria2 capability | Firelink status | Reason / next step | -| --- | --- | --- | -| `bt-load-saved-metadata` | App-equivalent implemented | Firelink owns a validated, atomic, info-hash-keyed metadata cache for plain magnets, limited to metadata without source-specific outer tracker/web-seed fields, while preserving the current draft-ID/rekey contract. Source-specific magnet parameters intentionally bypass reuse; Aria2's daemon option is not exposed directly. | -| `dht-message-timeout` | Not exposed | Global DHT/UDP timeout tuning is not yet represented in settings. Add only with bounded validation and a runtime/startup contract. | -| `dht-file-path`, `dht-file-path6` | Not explicitly controlled | Aria2 can persist DHT routing tables, but Firelink does not choose app-managed paths or report their health. Decide whether portable-mode and privacy behavior justify exposing this. | -| `bt-detach-seed-only` | Not used | Aria2's concurrent-download accounting does not replace Firelink's permit ownership. Enabling it blindly would create two competing concurrency models. Revisit only with an explicit seed-slot policy. | -| `follow-torrent=true/mem` | Intentionally disabled for generic URLs | The child GID has no durable Firelink identity, permit, output ownership, or restart recovery record. Implement only after a parent/child lifecycle model exists and remote metadata validation is preserved. | -| `bt-metadata-only` / `bt-save-metadata` as user actions | Internal probe only | The Add window resolves metadata before enqueue; a separate user-visible metadata-only job is not currently a product need. | -| `on-bt-download-complete` and other hooks | Out of scope | Aria2 executes arbitrary commands. Firelink does not expose a shell-command injection surface; any future automation should be a bounded, app-owned event system. | -| `bt-enable-hook-after-hash-check` | Out of scope with hooks | It has no useful standalone meaning while arbitrary hooks are excluded. | -| `rpc-save-upload-metadata`, `save-session`, and other daemon-admin RPC policy | Out of scope / replaced | Firelink owns metadata retention and durable download state; enabling Aria2's uploaded-metadata persistence would create a second storage contract. | -| Aria2 CLI-only `show-files` / `torrent-file` controls | Product-equivalent path exists | Firelink provides a validated Add-window preview and managed `addTorrent` path rather than exposing CLI flags. | - -The comparison intentionally does not treat Aria2 defaults as Firelink -features. For example, Aria2 defaults `follow-torrent` to true, but Firelink -must override it to false on every generic `addUri` path until child ownership -is durable. - -## Priority tiers for future work - -### Tier 0 — correctness and safety gates - -No unstarted Tier 0 feature is approved. The global aggregate upload ceiling -was the highest-impact missing control and is now implemented as a persisted, -startup, and live-RPC contract. - -Before any new Torrent feature is promoted, keep these gates mandatory: - -1. Every Aria2 GID must remain attached to one Firelink identity, lifecycle - epoch, permit, and owned-path contract. -2. Every awaited RPC must re-check lifecycle ownership before mutating UI, - persistence, or queue state. -3. Any cleanup that can delete files must prove ownership and remain - conservative after cancellation, daemon loss, restart, and missed events. -4. Generic followed child GIDs remain disabled until their full lifecycle is - modeled and tested. - -### Tier 1 — high-value user behavior - -1. **DHT routing-table persistence policy.** Decide and implement app-managed - `dht-file-path`/`dht-file-path6` behavior, especially for portable mode, - permissions, reset, and privacy. This should be opt-in if it expands data - retention beyond the current download metadata contract. - -### Tier 2 — advanced tuning and ownership expansion - -1. Expose bounded `dht-message-timeout` if real tracker/DHT diagnostics show a - user-visible need; validate it at startup and document that it affects DHT - and UDP tracker waits, not HTTP metadata fetches. -2. Add an explicit seed-slot policy only if Firelink wants seeding to stop - consuming a queue permit. Aria2 `bt-detach-seed-only` alone is insufficient; - Firelink's queue and power-management semantics must agree first. -3. Model generic followed Torrent children (`true` or `mem`) with durable - parent/child IDs, admission accounting, output ownership, cancellation, - retry/GID replacement, restart discovery, and bounded metadata validation. - This remains a substantial architecture change, not a one-line option. - -## Deliberately not planned - -- Arbitrary shell hooks from Aria2. -- Direct daemon-admin/session-management controls that duplicate Firelink's - persistence and ownership system. -- Claims of public tracker/DHT readiness from local deterministic fixtures. -- A second Torrent engine. Firelink's existing Aria2 queue, permit, GID, and - recovery contracts are the intended transfer architecture. - -## Validation commands - -Run focused checks first, then the relevant broader gates: - -```sh -npm test -- --run -npm run check:i18n -npm run bindings -cd src-tauri -cargo test --test torrent_rpc -- --nocapture -cargo test --all-targets -cd .. -npm run smoke:torrent -npm run smoke:torrent:failure-paths -git diff --check -``` - -Native Windows/Linux behavior, packaged-app startup, public magnets, and -router/firewall port forwarding remain separate evidence slices and must not be -implied by these local checks. diff --git a/scripts/smoke-torrent.js b/scripts/smoke-torrent.js index 4ed8daa..6e5db3d 100644 --- a/scripts/smoke-torrent.js +++ b/scripts/smoke-torrent.js @@ -632,7 +632,37 @@ async function main() { const cancelDir = path.join(tempRoot, 'cancel'); const removeUnselectedDir = path.join(tempRoot, 'remove-unselected'); const stallDir = path.join(tempRoot, 'stall'); - for (const directory of [seedRoot, probeDir, finalDir, integrityDir, encryptionDir, cancelDir, removeUnselectedDir, stallDir]) fs.mkdirSync(directory, { recursive: true }); + const dhtStateDir = path.join(tempRoot, 'aria2-state'); + for (const directory of [seedRoot, probeDir, finalDir, integrityDir, encryptionDir, cancelDir, removeUnselectedDir, stallDir, dhtStateDir]) fs.mkdirSync(directory, { recursive: true }); + + const dhtPath = path.join(dhtStateDir, 'dht.dat'); + const dht6Path = path.join(dhtStateDir, 'dht6.dat'); + const dhtFirst = await startDaemon({ + name: 'dht-paths-first', + rpcPort: await findAvailablePort(), + listenPort: await findAvailablePort(), + directory: dhtStateDir, + extraArgs: [`--dht-file-path=${dhtPath}`, `--dht-file-path6=${dht6Path}`, '--enable-dht=true', '--enable-dht6=true', '--dht-message-timeout=42'], + }); + const firstDhtOptions = await rpc(dhtFirst.rpcPort, dhtFirst.secret, 'aria2.getGlobalOption'); + assert(firstDhtOptions['dht-file-path'] === dhtPath, `first daemon did not retain dht-file-path: ${JSON.stringify(firstDhtOptions['dht-file-path'])}`); + assert(firstDhtOptions['dht-file-path6'] === dht6Path, `first daemon did not retain dht-file-path6: ${JSON.stringify(firstDhtOptions['dht-file-path6'])}`); + assert(firstDhtOptions['dht-message-timeout'] === '42', `first daemon did not retain dht-message-timeout: ${JSON.stringify(firstDhtOptions['dht-message-timeout'])}`); + await stopDaemon(dhtFirst); + + const dhtSecond = await startDaemon({ + name: 'dht-paths-second', + rpcPort: await findAvailablePort(), + listenPort: await findAvailablePort(), + directory: dhtStateDir, + extraArgs: [`--dht-file-path=${dhtPath}`, `--dht-file-path6=${dht6Path}`, '--enable-dht=true', '--enable-dht6=true', '--dht-message-timeout=42'], + }); + const secondDhtOptions = await rpc(dhtSecond.rpcPort, dhtSecond.secret, 'aria2.getGlobalOption'); + assert(secondDhtOptions['dht-file-path'] === dhtPath, `second daemon did not retain dht-file-path: ${JSON.stringify(secondDhtOptions['dht-file-path'])}`); + assert(secondDhtOptions['dht-file-path6'] === dht6Path, `second daemon did not retain dht-file-path6: ${JSON.stringify(secondDhtOptions['dht-file-path6'])}`); + assert(secondDhtOptions['dht-message-timeout'] === '42', `second daemon did not retain dht-message-timeout: ${JSON.stringify(secondDhtOptions['dht-message-timeout'])}`); + await stopDaemon(dhtSecond); + console.log('[OK] Aria2 DHT routing-table paths and message timeout remained explicit across two launches'); const seederListenPort = await findAvailablePort(); const clientListenPort = await findAvailablePort(); diff --git a/src-tauri/src/ipc.rs b/src-tauri/src/ipc.rs index 0fff014..0caf164 100644 --- a/src-tauri/src/ipc.rs +++ b/src-tauri/src/ipc.rs @@ -34,6 +34,18 @@ fn default_torrent_max_open_files() -> u32 { crate::queue::DEFAULT_TORRENT_MAX_OPEN_FILES } +fn default_torrent_dht_message_timeout() -> u32 { + crate::queue::DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT +} + +fn default_torrent_separate_seed_slots() -> bool { + false +} + +fn default_torrent_max_concurrent_seeds() -> u32 { + crate::queue::DEFAULT_TORRENT_MAX_CONCURRENT_SEEDS +} + #[derive(Clone, Copy, Debug, Serialize, Deserialize, TS)] #[serde(rename_all = "lowercase")] #[ts(export, export_to = "../../src/bindings/")] @@ -49,6 +61,10 @@ pub enum DownloadStatus { /// A BitTorrent download has all selected data and is still seeding. /// The Aria2 GID and queue permit remain live until seeding ends. Seeding, + /// A BitTorrent download is complete but paused while waiting for a + /// Firelink-owned seeding slot. + #[serde(rename = "waitingToSeed")] + WaitingToSeed, Paused, Completed, Failed, @@ -66,6 +82,7 @@ impl DownloadStatus { Self::Downloading => "downloading", Self::Processing => "processing", Self::Seeding => "seeding", + Self::WaitingToSeed => "waitingToSeed", Self::Paused => "paused", Self::Completed => "completed", Self::Failed => "failed", @@ -188,6 +205,12 @@ pub struct DownloadItem { pub torrent_seed_ratio: Option, #[serde(default)] #[ts(optional)] + pub torrent_seed_remaining: Option, + #[serde(default)] + #[ts(optional)] + pub torrent_web_seeds: Option>, + #[serde(default)] + #[ts(optional)] pub torrent_upload_limit: Option, #[serde(default)] #[ts(optional)] @@ -252,6 +275,48 @@ pub struct TorrentPeerDiagnostics { pub truncated: bool, } +#[derive(Clone, Debug, Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export, export_to = "../../src/bindings/")] +pub struct TorrentFileProgress { + pub index: u32, + pub relative_path: String, + #[ts(type = "number")] + pub length: u64, + #[ts(type = "number")] + pub completed_length: u64, + pub selected: bool, +} + +#[derive(Clone, Debug, Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export, export_to = "../../src/bindings/")] +pub struct TorrentFileProgressSnapshot { + pub files: Vec, +} + +#[derive(Clone, Debug, Serialize, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export, export_to = "../../src/bindings/")] +pub struct TorrentPieceProgressSnapshot { + #[ts(type = "number")] + pub piece_length: u64, + #[ts(type = "number")] + pub num_pieces: u64, + #[ts(type = "number")] + pub completed_pieces: u64, + pub buckets: Vec, +} + +#[derive(Clone, Debug, Serialize, Deserialize, TS, PartialEq, Eq)] +#[serde(rename_all = "camelCase")] +#[ts(export, export_to = "../../src/bindings/")] +pub struct TorrentWebSeed { + #[ts(type = "number")] + pub file_index: u32, + pub uri: String, +} + #[derive(Clone, Debug, Serialize, Deserialize, TS)] #[serde(rename_all = "camelCase")] #[ts(export, export_to = "../../src/bindings/")] @@ -501,6 +566,12 @@ pub struct PersistedSettings { pub torrent_enable_lpd: bool, #[serde(default = "default_torrent_max_open_files")] pub torrent_max_open_files: u32, + #[serde(default = "default_torrent_dht_message_timeout")] + pub torrent_dht_message_timeout: u32, + #[serde(default = "default_torrent_separate_seed_slots")] + pub torrent_separate_seed_slots: bool, + #[serde(default = "default_torrent_max_concurrent_seeds")] + pub torrent_max_concurrent_seeds: u32, #[serde(default)] pub torrent_listen_port: String, #[serde(default)] @@ -559,6 +630,8 @@ pub struct DownloadStateEvent { pub error: Option, #[ts(optional)] pub file_name: Option, + #[ts(optional)] + pub torrent_seed_remaining: Option, } impl DownloadStateEvent { @@ -568,6 +641,7 @@ impl DownloadStateEvent { status: status.as_str().to_string(), error: None, file_name: None, + torrent_seed_remaining: None, } } @@ -577,6 +651,7 @@ impl DownloadStateEvent { status: DownloadStatus::Failed.as_str().to_string(), error: Some(error.into()), file_name: None, + torrent_seed_remaining: None, } } @@ -586,6 +661,17 @@ impl DownloadStateEvent { status: DownloadStatus::Paused.as_str().to_string(), error: Some(error.into()), file_name: None, + torrent_seed_remaining: None, + } + } + + pub fn paused_with_seed_remaining(id: impl Into, remaining: Option) -> Self { + Self { + id: id.into(), + status: DownloadStatus::Paused.as_str().to_string(), + error: None, + file_name: None, + torrent_seed_remaining: remaining, } } @@ -595,6 +681,7 @@ impl DownloadStateEvent { status: DownloadStatus::Completed.as_str().to_string(), error: None, file_name: Some(file_name.into()), + torrent_seed_remaining: None, } } @@ -606,6 +693,17 @@ impl DownloadStateEvent { status: DownloadStatus::Retrying.as_str().to_string(), error: Some(reason.into()), file_name: None, + torrent_seed_remaining: None, + } + } + + pub fn waiting_to_seed(id: impl Into, remaining: Option) -> Self { + Self { + id: id.into(), + status: DownloadStatus::WaitingToSeed.as_str().to_string(), + error: None, + file_name: None, + torrent_seed_remaining: remaining, } } } diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 1423887..b9348b2 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -4633,11 +4633,17 @@ async fn pause_download( } } + let seed_remaining = if state.queue_manager.is_seed_owner(&id) { + state.queue_manager.capture_seed_remaining(&id).await + } else { + None + }; + state.queue_manager.release_seed_tracking(&id); state.queue_manager.release_permit(&id).await; use tauri::Emitter; let _ = app_handle.emit( "download-state", - crate::ipc::DownloadStateEvent::new(id, crate::ipc::DownloadStatus::Paused), + crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining), ); return Ok(()); } @@ -4677,6 +4683,12 @@ async fn pause_download( .await; } + let seed_remaining = if state.queue_manager.is_seed_owner(&id) { + state.queue_manager.capture_seed_remaining(&id).await + } else { + None + }; + state.queue_manager.release_seed_tracking(&id); state.queue_manager.release_permit(&id).await; if registered_lifecycle_generation.is_some() || removed_pending @@ -4694,7 +4706,7 @@ async fn pause_download( use tauri::Emitter; let _ = app_handle.emit( "download-state", - crate::ipc::DownloadStateEvent::new(id, crate::ipc::DownloadStatus::Paused), + crate::ipc::DownloadStateEvent::paused_with_seed_remaining(id, seed_remaining), ); Ok(()) } @@ -4723,6 +4735,14 @@ async fn resume_download( state.queue_manager.release_registered_id(&id).await; return Ok(false); }; + if state.queue_manager.is_waiting_to_seed(&id) { + // WaitingToSeed is an intentional paused GID with no download + // permit. A resume request only wakes the Firelink seed scheduler; + // it must not bypass the seed-slot admission gate. + state.queue_manager.wake_seed_waiters(); + drop(control_guard); + return Ok(true); + } let status = aria2_download_status( state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), &state.aria2_secret, @@ -6489,6 +6509,297 @@ async fn get_torrent_peers( state.queue_manager.get_aria2_torrent_peers(&id).await } +#[tauri::command] +async fn get_torrent_file_progress( + state: tauri::State<'_, AppState>, + id: String, +) -> Result { + state + .queue_manager + .get_aria2_torrent_file_progress(&id) + .await +} + +#[tauri::command] +async fn get_torrent_piece_progress( + state: tauri::State<'_, AppState>, + id: String, +) -> Result { + state + .queue_manager + .get_aria2_torrent_piece_progress(&id) + .await +} + +fn replace_persisted_torrent_web_seeds( + database: &crate::db::DbState, + id: &str, + seeds: &[crate::ipc::TorrentWebSeed], +) -> Result, String> { + let mut connection = database.lock()?; + let records = crate::db::load_downloads(&connection)?; + let next_seeds = serde_json::to_value(seeds) + .map_err(|error| format!("failed to encode Torrent web seeds: {error}"))?; + let mut previous_seeds = None; + let mut changed = false; + let mut next = Vec::with_capacity(records.len()); + for record in records { + let mut value: serde_json::Value = match serde_json::from_str(&record) { + Ok(value) => value, + Err(_) => { + // Preserve unrelated legacy/corrupt rows byte-for-byte. A + // web-seed update must not fail its own transaction merely + // because another download cannot be decoded. + next.push(record); + continue; + } + }; + if value + .get("id") + .and_then(serde_json::Value::as_str) + == Some(id) + { + let object = value + .as_object_mut() + .ok_or_else(|| "persisted download is not an object".to_string())?; + previous_seeds = object.get("torrentWebSeeds").cloned(); + object.insert("torrentWebSeeds".to_string(), next_seeds.clone()); + changed = true; + } + next.push( + serde_json::to_string(&value) + .map_err(|error| format!("failed to encode persisted download: {error}"))?, + ); + } + if !changed { + return Err("download is not persisted".to_string()); + } + let next_data = serde_json::to_string(&next) + .map_err(|error| format!("failed to encode persisted downloads: {error}"))?; + crate::db::replace_downloads(&mut connection, &next_data, database.is_portable())?; + Ok(previous_seeds) +} + +fn restore_persisted_torrent_web_seeds( + database: &crate::db::DbState, + id: &str, + expected_seeds: &[crate::ipc::TorrentWebSeed], + previous_seeds: Option, +) -> Result<(), String> { + let mut connection = database.lock()?; + let records = crate::db::load_downloads(&connection)?; + let expected_value = serde_json::to_value(expected_seeds) + .map_err(|error| format!("failed to encode expected Torrent web seeds: {error}"))?; + let mut found = false; + let mut changed = false; + let mut next = Vec::with_capacity(records.len()); + for record in records { + let mut value: serde_json::Value = match serde_json::from_str(&record) { + Ok(value) => value, + Err(_) => { + next.push(record); + continue; + } + }; + if value + .get("id") + .and_then(serde_json::Value::as_str) + == Some(id) + { + found = true; + let object = value + .as_object_mut() + .ok_or_else(|| "persisted download is not an object".to_string())?; + if object.get("torrentWebSeeds") == Some(&expected_value) { + match previous_seeds.clone() { + Some(previous) => { + object.insert("torrentWebSeeds".to_string(), previous); + } + None => { + object.remove("torrentWebSeeds"); + } + } + changed = true; + } else { + log::warn!( + "Torrent web-seed rollback [{}] skipped because persisted state changed concurrently", + id + ); + } + } + next.push( + serde_json::to_string(&value) + .map_err(|error| format!("failed to encode persisted download: {error}"))?, + ); + } + if !found { + return Err("download is no longer persisted".to_string()); + } + if changed { + let next_data = serde_json::to_string(&next) + .map_err(|error| format!("failed to encode persisted downloads: {error}"))?; + crate::db::replace_downloads(&mut connection, &next_data, database.is_portable())?; + } + Ok(()) +} + +async fn normalize_persisted_torrent_web_seeds( + database: &crate::db::DbState, + app_handle: &tauri::AppHandle, + id: &str, + seeds: &[crate::ipc::TorrentWebSeed], +) -> Result, String> { + let record = { + let connection = database.lock()?; + crate::db::load_downloads(&connection)? + .into_iter() + .find_map(|record| { + serde_json::from_str::(&record) + .ok() + .filter(|item| item.id == id) + }) + .ok_or_else(|| "download is not persisted".to_string())? + }; + let path = record + .torrent_path + .as_deref() + .ok_or_else(|| "Torrent metadata is unavailable for web-seed management".to_string())?; + let path = crate::torrent::validate_managed_torrent_path(app_handle, id, path)?; + let bytes = tokio::fs::read(path) + .await + .map_err(|error| format!("could not read cached Torrent metadata: {error}"))?; + let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; + crate::queue::normalize_torrent_web_seeds(Some(seeds), &metadata.files) +} + +#[tauri::command] +async fn get_torrent_web_seeds( + database: tauri::State<'_, crate::db::DbState>, + state: tauri::State<'_, AppState>, + id: String, +) -> Result, String> { + if state.queue_manager.is_registered(&id).await + && matches!(state.queue_manager.active_kind(&id).await, Some(crate::queue::TaskKind::Aria2)) + { + return state.queue_manager.get_aria2_torrent_web_seeds(&id).await; + } + let persisted_seeds = { + let connection = database.lock()?; + crate::db::load_downloads(&connection)? + .into_iter() + .find_map(|record| { + serde_json::from_str::(&record) + .ok() + .filter(|item| item.id == id) + .map(|item| item.torrent_web_seeds) + }) + .ok_or_else(|| "download is not persisted".to_string())? + }; + let Some(seeds) = persisted_seeds else { + return Ok(Vec::new()); + }; + if seeds.is_empty() { + return Ok(seeds); + } + normalize_persisted_torrent_web_seeds( + database.inner(), + &state.queue_manager.app_handle(), + &id, + &seeds, + ) + .await +} + +#[tauri::command] +async fn set_torrent_web_seeds( + database: tauri::State<'_, crate::db::DbState>, + state: tauri::State<'_, AppState>, + id: String, + seeds: Vec, +) -> Result, String> { + let active = state.queue_manager.is_registered(&id).await + && matches!(state.queue_manager.active_kind(&id).await, Some(crate::queue::TaskKind::Aria2)); + let normalized = if active { + state + .queue_manager + .normalize_aria2_torrent_web_seeds(&id, &seeds) + .await? + } else { + normalize_persisted_torrent_web_seeds( + database.inner(), + &state.queue_manager.app_handle(), + &id, + &seeds, + ) + .await? + }; + let previous_seeds = + replace_persisted_torrent_web_seeds(database.inner(), &id, &normalized)?; + // The download can cross the queued/active boundary while metadata is + // being normalized and persistence is updated. Recheck before returning + // so a newly active Torrent receives the live Aria2 change instead of + // waiting for a restart to apply its persisted value. + let active_now = state.queue_manager.is_registered(&id).await + && matches!(state.queue_manager.active_kind(&id).await, Some(crate::queue::TaskKind::Aria2)); + if !active_now { + return Ok(normalized); + } + match state + .queue_manager + .set_aria2_torrent_web_seeds(&id, normalized.clone()) + .await + { + Ok((result, previous_live_seeds)) => { + if let Err(persist_error) = + replace_persisted_torrent_web_seeds(database.inner(), &id, &result) + { + // The live operation succeeded, but the durable value must + // remain transactional. Try to restore the exact value that + // was persisted before this command before reporting the + // persistence failure to the caller. + if let Err(rollback_error) = state + .queue_manager + .set_aria2_torrent_web_seeds(&id, previous_live_seeds) + .await + { + log::error!( + "Torrent web-seed live rollback [{}] failed after persistence error: {}", + id, + rollback_error + ); + } + if let Err(restore_error) = restore_persisted_torrent_web_seeds( + database.inner(), + &id, + &normalized, + previous_seeds.clone(), + ) { + log::error!( + "Torrent web-seed persistence rollback [{}] failed: {}", + id, + restore_error + ); + } + return Err(format!( + "Torrent web seeds changed live but could not be persisted: {persist_error}" + )); + } + Ok(result) + } + Err(error) => { + if let Err(restore_error) = restore_persisted_torrent_web_seeds( + database.inner(), + &id, + &normalized, + previous_seeds, + ) { + log::error!("Torrent web-seed rollback [{}] failed: {}", id, restore_error); + } + Err(error) + } + } +} + pub(crate) fn normalize_speed_limit_for_aria2(limit: &str) -> Option { let trimmed = limit.trim(); if trimmed.is_empty() { @@ -6568,6 +6879,25 @@ fn apply_aria2_torrent_network_options( } } +fn apply_aria2_torrent_dht_paths( + command: &mut std::process::Command, + dht_path: &std::path::Path, + dht6_path: &std::path::Path, +) { + command + .arg(format!("--dht-file-path={}", dht_path.display())) + .arg(format!("--dht-file-path6={}", dht6_path.display())); +} + +fn apply_aria2_torrent_dht_options( + command: &mut std::process::Command, + message_timeout: u32, +) { + let timeout = queue::normalize_torrent_dht_message_timeout(message_timeout) + .unwrap_or(queue::DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT); + command.arg(format!("--dht-message-timeout={timeout}")); +} + fn apply_aria2_torrent_peer_identity_options( command: &mut std::process::Command, peer_id_prefix: &str, @@ -7255,8 +7585,12 @@ fn db_save_settings( let prevent_system_sleep = decoded.prevents_sleep_while_downloading; let prevent_display_sleep = decoded.prevents_display_sleep_while_downloading; if let Ok(mut cached) = app_state.scheduler_settings.write() { - *cached = Some(decoded); + *cached = Some(decoded.clone()); } + app_state.queue_manager.configure_seed_capacity( + decoded.torrent_separate_seed_slots, + decoded.torrent_max_concurrent_seeds, + ); if let Err(error) = app_state .power_manager .set_preferences(prevent_system_sleep, prevent_display_sleep) @@ -7782,6 +8116,8 @@ mod tests { apply_aria2_torrent_network_options, apply_aria2_torrent_peer_identity_options, apply_aria2_torrent_peer_discovery_options, + apply_aria2_torrent_dht_paths, + apply_aria2_torrent_dht_options, aria2_rpc_port_is_occupied, parse_firelink_deep_link, parse_ffmpeg_version, parse_media_progress_line, redact_log_line, redact_log_line_for_output, sanitize_ytdlp_config_value, @@ -7834,6 +8170,50 @@ mod tests { ); } + #[test] + fn aria2_torrent_dht_paths_are_explicit_and_owned_by_firelink() { + let root = tempfile::tempdir().unwrap(); + let dht_path = root.path().join("aria2/dht.dat"); + let dht6_path = root.path().join("aria2/dht6.dat"); + let mut command = std::process::Command::new("aria2c"); + + apply_aria2_torrent_dht_paths(&mut command, &dht_path, &dht6_path); + + assert_eq!( + command + .get_args() + .map(|arg| arg.to_string_lossy().into_owned()) + .collect::>(), + vec![ + format!("--dht-file-path={}", dht_path.display()), + format!("--dht-file-path6={}", dht6_path.display()), + ] + ); + } + + #[test] + fn aria2_torrent_dht_message_timeout_is_bounded_and_launch_scoped() { + let mut command = std::process::Command::new("aria2c"); + apply_aria2_torrent_dht_options(&mut command, 42); + assert_eq!( + command + .get_args() + .map(|arg| arg.to_string_lossy().into_owned()) + .collect::>(), + vec!["--dht-message-timeout=42"] + ); + + let mut command = std::process::Command::new("aria2c"); + apply_aria2_torrent_dht_options(&mut command, 0); + assert_eq!( + command + .get_args() + .map(|arg| arg.to_string_lossy().into_owned()) + .collect::>(), + vec!["--dht-message-timeout=10"] + ); + } + #[test] fn aria2_torrent_global_options_are_bounded_and_explicit() { let mut command = std::process::Command::new("aria2c"); @@ -10466,6 +10846,12 @@ pub fn run() { let database = crate::db::init(&storage_layout) .map_err(|error| format!("failed to initialize persistence: {error}"))?; + // Establish Firelink-owned Aria2 routing-table paths after the + // existing data-root initializer has created the selected storage + // directory, but before the daemon launcher is scheduled. A + // conflict must fail startup; silently allowing Aria2 to fall back + // to a user-global dht.dat would escape the storage boundary. + let aria2_dht_paths = storage_layout.prepare_aria2_dht_paths()?; if let Err(error) = crate::torrent::remove_orphaned_probe_dirs(app.handle()) { log::warn!("could not remove orphaned torrent probes: {error}"); } @@ -10564,6 +10950,12 @@ pub fn run() { let scheduler_settings = Arc::new(RwLock::new(persisted_settings.clone())); let queue_manager = Arc::new(queue::QueueManager::new(app.handle().clone(), max_concurrent)); + if let Some(settings) = persisted_settings.as_ref() { + queue_manager.configure_seed_capacity( + settings.torrent_separate_seed_slots, + settings.torrent_max_concurrent_seeds, + ); + } let power_manager = queue_manager.power_manager(); if let Some(settings) = persisted_settings.as_ref() { let _ = power_manager.set_preferences( @@ -10723,6 +11115,15 @@ pub fn run() { torrent_max_open_files, Some(&torrent_overall_upload_limit), ); + apply_aria2_torrent_dht_paths( + &mut cmd, + &aria2_dht_paths.0, + &aria2_dht_paths.1, + ); + apply_aria2_torrent_dht_options( + &mut cmd, + torrent_startup_settings.dht_message_timeout, + ); apply_aria2_torrent_peer_discovery_options( &mut cmd, @@ -11332,7 +11733,7 @@ pub fn run() { authorize_keychain_access, acknowledge_pairing_token_change, check_file_exists, toggle_tray_icon, set_extension_pairing_token, - get_extension_server_port, set_extension_frontend_ready, ack_extension_download, set_concurrent_limit, set_queue_concurrency_limits, set_download_speed_limit, set_torrent_upload_limit, set_torrent_peer_options, get_torrent_peers, set_torrent_max_open_files, set_torrent_overall_upload_limit, set_global_speed_limit, remove_download, get_download_primary_path, + get_extension_server_port, set_extension_frontend_ready, ack_extension_download, set_concurrent_limit, set_queue_concurrency_limits, set_download_speed_limit, set_torrent_upload_limit, set_torrent_peer_options, get_torrent_peers, get_torrent_file_progress, get_torrent_piece_progress, get_torrent_web_seeds, set_torrent_web_seeds, set_torrent_max_open_files, set_torrent_overall_upload_limit, set_global_speed_limit, remove_download, get_download_primary_path, detach_download_for_reconfigure, enqueue_download, enqueue_many, cancel_enqueue_generation, move_in_queue, move_many_in_queue, remove_from_queue, get_pending_order, commands::reveal_in_file_manager, commands::open_downloaded_file, diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index a7ca7be..576f307 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -27,9 +27,18 @@ pub const MAX_TORRENT_TRACKER_INTERVAL: u32 = 604_800; pub const DEFAULT_TORRENT_MAX_OPEN_FILES: u32 = 100; pub const MIN_TORRENT_MAX_OPEN_FILES: u32 = 1; pub const MAX_TORRENT_MAX_OPEN_FILES: u32 = 4_096; +pub const DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT: u32 = 10; +pub const MIN_TORRENT_DHT_MESSAGE_TIMEOUT: u32 = 1; +pub const MAX_TORRENT_DHT_MESSAGE_TIMEOUT: u32 = 600; +pub const DEFAULT_TORRENT_MAX_CONCURRENT_SEEDS: u32 = 2; +pub const MIN_TORRENT_MAX_CONCURRENT_SEEDS: u32 = 1; +pub const MAX_TORRENT_MAX_CONCURRENT_SEEDS: u32 = 64; pub const MAX_TORRENT_NETWORK_VALUE_LENGTH: usize = 256; pub const MAX_TORRENT_PEER_ID_PREFIX_BYTES: usize = 20; pub const MAX_TORRENT_PEER_AGENT_LENGTH: usize = 128; +pub const MAX_TORRENT_PIECES_FOR_PROGRESS: u64 = 10_000_000; +pub const MAX_TORRENT_WEB_SEEDS: usize = 256; +pub const MAX_TORRENT_WEB_SEED_URI_LENGTH: usize = 2_048; pub const MIN_TORRENT_LISTEN_PORT: u16 = 1024; pub const DEFAULT_TORRENT_LISTEN_PORT_SPEC: &str = "6881-6999"; @@ -46,6 +55,133 @@ pub fn normalize_torrent_max_open_files(value: u32) -> Result { Ok(value) } +pub fn normalize_torrent_dht_message_timeout(value: u32) -> Result { + if !(MIN_TORRENT_DHT_MESSAGE_TIMEOUT..=MAX_TORRENT_DHT_MESSAGE_TIMEOUT).contains(&value) { + return Err(format!( + "DHT message timeout must be between {MIN_TORRENT_DHT_MESSAGE_TIMEOUT} and {MAX_TORRENT_DHT_MESSAGE_TIMEOUT} seconds" + )); + } + Ok(value) +} + +fn normalize_torrent_web_seed_uri(value: &str) -> Result { + let value = value.trim(); + if value.is_empty() || value.len() > MAX_TORRENT_WEB_SEED_URI_LENGTH { + return Err(format!( + "Torrent web-seed URIs must be between 1 and {MAX_TORRENT_WEB_SEED_URI_LENGTH} bytes" + )); + } + if value.chars().any(char::is_control) { + return Err("Torrent web-seed URIs must not contain control characters".to_string()); + } + let parsed = url::Url::parse(value).map_err(|_| "Torrent web-seed URI is invalid".to_string())?; + if !matches!(parsed.scheme(), "http" | "https") { + return Err("Torrent web-seed URI must use HTTP or HTTPS".to_string()); + } + if parsed.host_str().is_none_or(str::is_empty) + || !parsed.username().is_empty() + || parsed.password().is_some() + || parsed.fragment().is_some() + { + return Err("Torrent web-seed URI must have a host and no credentials or fragment".to_string()); + } + Ok(parsed.to_string()) +} + +pub fn normalize_torrent_web_seeds( + seeds: Option<&[crate::ipc::TorrentWebSeed]>, + files: &[crate::ipc::TorrentFile], +) -> Result, String> { + let Some(seeds) = seeds else { + return Ok(Vec::new()); + }; + if seeds.len() > MAX_TORRENT_WEB_SEEDS { + return Err(format!("a Torrent may have at most {MAX_TORRENT_WEB_SEEDS} web seeds")); + } + let mut normalized = Vec::with_capacity(seeds.len()); + let mut seen = HashSet::new(); + for seed in seeds { + if !files.iter().any(|file| file.index == seed.file_index) { + return Err(format!("Torrent web-seed file index {} is out of range", seed.file_index)); + } + let uri = normalize_torrent_web_seed_uri(&seed.uri)?; + if seen.insert((seed.file_index, uri.clone())) { + normalized.push(crate::ipc::TorrentWebSeed { + file_index: seed.file_index, + uri, + }); + } + } + Ok(normalized) +} + +fn expand_torrent_web_seed_uri( + seed: &crate::ipc::TorrentWebSeed, + files: &[crate::ipc::TorrentFile], +) -> Result { + let uri = normalize_torrent_web_seed_uri(&seed.uri)?; + if files.len() <= 1 { + return Ok(uri); + } + let file = files + .iter() + .find(|file| file.index == seed.file_index) + .ok_or_else(|| "Torrent web-seed file index is out of range".to_string())?; + let mut parsed = url::Url::parse(&uri).map_err(|_| "Torrent web-seed URI is invalid".to_string())?; + let base_path = parsed.path().trim_end_matches('/').to_string(); + parsed.set_path(&base_path); + { + let mut path = parsed + .path_segments_mut() + .map_err(|_| "Torrent web-seed URI cannot accept a file path".to_string())?; + for segment in file.path.replace('\\', "/").split('/') { + if !segment.is_empty() && segment != "." && segment != ".." { + path.push(segment); + } + } + } + Ok(parsed.to_string()) +} + +pub fn expand_torrent_web_seeds( + seeds: &[crate::ipc::TorrentWebSeed], + files: &[crate::ipc::TorrentFile], +) -> Result, String> { + let normalized = normalize_torrent_web_seeds(Some(seeds), files)?; + normalized + .iter() + .map(|seed| Ok((seed.file_index, expand_torrent_web_seed_uri(seed, files)?))) + .collect() +} + +fn parse_aria2_web_seed_uris(value: &serde_json::Value) -> Result, String> { + let entries = value + .as_array() + .ok_or_else(|| "aria2.getUris returned a non-array result".to_string())?; + let mut uris = Vec::with_capacity(entries.len()); + let mut seen = HashSet::new(); + for entry in entries { + let uri = entry + .get("uri") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| "aria2.getUris returned a malformed URI entry".to_string())?; + let uri = normalize_torrent_web_seed_uri(uri)?; + if seen.insert(uri.clone()) { + uris.push(uri); + } + } + Ok(uris) +} + +pub fn normalize_torrent_max_concurrent_seeds(value: u32) -> Result { + if !(MIN_TORRENT_MAX_CONCURRENT_SEEDS..=MAX_TORRENT_MAX_CONCURRENT_SEEDS).contains(&value) { + return Err(format!( + "maximum concurrent Torrent seeds must be between {MIN_TORRENT_MAX_CONCURRENT_SEEDS} and {MAX_TORRENT_MAX_CONCURRENT_SEEDS}" + )); + } + Ok(value) +} + fn normalize_optional_torrent_network_value( value: Option<&str>, field: &str, @@ -337,6 +473,13 @@ pub enum Aria2RefreshOutcome { Complete, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Aria2SeedControlOutcome { + Resumed, + Paused, + Complete, +} + /// Result of rebuilding an aria2 job while retaining its partial file and /// queue permit. `Refresh` is the compatibility path for test/alternate /// spawners that do not own aria2's addUri options. @@ -383,6 +526,35 @@ struct QueuePermitOwnership { active: bool, } +#[derive(Debug, Clone)] +struct SeedWaiter { + id: String, + queue_id: String, + lifecycle_generation: u64, +} + +#[derive(Debug, Default)] +struct SeedCapacityState { + enabled: bool, + max_concurrent: usize, + owners: HashSet, + waiting: VecDeque, + starting: HashSet, +} + +#[derive(Debug, Clone, Copy)] +struct SeedBudget { + remaining_minutes: Option, + started_at: std::time::Instant, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SeedAdmissionOutcome { + Seeding, + Waiting, + Complete, +} + /// Args mirroring start_download / start_media_download. Kept untyped-loose /// (String/Option) to match the existing command signatures exactly. #[derive(Debug, Clone, Default)] @@ -409,6 +581,8 @@ pub struct SpawnPayload { pub torrent_file_indices: Option>, pub torrent_seed_time: Option, pub torrent_seed_ratio: Option, + pub torrent_seed_remaining: Option, + pub torrent_web_seeds: Option>, pub torrent_upload_limit: Option, pub torrent_max_peers: Option, pub torrent_peer_speed_limit: Option, @@ -489,6 +663,31 @@ pub trait SidecarSpawner: Send + Sync + 'static { Err("live torrent peer options are unavailable".to_string()) } + /// Pause a completed Torrent before releasing its download permit to the + /// Firelink-owned seed-slot pool. + async fn pause_for_seed(&self, _gid: &str) -> Result { + Err("live Torrent seed-slot pausing is unavailable".to_string()) + } + + /// Resume a Torrent after a Firelink seed slot has been reserved. + async fn resume_for_seed(&self, _gid: &str) -> Result { + Err("live Torrent seed-slot resuming is unavailable".to_string()) + } + + async fn get_torrent_uris(&self, _gid: &str) -> Result, String> { + Err("live Torrent web-seed inspection is unavailable".to_string()) + } + + async fn change_torrent_uris( + &self, + _gid: &str, + _file_index: u32, + _delete: &[String], + _add: &[String], + ) -> Result<(), String> { + Err("live Torrent web-seed changes are unavailable".to_string()) + } + /// Run a media download to completion. The permit is parked for the full /// duration; release is handled by QueueManager on the runner's exit. async fn run_media( @@ -526,6 +725,10 @@ pub struct QueueManager { target_capacity: AtomicUsize, slots_to_retire: AtomicUsize, notify: Notify, + /// Firelink-owned seed capacity. Seeders keep the Aria2 GID and mapping + /// alive, but release the download semaphore while they own a seed slot. + seed_capacity: StdMutex, + seed_budgets: StdMutex>, /// aria2 gid -> download id map (shared with the WS poller). pub aria2_gids: Arc>>, @@ -616,6 +819,12 @@ impl QueueManager { target_capacity: AtomicUsize::new(capacity), slots_to_retire: AtomicUsize::new(0), notify: Notify::new(), + seed_capacity: StdMutex::new(SeedCapacityState { + enabled: false, + max_concurrent: DEFAULT_TORRENT_MAX_CONCURRENT_SEEDS as usize, + ..SeedCapacityState::default() + }), + seed_budgets: StdMutex::new(HashMap::new()), aria2_gids: Arc::new(std::sync::RwLock::new(HashMap::new())), pending_completion: Arc::new(Mutex::new(HashMap::new())), aria2_payloads: Mutex::new(HashMap::new()), @@ -641,6 +850,10 @@ impl QueueManager { Arc::clone(&self.power_manager) } + pub fn app_handle(&self) -> AppHandle { + self.app_handle.clone() + } + pub fn activate_power_management(&self) -> Result<(), String> { self.power_manager.activate() } @@ -658,6 +871,366 @@ impl QueueManager { self.power_manager.set_system_prevention(enabled) } + /// Apply the persisted seed policy to the live queue manager. Existing + /// seeders are not paused when the setting changes; the new limit applies + /// to the next seed admission and to waiting seeders. + pub fn configure_seed_capacity(&self, enabled: bool, max_concurrent: u32) { + let max_concurrent = normalize_torrent_max_concurrent_seeds(max_concurrent) + .unwrap_or(DEFAULT_TORRENT_MAX_CONCURRENT_SEEDS) as usize; + if let Ok(mut state) = self.seed_capacity.lock() { + state.enabled = enabled; + state.max_concurrent = max_concurrent; + if enabled { + if let Ok(budgets) = self.seed_budgets.lock() { + state.owners.extend(budgets.keys().cloned()); + } + } + } + self.notify.notify_waiters(); + } + + fn seed_capacity_enabled(&self) -> bool { + self.seed_capacity + .lock() + .map(|state| state.enabled) + .unwrap_or(false) + } + + fn seed_owner(&self, id: &str) -> bool { + self.seed_capacity + .lock() + .map(|state| state.owners.contains(id)) + .unwrap_or(false) + } + + pub fn is_seed_owner(&self, id: &str) -> bool { + self.seed_owner(id) + } + + fn seed_waiting(&self, id: &str) -> bool { + self.seed_capacity.lock().map_or(false, |state| { + state.waiting.iter().any(|waiter| waiter.id == id) + || state.starting.contains(id) + }) + } + + fn add_seed_waiter(&self, waiter: SeedWaiter) { + if let Ok(mut state) = self.seed_capacity.lock() { + if !state.owners.contains(&waiter.id) + && !state.starting.contains(&waiter.id) + && !state.waiting.iter().any(|candidate| candidate.id == waiter.id) + { + state.waiting.push_back(waiter); + } + } + self.notify.notify_waiters(); + } + + fn reserve_seed_slot(&self, id: &str) -> bool { + let Ok(mut state) = self.seed_capacity.lock() else { + return false; + }; + if state.owners.contains(id) { + return true; + } + if !state.enabled || state.owners.len() >= state.max_concurrent { + return false; + } + state.owners.insert(id.to_string()); + true + } + + fn reserve_seed_waiter(&self) -> Option { + let Ok(mut state) = self.seed_capacity.lock() else { + return None; + }; + if state.enabled && state.owners.len() >= state.max_concurrent { + return None; + } + let waiter = state.waiting.pop_front()?; + state.owners.insert(waiter.id.clone()); + state.starting.insert(waiter.id.clone()); + Some(waiter) + } + + fn finish_seed_start(&self, id: &str) { + if let Ok(mut state) = self.seed_capacity.lock() { + state.starting.remove(id); + } + } + + fn abandon_seed_start(&self, id: &str) { + if let Ok(mut state) = self.seed_capacity.lock() { + state.starting.remove(id); + state.owners.remove(id); + } + self.notify.notify_waiters(); + } + + /// Stop tracking a seed lifecycle. This is called for terminal, manual + /// pause, and remove paths; releasing a download permit while a seed slot + /// is merely changing hands deliberately does not call it. + pub fn release_seed_tracking(&self, id: &str) { + if let Ok(mut state) = self.seed_capacity.lock() { + state.owners.remove(id); + state.starting.remove(id); + state.waiting.retain(|waiter| waiter.id != id); + } + if let Ok(mut budgets) = self.seed_budgets.lock() { + budgets.remove(id); + } + self.notify.notify_waiters(); + } + + pub fn is_waiting_to_seed(&self, id: &str) -> bool { + self.seed_waiting(id) + } + + pub fn wake_seed_waiters(&self) { + self.notify.notify_waiters(); + } + + async fn record_seed_started(&self, id: &str) { + let remaining_minutes = self + .aria2_payloads + .lock() + .await + .get(id) + .and_then(|payload| { + payload + .torrent_seed_remaining + .or(payload.torrent_seed_time) + .filter(|minutes| minutes.is_finite() && *minutes > 0.0) + }); + if let Ok(mut budgets) = self.seed_budgets.lock() { + budgets.insert( + id.to_string(), + SeedBudget { + remaining_minutes, + started_at: std::time::Instant::now(), + }, + ); + } + } + + /// Persist the remaining time in the in-memory payload before a seeder is + /// parked or manually paused. The frontend persistence path receives the + /// same value through the WaitingToSeed event and can restore it after a + /// restart. + pub async fn capture_seed_remaining(&self, id: &str) -> Option { + let remaining = self.seed_budgets.lock().ok().and_then(|mut budgets| { + let budget = budgets.get_mut(id)?; + let elapsed = budget.started_at.elapsed().as_secs_f64() / 60.0; + budget.remaining_minutes.map(|minutes| (minutes - elapsed).max(0.0)) + }); + if let Some(remaining) = remaining { + if let Some(payload) = self.aria2_payloads.lock().await.get_mut(id) { + payload.torrent_seed_time = Some(remaining); + payload.torrent_seed_remaining = Some(remaining); + } + } + remaining + } + + async fn release_download_permit_for_seed(&self, id: &str) { + let _admission_gate = self.admission_gate.lock().await; + let removed = self.active_permits.lock().await.remove(id).is_some(); + self.active_permit_generations.lock().await.remove(id); + self.queue_permit_ownership.lock().await.remove(id); + if removed { + self.notify.notify_waiters(); + } + drop(_admission_gate); + if removed { + self.sync_power_activity().await; + } + } + + async fn admit_seed_after_completion(&self, id: &str) -> SeedAdmissionOutcome { + if self.seed_owner(id) { + return SeedAdmissionOutcome::Seeding; + } + if self.seed_waiting(id) { + return SeedAdmissionOutcome::Waiting; + } + if self.reserve_seed_slot(id) { + self.release_download_permit_for_seed(id).await; + self.record_seed_started(id).await; + return SeedAdmissionOutcome::Seeding; + } + + let Some(gid) = self.aria2_gid_for_download(id) else { + return SeedAdmissionOutcome::Seeding; + }; + match self.spawner.pause_for_seed(&gid).await { + Ok(Aria2SeedControlOutcome::Paused) => { + self.record_seed_started(id).await; + let remaining = self.capture_seed_remaining(id).await; + let ownership = self.queue_permit_ownership.lock().await.get(id).cloned(); + let lifecycle_generation = ownership + .as_ref() + .map(|ownership| ownership.lifecycle_generation) + .or(self.registered_lifecycle_generation(id).await) + .unwrap_or_default(); + self.release_download_permit_for_seed(id).await; + self.add_seed_waiter(SeedWaiter { + id: id.to_string(), + queue_id: ownership + .as_ref() + .map(|ownership| ownership.queue_id.clone()) + .unwrap_or_else(|| "main".to_string()), + lifecycle_generation, + }); + let _ = remaining; + SeedAdmissionOutcome::Waiting + } + Ok(Aria2SeedControlOutcome::Complete) => SeedAdmissionOutcome::Complete, + Ok(Aria2SeedControlOutcome::Resumed) | Err(_) => { + // An ambiguous pause must retain the download permit and GID; + // treating the transfer as a live seeder is the conservative + // choice until the next daemon status event resolves it. + self.reserve_seed_slot_even_if_full(id); + self.record_seed_started(id).await; + SeedAdmissionOutcome::Seeding + } + } + } + + fn reserve_seed_slot_even_if_full(&self, id: &str) { + if let Ok(mut state) = self.seed_capacity.lock() { + state.owners.insert(id.to_string()); + } + } + + fn emit_waiting_to_seed(&self, id: &str, remaining: Option) { + use tauri::Emitter; + let _ = self.app_handle.emit( + "download-state", + DownloadStateEvent::waiting_to_seed(id, remaining), + ); + } + + async fn try_start_waiting_seed(self: &Arc) -> bool { + let Some(waiter) = self.reserve_seed_waiter() else { + return false; + }; + let manager = Arc::clone(self); + tauri::async_runtime::spawn(async move { + manager.resume_waiting_seed(waiter).await; + }); + true + } + + async fn resume_waiting_seed(self: Arc, waiter: SeedWaiter) { + let id = waiter.id.clone(); + let Some(gid) = self.aria2_gid_for_download(&id) else { + self.abandon_seed_start(&id); + return; + }; + if !self + .is_registered_generation_or_legacy(&id, waiter.lifecycle_generation) + .await + || !self.seed_waiting(&id) + || !matches!(self.active_kind(&id).await, Some(TaskKind::Aria2)) + { + self.abandon_seed_start(&id); + return; + } + let control_epoch = self.current_aria2_control_epoch(&id).await; + let Some(permit_candidate) = self + .acquire_aria2_permit_candidate_for_queue( + &id, + &waiter.queue_id, + waiter.lifecycle_generation, + control_epoch, + ) + .await + else { + self.abandon_seed_start(&id); + return; + }; + let _control_guard = self.acquire_aria2_control(&id).await; + if !self + .is_registered_generation_or_legacy(&id, waiter.lifecycle_generation) + .await + || !self.seed_waiting(&id) + || !matches!(self.active_kind(&id).await, Some(TaskKind::Aria2)) + || self.aria2_gid_for_download(&id).as_deref() != Some(gid.as_str()) + || !self.is_aria2_control_epoch_current(&id, control_epoch).await + { + self.release_aria2_permit_candidate(&id, waiter.lifecycle_generation) + .await; + self.abandon_seed_start(&id); + return; + } + let epoch = self.next_aria2_control_epoch(&id).await; + if !self.rebind_aria2_gid_epoch(&id, &gid, epoch).await { + self.release_aria2_permit_candidate(&id, waiter.lifecycle_generation) + .await; + self.abandon_seed_start(&id); + return; + } + if !self + .park_aria2_permit_if_missing_for_queue( + &id, + &waiter.queue_id, + waiter.lifecycle_generation, + permit_candidate, + ) + .await + { + self.release_aria2_permit_candidate(&id, waiter.lifecycle_generation) + .await; + self.abandon_seed_start(&id); + return; + } + match self.spawner.resume_for_seed(&gid).await { + Ok(Aria2SeedControlOutcome::Resumed) => { + if !self.is_aria2_control_epoch_current(&id, epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &Aria2GidMapping { id: id.clone(), epoch }) + { + let _ = self.spawner.pause_for_seed(&gid).await; + self.release_download_permit_for_seed(&id).await; + self.abandon_seed_start(&id); + return; + } + self.finish_seed_start(&id); + self.record_seed_started(&id).await; + self.release_download_permit_for_seed(&id).await; + self.emit_state(&id, DownloadStatus::Seeding); + } + Ok(Aria2SeedControlOutcome::Complete) => { + self.finish_seed_start(&id); + self.apply_completion_locked(&id, PendingOutcome::Complete).await; + } + Ok(Aria2SeedControlOutcome::Paused) => { + let remaining = self.capture_seed_remaining(&id).await; + self.release_download_permit_for_seed(&id).await; + self.finish_seed_start(&id); + self.abandon_seed_start(&id); + self.add_seed_waiter(SeedWaiter { + id: id.clone(), + queue_id: waiter.queue_id, + lifecycle_generation: waiter.lifecycle_generation, + }); + self.emit_waiting_to_seed(&id, remaining); + } + Err(error) => { + // An unverified unpause must not release either the seed + // owner or its download permit: the daemon may already be + // active even though the RPC/status check was unavailable. + // Keep the item fenced in the starting state until a + // terminal event or an explicit user pause resolves it. + log::warn!( + "Torrent seed resume [{}] could not be verified; retaining seed ownership and permit: {}", + id, + error + ); + self.emit_waiting_to_seed(&id, None); + } + } + } + async fn sync_power_activity(&self) { let active_transfers = self.active_permits.lock().await.len(); self.power_manager @@ -683,6 +1256,7 @@ impl QueueManager { // Epoch checks remain the authoritative guard; removing this marker // prevents terminal downloads from accumulating cancellation entries. self.aria2_retry_cancelled.lock().await.remove(id); + self.release_seed_tracking(id); self.notify.notify_waiters(); } @@ -707,6 +1281,13 @@ impl QueueManager { == Some(generation) } + async fn is_registered_generation_or_legacy(&self, id: &str, generation: u64) -> bool { + self.is_registered_generation(id, generation).await + || (generation == 0 + && self.registered_lifecycle_generation(id).await.is_none() + && self.is_registered(id).await) + } + pub(crate) async fn release_registered_id_for_generation(&self, id: &str, generation: u64) { let released = { let mut registered = self.registered_ids.lock().await; @@ -721,6 +1302,7 @@ impl QueueManager { }; if released { self.aria2_retry_cancelled.lock().await.remove(id); + self.release_seed_tracking(id); self.notify.notify_waiters(); } } @@ -940,6 +1522,358 @@ impl QueueManager { .is_some_and(torrent_seeding_requested) } + async fn torrent_files_for_payload( + &self, + id: &str, + payload: &SpawnPayload, + ) -> Result, String> { + let path = payload + .torrent_path + .as_deref() + .ok_or_else(|| "Torrent metadata is unavailable for web-seed management".to_string())?; + let path = crate::torrent::validate_managed_torrent_path(&self.app_handle, id, path)?; + let bytes = tokio::fs::read(path) + .await + .map_err(|error| format!("could not read cached Torrent metadata: {error}"))?; + Ok(crate::torrent::parse_torrent_bytes(&bytes)?.files) + } + + async fn current_torrent_mapping( + &self, + id: &str, + ) -> Result<(String, Aria2GidMapping), String> { + if !self.is_registered(id).await || !matches!(self.active_kind(id).await, Some(TaskKind::Aria2)) { + return Err("download is not an active Aria2 Torrent".to_string()); + } + let gid = self + .aria2_gid_for_download(id) + .ok_or_else(|| "active Torrent transfer has no GID".to_string())?; + let mapping = self + .aria2_gid_mapping(&gid) + .ok_or_else(|| "active Torrent transfer has no current GID mapping".to_string())?; + if mapping.id != id || !self.is_aria2_control_epoch_current(id, mapping.epoch).await { + return Err("Torrent lifecycle changed before web-seed inspection".to_string()); + } + Ok((gid, mapping)) + } + + pub async fn get_aria2_torrent_web_seeds( + &self, + id: &str, + ) -> Result, String> { + let _control_guard = self.acquire_aria2_control(id).await; + let payload = self + .aria2_payloads + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| "Torrent retry payload is unavailable".to_string())?; + let files = self.torrent_files_for_payload(id, &payload).await?; + let desired = normalize_torrent_web_seeds(payload.torrent_web_seeds.as_deref(), &files)?; + let (gid, mapping) = self.current_torrent_mapping(id).await?; + let current = self.spawner.get_torrent_uris(&gid).await?; + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + return Err("Torrent lifecycle changed while reading web seeds".to_string()); + } + let expected = expand_torrent_web_seeds(&desired, &files)?; + if expected.iter().any(|(_, uri)| !current.iter().any(|candidate| candidate == uri)) { + return Err("Aria2 web-seed state differs from Firelink's persisted state".to_string()); + } + Ok(desired + .into_iter() + .filter(|seed| { + expand_torrent_web_seed_uri(seed, &files) + .ok() + .is_some_and(|uri| current.iter().any(|candidate| candidate == &uri)) + }) + .collect()) + } + + pub async fn normalize_aria2_torrent_web_seeds( + &self, + id: &str, + seeds: &[crate::ipc::TorrentWebSeed], + ) -> Result, String> { + let _control_guard = self.acquire_aria2_control(id).await; + let payload = self + .aria2_payloads + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| "Torrent retry payload is unavailable".to_string())?; + let files = self.torrent_files_for_payload(id, &payload).await?; + normalize_torrent_web_seeds(Some(seeds), &files) + } + + async fn rollback_torrent_web_seed_changes( + &self, + id: &str, + gid: &str, + mapping: &Aria2GidMapping, + changes: &[(u32, Vec, Vec)], + ) -> bool { + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(gid, mapping) + { + return false; + } + let mut restored = true; + for (file_index, delete, add) in changes.iter().rev() { + if let Err(error) = self + .spawner + .change_torrent_uris(gid, *file_index, add, delete) + .await + { + restored = false; + log::warn!( + "Torrent web-seed rollback [{}] failed for gid {} file {}: {}", + id, + gid, + file_index, + error + ); + } + } + restored + } + + async fn restore_torrent_web_seed_payload_if_current( + &self, + id: &str, + gid: &str, + mapping: &Aria2GidMapping, + old_payload: &SpawnPayload, + ) { + if self.is_aria2_control_epoch_current(id, mapping.epoch).await + && self.is_current_aria2_gid_mapping(gid, mapping) + { + if let Some(payload) = self.aria2_payloads.lock().await.get_mut(id) { + payload.torrent_web_seeds = old_payload.torrent_web_seeds.clone(); + } + } + } + + pub async fn set_aria2_torrent_web_seeds( + &self, + id: &str, + seeds: Vec, + ) -> Result< + ( + Vec, + Vec, + ), + String, + > { + let _control_guard = self.acquire_aria2_control(id).await; + let old_payload = self + .aria2_payloads + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| "Torrent retry payload is unavailable".to_string())?; + let files = self.torrent_files_for_payload(id, &old_payload).await?; + let desired = normalize_torrent_web_seeds(Some(&seeds), &files)?; + let old = normalize_torrent_web_seeds(old_payload.torrent_web_seeds.as_deref(), &files)?; + let (gid, mapping) = self.current_torrent_mapping(id).await?; + let current = self.spawner.get_torrent_uris(&gid).await?; + let old_expanded = expand_torrent_web_seeds(&old, &files)?; + let new_expanded = expand_torrent_web_seeds(&desired, &files)?; + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + return Err("Torrent lifecycle changed while reading web seeds".to_string()); + } + if old_expanded.iter().any(|(_, uri)| !current.iter().any(|candidate| candidate == uri)) { + return Err("Aria2 web-seed state differs from Firelink's persisted state".to_string()); + } + + if let Some(payload) = self.aria2_payloads.lock().await.get_mut(id) { + payload.torrent_web_seeds = Some(desired.clone()); + } + let mut changes = Vec::<(u32, Vec, Vec)>::new(); + let mut current_set = current.into_iter().collect::>(); + let mut file_indices = old_expanded.iter().map(|(index, _)| *index).collect::>(); + file_indices.extend(new_expanded.iter().map(|(index, _)| *index)); + let mut file_indices = file_indices.into_iter().collect::>(); + file_indices.sort_unstable(); + for file_index in file_indices { + let delete = old_expanded + .iter() + .filter(|(index, uri)| *index == file_index && current_set.contains(uri)) + .map(|(_, uri)| uri.clone()) + .collect::>(); + let add = new_expanded + .iter() + .filter(|(index, uri)| *index == file_index && !current_set.contains(uri)) + .map(|(_, uri)| uri.clone()) + .collect::>(); + if delete.is_empty() && add.is_empty() { + continue; + } + changes.push((file_index, delete.clone(), add.clone())); + if let Err(error) = self.spawner.change_torrent_uris(&gid, file_index, &delete, &add).await { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + self.restore_torrent_web_seed_payload_if_current( + id, + &gid, + &mapping, + &old_payload, + ) + .await; + return Err(error); + } + for uri in &delete { + current_set.remove(uri); + } + current_set.extend(add.iter().cloned()); + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + self.restore_torrent_web_seed_payload_if_current( + id, + &gid, + &mapping, + &old_payload, + ) + .await; + return Err("Torrent lifecycle changed while changing web seeds".to_string()); + } + } + let readback = match self.spawner.get_torrent_uris(&gid).await { + Ok(readback) => readback, + Err(error) => { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + self.restore_torrent_web_seed_payload_if_current( + id, + &gid, + &mapping, + &old_payload, + ) + .await; + return Err(error); + } + }; + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + self.restore_torrent_web_seed_payload_if_current( + id, + &gid, + &mapping, + &old_payload, + ) + .await; + return Err("Torrent lifecycle changed while reading back web seeds".to_string()); + } + let expected = new_expanded.into_iter().map(|(_, uri)| uri).collect::>(); + if readback.iter().cloned().collect::>() != expected { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + self.restore_torrent_web_seed_payload_if_current( + id, + &gid, + &mapping, + &old_payload, + ) + .await; + return Err("Aria2 web-seed readback did not match the requested state".to_string()); + } + Ok((desired, old)) + } + + /// Attach the persisted typed seed set after addTorrent returns its GID. + /// The addTorrent URI parameter is intentionally left unscoped; this + /// method is the only path that supplies the required Aria2 file index. + pub async fn install_initial_torrent_web_seeds(&self, id: &str) -> Result<(), String> { + let _control_guard = self.acquire_aria2_control(id).await; + let payload = self + .aria2_payloads + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| "Torrent retry payload is unavailable".to_string())?; + let desired = payload.torrent_web_seeds.clone().unwrap_or_default(); + if desired.is_empty() { + return Ok(()); + } + let files = self.torrent_files_for_payload(id, &payload).await?; + let (gid, mapping) = self.current_torrent_mapping(id).await?; + let current = self.spawner.get_torrent_uris(&gid).await?; + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + return Err("Torrent lifecycle changed while reading initial web seeds".to_string()); + } + let expanded = expand_torrent_web_seeds(&desired, &files)?; + let mut current_set = current.into_iter().collect::>(); + let mut changes = Vec::<(u32, Vec, Vec)>::new(); + for (file_index, uri) in &expanded { + if current_set.contains(uri) { + continue; + } + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + return Err("Torrent lifecycle changed while attaching web seeds".to_string()); + } + let add = vec![uri.clone()]; + changes.push((*file_index, Vec::new(), add.clone())); + if let Err(error) = self + .spawner + .change_torrent_uris(&gid, *file_index, &[], &add) + .await + { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + return Err(error); + } + current_set.insert(uri.clone()); + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + return Err("Torrent lifecycle changed while attaching web seeds".to_string()); + } + } + let readback = match self.spawner.get_torrent_uris(&gid).await { + Ok(readback) => readback, + Err(error) => { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + return Err(error); + } + }; + if !self.is_aria2_control_epoch_current(id, mapping.epoch).await + || !self.is_current_aria2_gid_mapping(&gid, &mapping) + { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + return Err("Torrent lifecycle changed while reading initial web seeds".to_string()); + } + let expected = expanded.into_iter().map(|(_, uri)| uri).collect::>(); + if readback.into_iter().collect::>() != expected { + self.rollback_torrent_web_seed_changes(&id, &gid, &mapping, &changes) + .await; + return Err("Aria2 did not retain the persisted Torrent web-seed set".to_string()); + } + Ok(()) + } + pub fn set_aria2_global_speed_limit(&self, limit: Option) { *self .aria2_global_speed_limit @@ -1236,6 +2170,152 @@ impl QueueManager { Ok(diagnostics) } + /// Return a lifecycle-fenced, metadata-derived projection of Aria2's + /// per-file progress. The daemon's absolute paths and URI lists are never + /// copied across the boundary. + pub async fn get_aria2_torrent_file_progress( + &self, + id: &str, + ) -> Result { + let _control_guard = self.acquire_aria2_control(id).await; + if !self.is_registered(id).await + || !matches!(self.active_kind(id).await, Some(TaskKind::Aria2)) + { + return Err("live Torrent file progress is unavailable".to_string()); + } + let payload = self + .aria2_payloads + .lock() + .await + .get(id) + .cloned() + .ok_or_else(|| "live Torrent file progress is unavailable".to_string())?; + if !payload.is_torrent { + return Err("download is not a Torrent transfer".to_string()); + } + let torrent_path = payload + .torrent_path + .as_deref() + .ok_or_else(|| "live Torrent file progress is unavailable".to_string())?; + let torrent_path = crate::torrent::validate_managed_torrent_path( + &self.app_handle, + id, + torrent_path, + )?; + let bytes = tokio::fs::read(&torrent_path) + .await + .map_err(|_| "live Torrent file progress is unavailable".to_string())?; + let metadata = crate::torrent::parse_torrent_bytes(&bytes)?; + let gid = self + .aria2_gid_for_download(id) + .ok_or_else(|| "live Torrent file progress is unavailable".to_string())?; + let expected_mapping = self + .aria2_gid_mapping(&gid) + .ok_or_else(|| "live Torrent file progress is unavailable".to_string())?; + if expected_mapping.id != id + || !self + .is_aria2_control_epoch_current(id, expected_mapping.epoch) + .await + { + return Err("live Torrent file progress is unavailable".to_string()); + } + + let state = self.app_handle.state::(); + let result = crate::rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.getFiles", + serde_json::json!([gid]), + ) + .await + .map_err(|error| { + format!( + "aria2.getFiles failed: {}", + crate::redact_sensitive_text(&error) + ) + })?; + let snapshot = parse_torrent_file_progress(result, &metadata.files)?; + + let still_current = self.is_registered(id).await + && matches!(self.active_kind(id).await, Some(TaskKind::Aria2)) + && self + .is_aria2_control_epoch_current(id, expected_mapping.epoch) + .await + && self.is_current_aria2_gid_mapping(&gid, &expected_mapping) + && self.aria2_gid_for_download(id).as_deref() == Some(gid.as_str()); + if !still_current { + return Err("Torrent lifecycle changed while reading file progress".to_string()); + } + + Ok(snapshot) + } + + /// Return a bounded, lifecycle-fenced projection of Aria2's piece + /// bitfield. The raw bitfield never crosses the native boundary; the UI + /// receives only exact counts and balanced percentage buckets. + pub async fn get_aria2_torrent_piece_progress( + &self, + id: &str, + ) -> Result { + let _control_guard = self.acquire_aria2_control(id).await; + if !self.is_registered(id).await + || !matches!(self.active_kind(id).await, Some(TaskKind::Aria2)) + { + return Err("live Torrent piece progress is unavailable".to_string()); + } + let is_torrent = self + .aria2_payloads + .lock() + .await + .get(id) + .is_some_and(|payload| payload.is_torrent); + if !is_torrent { + return Err("download is not a Torrent transfer".to_string()); + } + let gid = self + .aria2_gid_for_download(id) + .ok_or_else(|| "live Torrent piece progress is unavailable".to_string())?; + let expected_mapping = self + .aria2_gid_mapping(&gid) + .ok_or_else(|| "live Torrent piece progress is unavailable".to_string())?; + if expected_mapping.id != id + || !self + .is_aria2_control_epoch_current(id, expected_mapping.epoch) + .await + { + return Err("live Torrent piece progress is unavailable".to_string()); + } + + let state = self.app_handle.state::(); + let result = crate::rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.tellStatus", + serde_json::json!([gid, ["bitfield", "pieceLength", "numPieces"]]), + ) + .await + .map_err(|error| { + format!( + "aria2.tellStatus failed: {}", + crate::redact_sensitive_text(&error) + ) + })?; + let snapshot = parse_torrent_piece_progress(result)?; + + let still_current = self.is_registered(id).await + && matches!(self.active_kind(id).await, Some(TaskKind::Aria2)) + && self + .is_aria2_control_epoch_current(id, expected_mapping.epoch) + .await + && self.is_current_aria2_gid_mapping(&gid, &expected_mapping) + && self.aria2_gid_for_download(id).as_deref() == Some(gid.as_str()); + if !still_current { + return Err("Torrent lifecycle changed while reading piece progress".to_string()); + } + + Ok(snapshot) + } + /// Pop the next task, or None if empty. pub async fn pop_front(&self) -> Option { self.pending.lock().await.pop_front() @@ -1287,7 +2367,9 @@ impl QueueManager { control_epoch: u64, ) -> Option { loop { - if !self.is_registered_generation(id, lifecycle_generation).await + if !self + .is_registered_generation_or_legacy(id, lifecycle_generation) + .await || self.is_aria2_retry_cancelled(id).await || !self.is_aria2_control_epoch_current(id, control_epoch).await { @@ -1300,7 +2382,9 @@ impl QueueManager { .try_reserve_queue_slot(id, queue_id, lifecycle_generation) .await { - if self.is_registered_generation(id, lifecycle_generation).await + if self + .is_registered_generation_or_legacy(id, lifecycle_generation) + .await && !self.is_aria2_retry_cancelled(id).await && self .is_aria2_control_epoch_current(id, control_epoch) @@ -1783,6 +2867,9 @@ impl QueueManager { let notified = self.notify.notified(); tokio::pin!(notified); notified.as_mut().enable(); + if self.try_start_waiting_seed().await { + continue; + } if let Some((permit, task)) = self.try_admit_next_task().await { // Admission owns the global and per-queue reservation before // this task is spawned. Keep the dispatcher free to admit @@ -1913,8 +3000,59 @@ impl QueueManager { return; } let buffered_outcome = self.remember_gid(id.clone(), gid.clone()).await; + let install_web_seeds = buffered_outcome.is_none() + && task.payload.is_torrent + && task.payload.torrent_web_seeds.is_some(); self.finish_aria2_dispatch(&id, lifecycle_epoch).await; drop(control_guard); + if install_web_seeds { + if let Err(error) = self.install_initial_torrent_web_seeds(&id).await { + // Initial web-seed installation is a + // post-GID lifecycle step. Reacquire the + // control lock and fence the failure against + // the dispatch epoch before retiring the GID; + // an old install error must never fail a + // newer retry/resume lifecycle. + let _control_guard = self.acquire_aria2_control(&id).await; + let current_gid = self.aria2_gid_for_download(&id); + let current = self + .is_aria2_control_epoch_current(&id, lifecycle_epoch) + .await + && self.is_registered(&id).await + && matches!(self.active_kind(&id).await, Some(TaskKind::Aria2)) + && current_gid.as_deref().is_some_and(|gid| { + self.aria2_gid_mapping(gid).is_some_and(|mapping| { + mapping.id == id && mapping.epoch == lifecycle_epoch + }) + }); + if current { + if let Some(current_gid) = current_gid { + if let Err(remove_error) = self.spawner.remove_uri(¤t_gid).await { + log::warn!( + "aria2 dispatch web-seed failure [{}]: could not remove gid {}: {}", + id, + current_gid, + remove_error + ); + } + self.ignore_aria2_gid(¤t_gid).await; + } + self.apply_completion_locked( + &id, + PendingOutcome::Error(format!( + "could not attach Torrent web seeds: {error}" + )), + ) + .await; + } else { + log::info!( + "aria2 dispatch web-seed failure [{}]: ignoring stale install error after a newer lifecycle took ownership", + id + ); + } + return; + } + } if let Some(outcome) = buffered_outcome { self.handle_aria2_event(&gid, outcome).await; } @@ -2085,8 +3223,33 @@ impl QueueManager { pub(crate) async fn apply_completion_locked(&self, id: &str, outcome: PendingOutcome) { let outcome = match outcome { PendingOutcome::Seeding if self.aria2_torrent_seeding_requested(id).await => { - self.emit_state(id, DownloadStatus::Seeding); - return; + if !self.seed_capacity_enabled() { + // Keep a budget record even when the legacy single-pool + // mode is active. If the user enables separate seed + // capacity while this Torrent is already seeding, the + // existing seeder must be counted before admitting new + // seeders. + self.record_seed_started(id).await; + self.emit_state(id, DownloadStatus::Seeding); + return; + } + match self.admit_seed_after_completion(id).await { + SeedAdmissionOutcome::Seeding => { + self.emit_state(id, DownloadStatus::Seeding); + return; + } + SeedAdmissionOutcome::Waiting => { + let remaining = self + .aria2_payloads + .lock() + .await + .get(id) + .and_then(|payload| payload.torrent_seed_remaining); + self.emit_waiting_to_seed(id, remaining); + return; + } + SeedAdmissionOutcome::Complete => PendingOutcome::Complete, + } } // `onBtDownloadComplete` means that Aria2 is still seeding. If // Firelink has no seeding policy for this payload, reconcile it @@ -3619,6 +4782,200 @@ pub(crate) fn parse_torrent_peer_diagnostics( }) } +fn parse_aria2_decimal(value: Option<&serde_json::Value>, field: &str) -> Result { + let value = value + .and_then(serde_json::Value::as_str) + .ok_or_else(|| format!("aria2.getFiles returned an invalid {field}"))?; + if value.is_empty() || !value.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(format!("aria2.getFiles returned an invalid {field}")); + } + value + .parse::() + .map_err(|_| format!("aria2.getFiles returned an invalid {field}")) +} + +fn parse_aria2_selected(value: Option<&serde_json::Value>) -> Result { + match value { + Some(serde_json::Value::String(value)) if value == "true" => Ok(true), + Some(serde_json::Value::String(value)) if value == "false" => Ok(false), + Some(serde_json::Value::Bool(value)) => Ok(*value), + _ => Err("aria2.getFiles returned an invalid selected flag".to_string()), + } +} + +pub(crate) fn parse_torrent_file_progress( + result: serde_json::Value, + metadata_files: &[crate::ipc::TorrentFile], +) -> Result { + let files = result + .as_array() + .ok_or_else(|| "aria2.getFiles returned a non-array result".to_string())?; + if files.len() != metadata_files.len() { + return Err("aria2.getFiles returned an unexpected file count".to_string()); + } + + let mut parsed = HashMap::::with_capacity(files.len()); + for file in files { + let object = file + .as_object() + .ok_or_else(|| "aria2.getFiles returned malformed file data".to_string())?; + let index = parse_aria2_decimal(object.get("index"), "file index")? + .try_into() + .map_err(|_| "aria2.getFiles returned an invalid file index".to_string())?; + let metadata = metadata_files + .iter() + .find(|metadata| metadata.index == index) + .ok_or_else(|| "aria2.getFiles returned an unknown file index".to_string())?; + if parsed.contains_key(&index) { + return Err("aria2.getFiles returned a duplicate file index".to_string()); + } + let length = parse_aria2_decimal(object.get("length"), "file length")?; + if length != metadata.length { + return Err("aria2.getFiles returned a mismatched file length".to_string()); + } + let completed_length = parse_aria2_decimal( + object.get("completedLength"), + "completed file length", + )?; + if completed_length > length { + return Err("aria2.getFiles returned over-complete file data".to_string()); + } + parsed.insert( + index, + crate::ipc::TorrentFileProgress { + index, + relative_path: metadata.path.clone(), + length, + completed_length, + selected: parse_aria2_selected(object.get("selected"))?, + }, + ); + } + + let files = metadata_files + .iter() + .map(|metadata| { + parsed + .remove(&metadata.index) + .ok_or_else(|| "aria2.getFiles returned a missing file index".to_string()) + }) + .collect::, _>>()?; + if !parsed.is_empty() { + return Err("aria2.getFiles returned unexpected file data".to_string()); + } + Ok(crate::ipc::TorrentFileProgressSnapshot { files }) +} + +fn parse_aria2_status_decimal( + object: &serde_json::Map, + field: &str, +) -> Result { + let value = object + .get(field) + .and_then(serde_json::Value::as_str) + .ok_or_else(|| format!("aria2.tellStatus returned an invalid {field}"))?; + if value.is_empty() || !value.bytes().all(|byte| byte.is_ascii_digit()) { + return Err(format!("aria2.tellStatus returned an invalid {field}")); + } + value + .parse::() + .map_err(|_| format!("aria2.tellStatus returned an invalid {field}")) +} + +fn decode_torrent_piece_bitfield( + value: &str, + num_pieces: u64, +) -> Result, String> { + if num_pieces == 0 { + return Err("aria2.tellStatus returned a non-positive piece count".to_string()); + } + if num_pieces > MAX_TORRENT_PIECES_FOR_PROGRESS { + return Err("aria2.tellStatus returned an unsupported piece count".to_string()); + } + let byte_count = num_pieces + .checked_add(7) + .and_then(|value| value.checked_div(8)) + .ok_or_else(|| "aria2.tellStatus returned an unsupported piece count".to_string())?; + let hex_length = byte_count + .checked_mul(2) + .ok_or_else(|| "aria2.tellStatus returned an oversized bitfield".to_string())?; + if value.len() != usize::try_from(hex_length).unwrap_or(usize::MAX) + || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) + { + return Err("aria2.tellStatus returned an invalid piece bitfield".to_string()); + } + + let mut bytes = Vec::with_capacity(usize::try_from(byte_count).unwrap_or_default()); + for pair in value.as_bytes().chunks_exact(2) { + let high = char::from(pair[0]).to_digit(16).ok_or_else(|| { + "aria2.tellStatus returned an invalid piece bitfield".to_string() + })?; + let low = char::from(pair[1]).to_digit(16).ok_or_else(|| { + "aria2.tellStatus returned an invalid piece bitfield".to_string() + })?; + bytes.push(((high << 4) | low) as u8); + } + + let remainder = (num_pieces % 8) as u8; + if remainder != 0 { + let overflow_mask = (1u8 << (8 - remainder)) - 1; + if bytes.last().is_some_and(|byte| byte & overflow_mask != 0) { + return Err("aria2.tellStatus returned set overflow bits".to_string()); + } + } + Ok(bytes) +} + +fn torrent_piece_is_complete(bitfield: &[u8], index: u64) -> bool { + let byte = bitfield[(index / 8) as usize]; + byte & (1 << (7 - (index % 8))) != 0 +} + +pub(crate) fn parse_torrent_piece_progress( + result: serde_json::Value, +) -> Result { + let object = result + .as_object() + .ok_or_else(|| "aria2.tellStatus returned malformed piece progress".to_string())?; + let piece_length = parse_aria2_status_decimal(object, "pieceLength")?; + let num_pieces = parse_aria2_status_decimal(object, "numPieces")?; + if piece_length == 0 { + return Err("aria2.tellStatus returned a non-positive piece length".to_string()); + } + let bitfield = object + .get("bitfield") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| "aria2.tellStatus has no piece bitfield yet".to_string())?; + let bitfield = decode_torrent_piece_bitfield(bitfield, num_pieces)?; + let completed_pieces = (0..num_pieces) + .filter(|index| torrent_piece_is_complete(&bitfield, *index)) + .count() as u64; + + let bucket_count = num_pieces.min(256) as usize; + let mut buckets = Vec::with_capacity(bucket_count); + for bucket_index in 0..bucket_count { + let start = num_pieces * bucket_index as u64 / bucket_count as u64; + let end = num_pieces * (bucket_index as u64 + 1) / bucket_count as u64; + let bucket_length = end.saturating_sub(start); + let completed = (start..end) + .filter(|index| torrent_piece_is_complete(&bitfield, *index)) + .count() as u64; + let percentage = if bucket_length == 0 { + 0 + } else { + ((completed * 100) / bucket_length).min(100) as u8 + }; + buckets.push(percentage); + } + + Ok(crate::ipc::TorrentPieceProgressSnapshot { + piece_length, + num_pieces, + completed_pieces, + buckets, + }) +} + fn normalize_torrent_tracker_list( value: Option<&str>, allow_wildcard: bool, @@ -3966,6 +5323,48 @@ impl ProductionSpawner { } } } + + async fn control_seed_rpc( + &self, + gid: &str, + method: &str, + ) -> Result { + let state = self.app_handle.state::(); + let port = state.aria2_port.load(std::sync::atomic::Ordering::Relaxed); + let secret = &state.aria2_secret; + let rpc_error = match crate::rpc_call(port, secret, method, serde_json::json!([gid])).await { + Ok(result) => crate::ensure_aria2_gid_result(method, gid, &result) + .err() + .map(|error| error.to_string()), + Err(error) => Some(crate::redact_sensitive_text(&error.to_string())), + }; + let status = crate::aria2_download_status(port, secret, gid) + .await + .map_err(|error| { + rpc_error + .as_ref() + .map(|rpc_error| format!("{rpc_error}; status verification failed: {error}")) + .unwrap_or_else(|| format!("status verification failed: {error}")) + })?; + let outcome = match status.as_str() { + "active" | "waiting" => Aria2SeedControlOutcome::Resumed, + "paused" => Aria2SeedControlOutcome::Paused, + "complete" => Aria2SeedControlOutcome::Complete, + other => { + return Err(format!( + "aria2 {method} left gid {gid} in unhandled state {other}" + )); + } + }; + if let Some(error) = rpc_error { + log::warn!( + "aria2 {method} for seed gid {} returned an error after status verification: {}", + gid, + error + ); + } + Ok(outcome) + } } #[async_trait::async_trait] @@ -4072,11 +5471,18 @@ impl SidecarSpawner for ProductionSpawner { ); } let encoded = base64::engine::general_purpose::STANDARD.encode(bytes); - let uris = payload - .mirrors - .as_deref() - .map(|mirrors| crate::collect_download_uris("", Some(mirrors))) - .unwrap_or_default(); + let uris = if payload.torrent_web_seeds.is_some() { + // Typed per-file web seeds are attached after the GID is + // known through changeUri. addTorrent accepts only one + // unscoped URI list, which cannot represent fileIndex. + Vec::new() + } else { + payload + .mirrors + .as_deref() + .map(|mirrors| crate::collect_download_uris("", Some(mirrors))) + .unwrap_or_default() + }; ("aria2.addTorrent", serde_json::json!([encoded, uris, options])) } else { let parsed = url::Url::parse(&payload.url) @@ -4216,6 +5622,56 @@ impl SidecarSpawner for ProductionSpawner { } } + async fn pause_for_seed( + &self, + gid: &str, + ) -> Result { + self.control_seed_rpc(gid, "aria2.forcePause").await + } + + async fn resume_for_seed( + &self, + gid: &str, + ) -> Result { + self.control_seed_rpc(gid, "aria2.unpause").await + } + + async fn get_torrent_uris(&self, gid: &str) -> Result, String> { + let state = self.app_handle.state::(); + let result = crate::rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.getUris", + serde_json::json!([gid]), + ) + .await + .map_err(|error| format!("aria2.getUris failed for gid {gid}: {}", crate::redact_sensitive_text(&error.to_string())))?; + parse_aria2_web_seed_uris(&result) + } + + async fn change_torrent_uris( + &self, + gid: &str, + file_index: u32, + delete: &[String], + add: &[String], + ) -> Result<(), String> { + let state = self.app_handle.state::(); + let result = crate::rpc_call( + state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), + &state.aria2_secret, + "aria2.changeUri", + serde_json::json!([gid, file_index, delete, add]), + ) + .await + .map_err(|error| format!("aria2.changeUri failed for gid {gid}: {}", crate::redact_sensitive_text(&error.to_string())))?; + match result.as_str() { + Some("OK") => Ok(()), + Some(value) => Err(format!("aria2.changeUri returned unexpected result {value}")), + None => Err("aria2.changeUri returned a non-string result".to_string()), + } + } + async fn recreate_uri( &self, id: &str, @@ -4513,6 +5969,12 @@ pub struct EnqueueItem { pub torrent_seed_ratio: Option, #[serde(default)] #[ts(optional)] + pub torrent_seed_remaining: Option, + #[serde(default)] + #[ts(optional)] + pub torrent_web_seeds: Option>, + #[serde(default)] + #[ts(optional)] pub torrent_upload_limit: Option, #[serde(default)] #[ts(optional)] @@ -4594,8 +6056,10 @@ impl EnqueueItem { is_torrent: self.is_torrent.unwrap_or(false), torrent_path: self.torrent_path, torrent_file_indices: self.torrent_file_indices, - torrent_seed_time: self.torrent_seed_time, + torrent_seed_time: self.torrent_seed_remaining.or(self.torrent_seed_time), torrent_seed_ratio: self.torrent_seed_ratio, + torrent_seed_remaining: self.torrent_seed_remaining, + torrent_web_seeds: self.torrent_web_seeds, torrent_upload_limit: self.torrent_upload_limit, torrent_max_peers: self.torrent_max_peers, torrent_peer_speed_limit: self.torrent_peer_speed_limit, @@ -4642,6 +6106,36 @@ mod tests { } } + struct SeedSpawner; + + #[async_trait::async_trait] + impl SidecarSpawner for SeedSpawner { + async fn add_uri(&self, id: &str, _payload: &SpawnPayload) -> Result { + Ok(format!("gid-{id}")) + } + + async fn remove_uri(&self, _gid: &str) -> Result<(), String> { + Ok(()) + } + + async fn pause_for_seed(&self, _gid: &str) -> Result { + Ok(Aria2SeedControlOutcome::Paused) + } + + async fn resume_for_seed(&self, _gid: &str) -> Result { + Ok(Aria2SeedControlOutcome::Resumed) + } + + async fn run_media( + &self, + _id: &str, + _payload: &SpawnPayload, + _lifecycle_generation: u64, + ) -> Result<(), String> { + Ok(()) + } + } + #[test] fn aria2_connection_options_enable_requested_ranges_for_small_release_assets() { let mut options = serde_json::Map::new(); @@ -5388,6 +6882,246 @@ mod tests { assert!(error.contains("malformed")); } + fn test_torrent_progress_metadata() -> Vec { + vec![ + crate::ipc::TorrentFile { + index: 1, + path: "folder/one.bin".to_string(), + length: 12, + }, + crate::ipc::TorrentFile { + index: 2, + path: "folder/two.bin".to_string(), + length: 7, + }, + ] + } + + #[test] + fn torrent_file_progress_uses_validated_metadata_and_discards_daemon_paths() { + let result = serde_json::json!([ + { + "index": "2", + "path": "/private/secret/two.bin", + "uris": [{"uri": "https://secret.example/file"}], + "length": "7", + "completedLength": "3", + "selected": "false" + }, + { + "index": "1", + "path": "/private/secret/one.bin", + "length": "12", + "completedLength": "12", + "selected": "true" + } + ]); + + let snapshot = parse_torrent_file_progress(result, &test_torrent_progress_metadata()).unwrap(); + assert_eq!(snapshot.files[0].relative_path, "folder/one.bin"); + assert_eq!(snapshot.files[0].completed_length, 12); + assert!(snapshot.files[0].selected); + assert_eq!(snapshot.files[1].relative_path, "folder/two.bin"); + assert_eq!(snapshot.files[1].completed_length, 3); + assert!(!snapshot.files[1].selected); + let serialized = serde_json::to_string(&snapshot).unwrap(); + assert!(!serialized.contains("/private/secret")); + assert!(!serialized.contains("secret.example")); + } + + #[test] + fn torrent_file_progress_rejects_malformed_and_inconsistent_rows() { + let metadata = test_torrent_progress_metadata(); + let cases = [ + ( + serde_json::json!([{ + "index": "1", + "length": "not-a-number", + "completedLength": "0", + "selected": "true" + }, { + "index": "2", + "length": "7", + "completedLength": "0", + "selected": "true" + }]), + "invalid file length", + ), + ( + serde_json::json!([{ + "index": "1", + "length": "12", + "completedLength": "13", + "selected": "true" + }, { + "index": "2", + "length": "7", + "completedLength": "0", + "selected": "true" + }]), + "over-complete", + ), + ( + serde_json::json!([{ + "index": "1", + "length": "12", + "completedLength": "0", + "selected": "true" + }, { + "index": "1", + "length": "12", + "completedLength": "0", + "selected": "true" + }]), + "duplicate", + ), + ( + serde_json::json!([{ + "index": "3", + "length": "12", + "completedLength": "0", + "selected": "true" + }, { + "index": "2", + "length": "7", + "completedLength": "0", + "selected": "true" + }]), + "unknown", + ), + ]; + + for (result, expected) in cases { + let error = parse_torrent_file_progress(result, &metadata).unwrap_err(); + assert!(error.contains(expected), "{error}"); + } + let error = parse_torrent_file_progress(serde_json::json!([]), &metadata).unwrap_err(); + assert!(error.contains("file count")); + } + + #[test] + fn torrent_piece_progress_decodes_high_bit_first_and_buckets_small_torrents() { + let snapshot = parse_torrent_piece_progress(serde_json::json!({ + "pieceLength": "16384", + "numPieces": "10", + "bitfield": "9040" + })) + .unwrap(); + + assert_eq!(snapshot.piece_length, 16_384); + assert_eq!(snapshot.num_pieces, 10); + assert_eq!(snapshot.completed_pieces, 3); + assert_eq!(snapshot.buckets, vec![100, 0, 0, 100, 0, 0, 0, 0, 0, 100]); + } + + #[test] + fn torrent_piece_progress_aggregates_to_at_most_256_balanced_buckets() { + let bitfield = "ff".repeat(64); + let snapshot = parse_torrent_piece_progress(serde_json::json!({ + "pieceLength": "1", + "numPieces": "512", + "bitfield": bitfield + })) + .unwrap(); + + assert_eq!(snapshot.completed_pieces, 512); + assert_eq!(snapshot.buckets.len(), 256); + assert!(snapshot.buckets.iter().all(|value| *value == 100)); + } + + #[test] + fn torrent_piece_progress_rejects_missing_fields_malformed_hex_and_overflow_bits() { + let cases = [ + (serde_json::json!({}), "invalid pieceLength"), + ( + serde_json::json!({ + "pieceLength": "16384", + "numPieces": "0", + "bitfield": "" + }), + "non-positive", + ), + ( + serde_json::json!({ + "pieceLength": "16384", + "numPieces": "8", + "bitfield": "xz" + }), + "invalid piece bitfield", + ), + ( + serde_json::json!({ + "pieceLength": "16384", + "numPieces": "10", + "bitfield": "904f" + }), + "overflow", + ), + ]; + + for (result, expected) in cases { + let error = parse_torrent_piece_progress(result).unwrap_err(); + assert!(error.contains(expected), "{error}"); + } + } + + #[test] + fn torrent_network_limits_and_web_seed_normalization_are_bounded() { + assert_eq!(normalize_torrent_dht_message_timeout(1).unwrap(), 1); + assert_eq!(normalize_torrent_dht_message_timeout(600).unwrap(), 600); + assert!(normalize_torrent_dht_message_timeout(0).is_err()); + assert_eq!(normalize_torrent_max_concurrent_seeds(2).unwrap(), 2); + assert!(normalize_torrent_max_concurrent_seeds(65).is_err()); + + let files = vec![ + crate::ipc::TorrentFile { + index: 1, + path: "folder/file.bin".to_string(), + length: 3, + }, + crate::ipc::TorrentFile { + index: 2, + path: "other.txt".to_string(), + length: 4, + }, + ]; + let seeds = normalize_torrent_web_seeds( + Some(&[ + crate::ipc::TorrentWebSeed { + file_index: 1, + uri: " https://mirror.example/base/ ".to_string(), + }, + crate::ipc::TorrentWebSeed { + file_index: 1, + uri: "https://mirror.example/base/".to_string(), + }, + ]), + &files, + ) + .unwrap(); + assert_eq!(seeds.len(), 1); + assert_eq!( + expand_torrent_web_seeds(&seeds, &files).unwrap()[0].1, + "https://mirror.example/base/folder/file.bin" + ); + assert!(normalize_torrent_web_seeds( + Some(&[crate::ipc::TorrentWebSeed { + file_index: 1, + uri: "ftp://mirror.example/file".to_string(), + }]), + &files, + ) + .is_err()); + assert!(normalize_torrent_web_seeds( + Some(&[crate::ipc::TorrentWebSeed { + file_index: 9, + uri: "https://mirror.example/file".to_string(), + }]), + &files, + ) + .is_err()); + } + #[test] fn enqueue_item_carries_torrent_trackers_into_the_spawn_payload() { let item: EnqueueItem = serde_json::from_value(serde_json::json!({ @@ -5536,6 +7270,102 @@ mod tests { assert_eq!(manager.aria2_gid_for_download("torrent").as_deref(), Some("test-gid")); assert_eq!(manager.available_permits(), 0); + assert!(manager + .capture_seed_remaining("torrent") + .await + .is_some()); + } + + #[tokio::test] + async fn enabling_separate_seed_capacity_counts_existing_seeders() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(SeedSpawner)); + assert!(manager.ensure_aria2_permit("existing").await); + manager + .aria2_payloads + .lock() + .await + .insert( + "existing".to_string(), + SpawnPayload { + is_torrent: true, + torrent_seed_time: Some(5.0), + ..Default::default() + }, + ); + manager + .remember_gid("existing".to_string(), "gid-existing".to_string()) + .await; + manager + .apply_completion("existing", PendingOutcome::Seeding) + .await; + + assert!(!manager.is_seed_owner("existing")); + manager.configure_seed_capacity(true, 1); + assert!(manager.is_seed_owner("existing")); + assert!(manager.has_active_permit("existing").await); + } + + #[tokio::test] + async fn separate_seed_capacity_pauses_fairly_and_rebinds_before_resume() { + let app = tauri::test::mock_builder() + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("mock app"); + let manager = Arc::new(QueueManager::test_new( + app.handle().clone(), + 1, + Arc::new(SeedSpawner), + )); + manager.configure_seed_capacity(true, 1); + for id in ["first", "second"] { + manager.reserve_enqueue_generation(id, 0).await.unwrap(); + assert!(manager.ensure_aria2_permit(id).await); + manager + .aria2_payloads + .lock() + .await + .insert( + id.to_string(), + SpawnPayload { + is_torrent: true, + torrent_seed_time: Some(5.0), + ..Default::default() + }, + ); + manager + .remember_gid(id.to_string(), format!("gid-{id}")) + .await; + if id == "first" { + manager + .apply_completion(id, PendingOutcome::Seeding) + .await; + assert!(manager.is_seed_owner(id)); + assert!(!manager.has_active_permit(id).await); + } else { + manager + .apply_completion(id, PendingOutcome::Seeding) + .await; + assert!(manager.is_waiting_to_seed(id)); + assert!(!manager.has_active_permit(id).await); + } + } + + manager.release_registered_id("first").await; + assert!(manager.try_start_waiting_seed().await); + let result = tokio::time::timeout(Duration::from_secs(1), async { + loop { + if manager.is_seed_owner("second") && !manager.is_waiting_to_seed("second") { + break; + } + tokio::task::yield_now().await; + } + }) + .await; + assert!(result.is_ok(), "waiting seed did not resume after capacity was released"); + assert!(!manager.has_active_permit("second").await); + assert_eq!(manager.current_aria2_control_epoch("second").await, 1); } #[tokio::test] diff --git a/src-tauri/src/settings.rs b/src-tauri/src/settings.rs index c27ffe1..a0e91a7 100644 --- a/src-tauri/src/settings.rs +++ b/src-tauri/src/settings.rs @@ -18,6 +18,7 @@ pub struct TorrentStartupSettings { pub lpd_interface: String, pub peer_id_prefix: String, pub peer_agent: String, + pub dht_message_timeout: u32, } fn normalize_torrent_startup_value( @@ -85,6 +86,10 @@ pub fn torrent_startup_settings(settings: Option<&PersistedSettings>) -> Torrent &settings.torrent_peer_agent, crate::queue::normalize_torrent_peer_agent, ), + dht_message_timeout: crate::queue::normalize_torrent_dht_message_timeout( + settings.torrent_dht_message_timeout, + ) + .unwrap_or(crate::queue::DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT), } } @@ -145,6 +150,32 @@ pub fn canonicalize_torrent_network_settings(stored: &str) -> Result PersistedSettings { torrent_enable_pex: true, torrent_enable_lpd: false, torrent_max_open_files: crate::queue::DEFAULT_TORRENT_MAX_OPEN_FILES, + torrent_dht_message_timeout: crate::queue::DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT, + torrent_separate_seed_slots: false, + torrent_max_concurrent_seeds: crate::queue::DEFAULT_TORRENT_MAX_CONCURRENT_SEEDS, torrent_listen_port: String::new(), torrent_dht_listen_port: String::new(), torrent_external_ip: String::new(), @@ -1085,6 +1142,10 @@ mod tests { settings.torrent_max_open_files, crate::queue::DEFAULT_TORRENT_MAX_OPEN_FILES ); + assert_eq!( + settings.torrent_dht_message_timeout, + crate::queue::DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT + ); assert!(settings.torrent_listen_port.is_empty()); assert!(settings.torrent_dht_listen_port.is_empty()); assert!(settings.torrent_external_ip.is_empty()); @@ -1139,7 +1200,10 @@ mod tests { "torrentListenPort": " 6881-6999 ", "torrentExternalIp": "not-an-ip", "torrentPeerIdPrefix": "123456789012345678901", - "torrentPeerAgent": " Firelink/1.3.1 " + "torrentPeerAgent": " Firelink/1.3.1 ", + "torrentDhtMessageTimeout": 601, + "torrentMaxConcurrentSeeds": 65, + "torrentSeparateSeedSlots": "yes" }, "version": 6 }); @@ -1150,6 +1214,15 @@ mod tests { assert_eq!(canonical["state"]["torrentExternalIp"], ""); assert_eq!(canonical["state"]["torrentPeerIdPrefix"], ""); assert_eq!(canonical["state"]["torrentPeerAgent"], "Firelink/1.3.1"); + assert_eq!( + canonical["state"]["torrentDhtMessageTimeout"], + crate::queue::DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT + ); + assert_eq!( + canonical["state"]["torrentMaxConcurrentSeeds"], + crate::queue::DEFAULT_TORRENT_MAX_CONCURRENT_SEEDS + ); + assert_eq!(canonical["state"]["torrentSeparateSeedSlots"], false); } #[test] @@ -1166,12 +1239,21 @@ mod tests { assert!(startup.listen_port.is_empty()); assert!(startup.peer_id_prefix.is_empty()); assert_eq!(startup.peer_agent, "Firelink/1.3.1"); + assert_eq!( + startup.dht_message_timeout, + crate::queue::DEFAULT_TORRENT_DHT_MESSAGE_TIMEOUT + ); } #[test] fn opt_in_defaults_match_the_frontend_defaults() { assert!(!default_settings().play_completion_sound); assert!(!default_settings().auto_add_clipboard_links); + assert!(!default_settings().torrent_separate_seed_slots); + assert_eq!( + default_settings().torrent_max_concurrent_seeds, + crate::queue::DEFAULT_TORRENT_MAX_CONCURRENT_SEEDS + ); } #[test] diff --git a/src-tauri/src/storage.rs b/src-tauri/src/storage.rs index b918d30..a8f7ba8 100644 --- a/src-tauri/src/storage.rs +++ b/src-tauri/src/storage.rs @@ -5,6 +5,9 @@ pub const PORTABLE_MARKER: &str = "portable.flag"; const PORTABLE_DATA_DIR: &str = "data"; const PORTABLE_LOG_DIR: &str = "logs"; const PORTABLE_WEBVIEW_DIR: &str = "webview"; +const ARIA2_DATA_DIR: &str = "aria2"; +const ARIA2_DHT_FILE: &str = "dht.dat"; +const ARIA2_DHT6_FILE: &str = "dht6.dat"; #[derive(Debug, Clone, PartialEq, Eq)] pub enum StorageMode { @@ -104,6 +107,59 @@ impl StorageLayout { pub fn webview_dir(&self) -> &Path { &self.webview_dir } + + pub fn aria2_dht_paths(&self) -> (PathBuf, PathBuf) { + let directory = self.data_dir.join(ARIA2_DATA_DIR); + ( + directory.join(ARIA2_DHT_FILE), + directory.join(ARIA2_DHT6_FILE), + ) + } + + /// Create and validate only Firelink's Aria2 state directory. Aria2 owns + /// the table contents; Firelink owns this exact location and must never + /// fall back to a user-global default when it cannot establish it. + pub fn prepare_aria2_dht_paths(&self) -> Result<(PathBuf, PathBuf), String> { + let directory = self.data_dir.join(ARIA2_DATA_DIR); + if crate::path_has_symlink_component(&directory) { + return Err(format!( + "Aria2 state directory contains a symlink: '{}'", + directory.display() + )); + } + + match std::fs::symlink_metadata(&directory) { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(format!( + "Aria2 state directory is a symlink: '{}'", + directory.display() + )); + } + Ok(metadata) if !metadata.is_dir() => { + return Err(format!( + "Aria2 state path is not a directory: '{}'", + directory.display() + )); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + std::fs::create_dir(&directory).map_err(|error| { + format!( + "failed to create Aria2 state directory '{}': {error}", + directory.display() + ) + })?; + } + Err(error) => { + return Err(format!( + "failed to inspect Aria2 state directory '{}': {error}", + directory.display() + )); + } + } + + Ok(self.aria2_dht_paths()) + } } fn canonicalize_storage_path(path: &Path) -> Result { @@ -154,7 +210,7 @@ fn canonicalize_storage_path(path: &Path) -> Result { #[cfg(test)] mod tests { - use super::{canonicalize_storage_path, StorageMode, PORTABLE_MARKER}; + use super::{canonicalize_storage_path, StorageLayout, StorageMode, PORTABLE_MARKER}; use std::fs; use std::path::Path; use tempfile::TempDir; @@ -182,6 +238,62 @@ mod tests { ); } + fn test_layout(data_dir: &Path) -> StorageLayout { + let data_dir = fs::canonicalize(data_dir).unwrap(); + StorageLayout { + mode: StorageMode::Standard, + data_dir: data_dir.clone(), + log_dir: data_dir.join("logs"), + webview_dir: data_dir.join("webview"), + } + } + + #[test] + fn aria2_dht_paths_are_owned_by_the_selected_data_directory() { + let root = TempDir::new().unwrap(); + let layout = test_layout(root.path()); + let root_path = fs::canonicalize(root.path()).unwrap(); + + assert_eq!( + layout.aria2_dht_paths(), + ( + root_path.join("aria2/dht.dat"), + root_path.join("aria2/dht6.dat") + ) + ); + let prepared = layout.prepare_aria2_dht_paths().unwrap(); + assert_eq!(prepared, layout.aria2_dht_paths()); + assert!(root_path.join("aria2").is_dir()); + } + + #[test] + fn aria2_dht_preparation_rejects_a_file_at_the_directory_boundary() { + let root = TempDir::new().unwrap(); + let root_path = fs::canonicalize(root.path()).unwrap(); + fs::write(root_path.join("aria2"), b"not a directory").unwrap(); + + let error = test_layout(root.path()) + .prepare_aria2_dht_paths() + .unwrap_err(); + assert!(error.contains("not a directory")); + } + + #[cfg(unix)] + #[test] + fn aria2_dht_preparation_rejects_a_symlinked_directory() { + use std::os::unix::fs::symlink; + + let root = TempDir::new().unwrap(); + let target = TempDir::new().unwrap(); + let root_path = fs::canonicalize(root.path()).unwrap(); + symlink(target.path(), root_path.join("aria2")).unwrap(); + + let error = test_layout(root.path()) + .prepare_aria2_dht_paths() + .unwrap_err(); + assert!(error.contains("symlink")); + } + #[cfg(unix)] #[test] fn rejects_symlinked_storage_directories() { diff --git a/src/bindings/DownloadItem.ts b/src/bindings/DownloadItem.ts index 1a9e4e7..cb6b5bd 100644 --- a/src/bindings/DownloadItem.ts +++ b/src/bindings/DownloadItem.ts @@ -1,5 +1,6 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. import type { DownloadCategory } from "./DownloadCategory"; import type { DownloadStatus } from "./DownloadStatus"; +import type { TorrentWebSeed } from "./TorrentWebSeed"; -export type DownloadItem = { id: string, url: string, fileName: string, status: DownloadStatus, fraction?: number, speed?: string, eta?: string, size?: string, downloadedBytes?: number, totalBytes?: number, totalIsEstimate?: boolean, category: DownloadCategory, dateAdded: string, resumable?: boolean, connections?: number, speedLimit?: string, username?: string, password?: string, headers?: string, checksum?: string, cookies?: string, mirrors?: string, destination?: string, isMedia?: boolean, mediaFormatSelector?: string, mediaQuality?: string, queueId?: string, queuePosition?: number, hasBeenDispatched?: boolean, lastError?: string, lastTry?: string, isTorrent?: boolean, torrentPath?: string, torrentFileIndices?: Array, torrentInfoHash?: string, torrentSeedTime?: number, torrentSeedRatio?: number, torrentUploadLimit?: string, torrentMaxPeers?: number, torrentPeerSpeedLimit?: string, torrentCheckIntegrity?: boolean, torrentTrackers?: string, torrentExcludeTrackers?: string, torrentTrackerConnectTimeout?: number, torrentTrackerTimeout?: number, torrentTrackerInterval?: number, torrentStopTimeout?: number, torrentPrioritizePiece?: string, torrentRemoveUnselectedFile?: boolean, torrentEncryptionPolicy?: string, }; +export type DownloadItem = { id: string, url: string, fileName: string, status: DownloadStatus, fraction?: number, speed?: string, eta?: string, size?: string, downloadedBytes?: number, totalBytes?: number, totalIsEstimate?: boolean, category: DownloadCategory, dateAdded: string, resumable?: boolean, connections?: number, speedLimit?: string, username?: string, password?: string, headers?: string, checksum?: string, cookies?: string, mirrors?: string, destination?: string, isMedia?: boolean, mediaFormatSelector?: string, mediaQuality?: string, queueId?: string, queuePosition?: number, hasBeenDispatched?: boolean, lastError?: string, lastTry?: string, isTorrent?: boolean, torrentPath?: string, torrentFileIndices?: Array, torrentInfoHash?: string, torrentSeedTime?: number, torrentSeedRatio?: number, torrentSeedRemaining?: number, torrentWebSeeds?: Array, torrentUploadLimit?: string, torrentMaxPeers?: number, torrentPeerSpeedLimit?: string, torrentCheckIntegrity?: boolean, torrentTrackers?: string, torrentExcludeTrackers?: string, torrentTrackerConnectTimeout?: number, torrentTrackerTimeout?: number, torrentTrackerInterval?: number, torrentStopTimeout?: number, torrentPrioritizePiece?: string, torrentRemoveUnselectedFile?: boolean, torrentEncryptionPolicy?: string, }; diff --git a/src/bindings/DownloadStateEvent.ts b/src/bindings/DownloadStateEvent.ts index 8db08c3..177ceaf 100644 --- a/src/bindings/DownloadStateEvent.ts +++ b/src/bindings/DownloadStateEvent.ts @@ -1,3 +1,3 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type DownloadStateEvent = { id: string, status: string, error: string | null, fileName?: string, }; +export type DownloadStateEvent = { id: string, status: string, error: string | null, fileName?: string, torrentSeedRemaining?: number, }; diff --git a/src/bindings/DownloadStatus.ts b/src/bindings/DownloadStatus.ts index e2f7a06..9234fa8 100644 --- a/src/bindings/DownloadStatus.ts +++ b/src/bindings/DownloadStatus.ts @@ -1,3 +1,3 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. -export type DownloadStatus = "ready" | "staged" | "downloading" | "processing" | "seeding" | "paused" | "completed" | "failed" | "queued" | "retrying"; +export type DownloadStatus = "ready" | "staged" | "downloading" | "processing" | "seeding" | "waitingToSeed" | "paused" | "completed" | "failed" | "queued" | "retrying"; diff --git a/src/bindings/EnqueueItem.ts b/src/bindings/EnqueueItem.ts index f3b0465..cd68f0c 100644 --- a/src/bindings/EnqueueItem.ts +++ b/src/bindings/EnqueueItem.ts @@ -1,3 +1,4 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { TorrentWebSeed } from "./TorrentWebSeed"; -export type EnqueueItem = { id: string, queue_id: string, url: string, destination: string, filename: string, connections: number | null, speed_limit: string | null, username: string | null, password: string | null, headers: string | null, checksum: string | null, cookies: string | null, mirrors: string | null, user_agent: string | null, max_tries: number | null, proxy: string | null, format_selector: string | null, cookie_source: string | null, is_media: boolean | null, is_torrent?: boolean, torrent_path?: string, torrent_file_indices?: Array, torrent_info_hash?: string, torrent_seed_time?: number, torrent_seed_ratio?: number, torrent_upload_limit?: string, torrent_max_peers?: number, torrent_peer_speed_limit?: string, torrent_check_integrity?: boolean, torrent_trackers?: string, torrent_exclude_trackers?: string, torrent_tracker_connect_timeout?: number, torrent_tracker_timeout?: number, torrent_tracker_interval?: number, torrent_stop_timeout?: number, torrent_prioritize_piece?: string, torrent_remove_unselected_file?: boolean, torrent_encryption_policy?: string, lifecycle_generation?: string, }; +export type EnqueueItem = { id: string, queue_id: string, url: string, destination: string, filename: string, connections: number | null, speed_limit: string | null, username: string | null, password: string | null, headers: string | null, checksum: string | null, cookies: string | null, mirrors: string | null, user_agent: string | null, max_tries: number | null, proxy: string | null, format_selector: string | null, cookie_source: string | null, is_media: boolean | null, is_torrent?: boolean, torrent_path?: string, torrent_file_indices?: Array, torrent_info_hash?: string, torrent_seed_time?: number, torrent_seed_ratio?: number, torrent_seed_remaining?: number, torrent_web_seeds?: Array, torrent_upload_limit?: string, torrent_max_peers?: number, torrent_peer_speed_limit?: string, torrent_check_integrity?: boolean, torrent_trackers?: string, torrent_exclude_trackers?: string, torrent_tracker_connect_timeout?: number, torrent_tracker_timeout?: number, torrent_tracker_interval?: number, torrent_stop_timeout?: number, torrent_prioritize_piece?: string, torrent_remove_unselected_file?: boolean, torrent_encryption_policy?: string, lifecycle_generation?: string, }; diff --git a/src/bindings/PersistedSettings.ts b/src/bindings/PersistedSettings.ts index 066274e..aaede9b 100644 --- a/src/bindings/PersistedSettings.ts +++ b/src/bindings/PersistedSettings.ts @@ -11,4 +11,4 @@ import type { SiteLogin } from "./SiteLogin"; import type { Theme } from "./Theme"; import type { WindowControlStyle } from "./WindowControlStyle"; -export type PersistedSettings = { theme: Theme, fontFamily: FontFamily, windowControlStyle: WindowControlStyle, calendarPreference: CalendarPreference, language: string, baseDownloadFolder: string, categorySubfoldersEnabled: boolean, categorySubfolders: { [key in string]: string }, categoryDirectoryOverrides: { [key in string]: string }, approvedDownloadRoots: Array, maxConcurrentDownloads: number, globalSpeedLimit: string, torrentOverallUploadLimit: string, speedLimitPresetValues: Array, logsEnabled: boolean, isSidebarVisible: boolean, sidebarPosition: string, activeSettingsTab: SettingsTab, scheduler: SchedulerSettings, schedulerRunning: boolean, schedulerActiveDownloadIds: Array, schedulerLastStartKey: string, schedulerLastStopKey: string, lastCustomSpeedLimitKiB: number, lastCustomSpeedLimitUnit: string, perServerConnections: number, maxAutomaticRetries: number, showNotifications: boolean, playCompletionSound: boolean, autoAddClipboardLinks: boolean, appFontSize: AppFontSize, listRowDensity: ListRowDensity, showDockBadge: boolean, showMenuBarIcon: boolean, proxyMode: ProxyMode, proxyHost: string, proxyPort: number, torrentEnableDht: boolean, torrentEnableDht6: boolean, torrentEnablePex: boolean, torrentEnableLpd: boolean, torrentMaxOpenFiles: number, torrentListenPort: string, torrentDhtListenPort: string, torrentExternalIp: string, torrentDhtEntryPoint: string, torrentDhtEntryPoint6: string, torrentDhtListenAddr6: string, torrentLpdInterface: string, torrentPeerIdPrefix: string, torrentPeerAgent: string, customUserAgent: string, askWhereToSaveEachFile: boolean, rememberLastUsedDownloadDirectory: boolean, preventsSleepWhileDownloading: boolean, preventsDisplaySleepWhileDownloading: boolean, mediaCookieSource: MediaCookieSource, siteLogins: Array, autoCheckUpdates: boolean, keychainAccessGranted: boolean, }; +export type PersistedSettings = { theme: Theme, fontFamily: FontFamily, windowControlStyle: WindowControlStyle, calendarPreference: CalendarPreference, language: string, baseDownloadFolder: string, categorySubfoldersEnabled: boolean, categorySubfolders: { [key in string]: string }, categoryDirectoryOverrides: { [key in string]: string }, approvedDownloadRoots: Array, maxConcurrentDownloads: number, globalSpeedLimit: string, torrentOverallUploadLimit: string, speedLimitPresetValues: Array, logsEnabled: boolean, isSidebarVisible: boolean, sidebarPosition: string, activeSettingsTab: SettingsTab, scheduler: SchedulerSettings, schedulerRunning: boolean, schedulerActiveDownloadIds: Array, schedulerLastStartKey: string, schedulerLastStopKey: string, lastCustomSpeedLimitKiB: number, lastCustomSpeedLimitUnit: string, perServerConnections: number, maxAutomaticRetries: number, showNotifications: boolean, playCompletionSound: boolean, autoAddClipboardLinks: boolean, appFontSize: AppFontSize, listRowDensity: ListRowDensity, showDockBadge: boolean, showMenuBarIcon: boolean, proxyMode: ProxyMode, proxyHost: string, proxyPort: number, torrentEnableDht: boolean, torrentEnableDht6: boolean, torrentEnablePex: boolean, torrentEnableLpd: boolean, torrentMaxOpenFiles: number, torrentDhtMessageTimeout: number, torrentSeparateSeedSlots: boolean, torrentMaxConcurrentSeeds: number, torrentListenPort: string, torrentDhtListenPort: string, torrentExternalIp: string, torrentDhtEntryPoint: string, torrentDhtEntryPoint6: string, torrentDhtListenAddr6: string, torrentLpdInterface: string, torrentPeerIdPrefix: string, torrentPeerAgent: string, customUserAgent: string, askWhereToSaveEachFile: boolean, rememberLastUsedDownloadDirectory: boolean, preventsSleepWhileDownloading: boolean, preventsDisplaySleepWhileDownloading: boolean, mediaCookieSource: MediaCookieSource, siteLogins: Array, autoCheckUpdates: boolean, keychainAccessGranted: boolean, }; diff --git a/src/bindings/TorrentFileProgress.ts b/src/bindings/TorrentFileProgress.ts new file mode 100644 index 0000000..a19f195 --- /dev/null +++ b/src/bindings/TorrentFileProgress.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type TorrentFileProgress = { index: number, relativePath: string, length: number, completedLength: number, selected: boolean, }; diff --git a/src/bindings/TorrentFileProgressSnapshot.ts b/src/bindings/TorrentFileProgressSnapshot.ts new file mode 100644 index 0000000..b2a6561 --- /dev/null +++ b/src/bindings/TorrentFileProgressSnapshot.ts @@ -0,0 +1,4 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { TorrentFileProgress } from "./TorrentFileProgress"; + +export type TorrentFileProgressSnapshot = { files: Array, }; diff --git a/src/bindings/TorrentPieceProgressSnapshot.ts b/src/bindings/TorrentPieceProgressSnapshot.ts new file mode 100644 index 0000000..1e726d2 --- /dev/null +++ b/src/bindings/TorrentPieceProgressSnapshot.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type TorrentPieceProgressSnapshot = { pieceLength: number, numPieces: number, completedPieces: number, buckets: Array, }; diff --git a/src/bindings/TorrentWebSeed.ts b/src/bindings/TorrentWebSeed.ts new file mode 100644 index 0000000..1274592 --- /dev/null +++ b/src/bindings/TorrentWebSeed.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type TorrentWebSeed = { fileIndex: number, uri: string, }; diff --git a/src/components/PropertiesModal.tsx b/src/components/PropertiesModal.tsx index 35d6d37..5214479 100644 --- a/src/components/PropertiesModal.tsx +++ b/src/components/PropertiesModal.tsx @@ -4,6 +4,9 @@ import { useDownloadProgressStore } from '../store/downloadProgressStore'; import { useShallow } from 'zustand/react/shallow'; import { useSettingsStore } from '../store/useSettingsStore'; import type { TorrentPeerDiagnostics } from '../bindings/TorrentPeerDiagnostics'; +import type { TorrentFileProgressSnapshot } from '../bindings/TorrentFileProgressSnapshot'; +import type { TorrentPieceProgressSnapshot } from '../bindings/TorrentPieceProgressSnapshot'; +import type { TorrentWebSeed } from '../bindings/TorrentWebSeed'; import { invokeCommand as invoke } from '../ipc'; import { ChevronDown, ChevronRight, FolderPlus, Info, CheckCircle, AlertCircle, Play, Pause } from 'lucide-react'; import { open } from '@tauri-apps/plugin-dialog'; @@ -42,6 +45,9 @@ const formatLastTry = ( const isPeerDiagnosticsStatus = (status: string): boolean => ['downloading', 'seeding', 'retrying'].includes(status); +const isTorrentFileProgressStatus = (status: string): boolean => + ['downloading', 'seeding', 'waitingToSeed', 'retrying', 'paused'].includes(status); + const formatPeerSpeed = (bytesPerSecond: number): string => `${formatDownloadBytes(bytesPerSecond)}/s`; @@ -99,9 +105,18 @@ export const PropertiesModal = () => { const [torrentPeerDiagnostics, setTorrentPeerDiagnostics] = useState(null); const [torrentPeerDiagnosticsError, setTorrentPeerDiagnosticsError] = useState(false); const [isTorrentPeerDiagnosticsPending, setIsTorrentPeerDiagnosticsPending] = useState(false); + const [torrentFileProgress, setTorrentFileProgress] = useState(null); + const [torrentFileProgressError, setTorrentFileProgressError] = useState(false); + const [isTorrentFileProgressPending, setIsTorrentFileProgressPending] = useState(false); + const [torrentPieceProgress, setTorrentPieceProgress] = useState(null); + const [torrentPieceProgressError, setTorrentPieceProgressError] = useState(false); + const [isTorrentPieceProgressPending, setIsTorrentPieceProgressPending] = useState(false); const [isLiveSpeedLimitPending, setIsLiveSpeedLimitPending] = useState(false); const [isLiveTorrentUploadLimitPending, setIsLiveTorrentUploadLimitPending] = useState(false); const [isLiveTorrentPeerOptionsPending, setIsLiveTorrentPeerOptionsPending] = useState(false); + const [torrentWebSeedsText, setTorrentWebSeedsText] = useState(''); + const [torrentWebSeedsError, setTorrentWebSeedsError] = useState(false); + const [isTorrentWebSeedsPending, setIsTorrentWebSeedsPending] = useState(false); const [loginMode, setLoginMode] = useState('matching'); const [username, setUsername] = useState(''); @@ -119,6 +134,9 @@ export const PropertiesModal = () => { const [isPauseResumePending, setIsPauseResumePending] = useState(false); const actionRequestRef = useRef(0); const peerDiagnosticsRequestRef = useRef(0); + const torrentFileProgressRequestRef = useRef(0); + const torrentPieceProgressRequestRef = useRef(0); + const torrentWebSeedsRequestRef = useRef(0); const modalRef = useModalFocus(Boolean(selectedPropertiesDownloadId && item)); useEffect(() => { @@ -132,6 +150,17 @@ export const PropertiesModal = () => { setTorrentPeerDiagnostics(null); setTorrentPeerDiagnosticsError(false); setIsTorrentPeerDiagnosticsPending(false); + torrentFileProgressRequestRef.current += 1; + setTorrentFileProgress(null); + setTorrentFileProgressError(false); + setIsTorrentFileProgressPending(false); + torrentPieceProgressRequestRef.current += 1; + setTorrentPieceProgress(null); + setTorrentPieceProgressError(false); + setIsTorrentPieceProgressPending(false); + torrentWebSeedsRequestRef.current += 1; + setTorrentWebSeedsError(false); + setIsTorrentWebSeedsPending(false); }, [selectedPropertiesDownloadId]); useEffect(() => { @@ -205,6 +234,9 @@ export const PropertiesModal = () => { setTorrentTrackerInterval(activeItem.torrentTrackerInterval === undefined ? '0' : String(activeItem.torrentTrackerInterval)); setTorrentStopTimeout(activeItem.torrentStopTimeout === undefined ? '0' : String(activeItem.torrentStopTimeout)); setTorrentPrioritizePiece(activeItem.torrentPrioritizePiece || ''); + setTorrentWebSeedsText((activeItem.torrentWebSeeds || []) + .map(seed => `${seed.fileIndex}|${seed.uri}`) + .join('\n')); setErrorMessage(''); } else { setSelectedPropertiesDownloadId(null); @@ -229,6 +261,120 @@ export const PropertiesModal = () => { setIsTorrentPeerDiagnosticsPending(false); }, [item?.id, item?.isTorrent, item?.lastTry, item?.status]); + useEffect(() => { + torrentFileProgressRequestRef.current += 1; + setTorrentFileProgress(null); + setTorrentFileProgressError(false); + setIsTorrentFileProgressPending(false); + if ( + !selectedPropertiesDownloadId + || !item?.isTorrent + || !isTorrentFileProgressStatus(item.status) + ) return; + + const requestId = torrentFileProgressRequestRef.current; + const propertiesDownloadId = item.id; + setIsTorrentFileProgressPending(true); + void invoke('get_torrent_file_progress', { id: propertiesDownloadId }) + .then(snapshot => { + const currentItem = useDownloadStore.getState().downloads.find(download => download.id === propertiesDownloadId); + if ( + requestId === torrentFileProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + && currentItem?.isTorrent + && isTorrentFileProgressStatus(currentItem.status) + ) { + setTorrentFileProgress(snapshot); + } + }) + .catch(() => { + const currentItem = useDownloadStore.getState().downloads.find(download => download.id === propertiesDownloadId); + if ( + requestId === torrentFileProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + && currentItem?.isTorrent + && isTorrentFileProgressStatus(currentItem.status) + ) { + setTorrentFileProgressError(true); + } + }) + .finally(() => { + if (requestId === torrentFileProgressRequestRef.current) { + setIsTorrentFileProgressPending(false); + } + }); + }, [item?.id, item?.isTorrent, item?.lastTry, item?.status, selectedPropertiesDownloadId]); + + useEffect(() => { + torrentWebSeedsRequestRef.current += 1; + setTorrentWebSeedsError(false); + setIsTorrentWebSeedsPending(false); + if (!selectedPropertiesDownloadId || !item?.isTorrent || !isTorrentFileProgressStatus(item.status)) return; + const requestId = torrentWebSeedsRequestRef.current; + const propertiesDownloadId = item.id; + setIsTorrentWebSeedsPending(true); + void invoke('get_torrent_web_seeds', { id: propertiesDownloadId }) + .then(seeds => { + if ( + requestId === torrentWebSeedsRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + ) { + setTorrentWebSeedsText(seeds.map(seed => `${seed.fileIndex}|${seed.uri}`).join('\n')); + useDownloadStore.getState().updateDownload(propertiesDownloadId, { torrentWebSeeds: seeds }); + } + }) + .catch(() => { + if (requestId === torrentWebSeedsRequestRef.current) setTorrentWebSeedsError(true); + }) + .finally(() => { + if (requestId === torrentWebSeedsRequestRef.current) setIsTorrentWebSeedsPending(false); + }); + }, [item?.id, item?.isTorrent, item?.status, selectedPropertiesDownloadId]); + + useEffect(() => { + torrentPieceProgressRequestRef.current += 1; + setTorrentPieceProgress(null); + setTorrentPieceProgressError(false); + setIsTorrentPieceProgressPending(false); + if ( + !selectedPropertiesDownloadId + || !item?.isTorrent + || !isTorrentFileProgressStatus(item.status) + ) return; + + const requestId = torrentPieceProgressRequestRef.current; + const propertiesDownloadId = item.id; + setIsTorrentPieceProgressPending(true); + void invoke('get_torrent_piece_progress', { id: propertiesDownloadId }) + .then(snapshot => { + const currentItem = useDownloadStore.getState().downloads.find(download => download.id === propertiesDownloadId); + if ( + requestId === torrentPieceProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + && currentItem?.isTorrent + && isTorrentFileProgressStatus(currentItem.status) + ) { + setTorrentPieceProgress(snapshot); + } + }) + .catch(() => { + const currentItem = useDownloadStore.getState().downloads.find(download => download.id === propertiesDownloadId); + if ( + requestId === torrentPieceProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + && currentItem?.isTorrent + && isTorrentFileProgressStatus(currentItem.status) + ) { + setTorrentPieceProgressError(true); + } + }) + .finally(() => { + if (requestId === torrentPieceProgressRequestRef.current) { + setIsTorrentPieceProgressPending(false); + } + }); + }, [item?.id, item?.isTorrent, item?.lastTry, item?.status, selectedPropertiesDownloadId]); + useEffect(() => { setLiveTorrentMaxPeersValue( item?.torrentMaxPeers === undefined ? '' : String(item.torrentMaxPeers) @@ -321,6 +467,108 @@ export const PropertiesModal = () => { } }; + const handleRefreshTorrentFileProgress = async () => { + if ( + isTorrentFileProgressPending + || !item.isTorrent + || !isTorrentFileProgressStatus(item.status) + ) return; + + const requestId = ++torrentFileProgressRequestRef.current; + const propertiesDownloadId = item.id; + setIsTorrentFileProgressPending(true); + setTorrentFileProgressError(false); + try { + const snapshot = await invoke('get_torrent_file_progress', { id: propertiesDownloadId }); + const currentItem = useDownloadStore.getState().downloads.find(download => download.id === propertiesDownloadId); + if ( + requestId === torrentFileProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + && currentItem?.isTorrent + && isTorrentFileProgressStatus(currentItem.status) + ) { + setTorrentFileProgress(snapshot); + } + } catch { + if ( + requestId === torrentFileProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + ) { + setTorrentFileProgressError(true); + setTorrentFileProgress(null); + } + } finally { + if (requestId === torrentFileProgressRequestRef.current) { + setIsTorrentFileProgressPending(false); + } + } + }; + + const handleRefreshTorrentPieceProgress = async () => { + if ( + isTorrentPieceProgressPending + || !item.isTorrent + || !isTorrentFileProgressStatus(item.status) + ) return; + + const requestId = ++torrentPieceProgressRequestRef.current; + const propertiesDownloadId = item.id; + setIsTorrentPieceProgressPending(true); + setTorrentPieceProgressError(false); + try { + const snapshot = await invoke('get_torrent_piece_progress', { id: propertiesDownloadId }); + const currentItem = useDownloadStore.getState().downloads.find(download => download.id === propertiesDownloadId); + if ( + requestId === torrentPieceProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + && currentItem?.isTorrent + && isTorrentFileProgressStatus(currentItem.status) + ) { + setTorrentPieceProgress(snapshot); + } + } catch { + if ( + requestId === torrentPieceProgressRequestRef.current + && useDownloadStore.getState().selectedPropertiesDownloadId === propertiesDownloadId + ) { + setTorrentPieceProgressError(true); + setTorrentPieceProgress(null); + } + } finally { + if (requestId === torrentPieceProgressRequestRef.current) { + setIsTorrentPieceProgressPending(false); + } + } + }; + + const handleTorrentWebSeedsSave = async () => { + if (!item?.isTorrent || isTorrentWebSeedsPending) return; + const seeds: TorrentWebSeed[] = []; + for (const line of torrentWebSeedsText.split(/\r?\n/)) { + const trimmed = line.trim(); + if (!trimmed) continue; + const separator = trimmed.indexOf('|'); + const fileIndex = Number(separator >= 0 ? trimmed.slice(0, separator).trim() : ''); + const uri = separator >= 0 ? trimmed.slice(separator + 1).trim() : ''; + if (!Number.isInteger(fileIndex) || fileIndex < 0 || !uri) { + setTorrentWebSeedsError(true); + return; + } + seeds.push({ fileIndex, uri }); + } + setIsTorrentWebSeedsPending(true); + setTorrentWebSeedsError(false); + try { + const normalized = await invoke('set_torrent_web_seeds', { id: item.id, seeds }); + setTorrentWebSeedsText(normalized.map(seed => `${seed.fileIndex}|${seed.uri}`).join('\n')); + useDownloadStore.getState().updateDownload(item.id, { torrentWebSeeds: normalized }); + } catch { + setTorrentWebSeedsError(true); + } finally { + setIsTorrentWebSeedsPending(false); + } + }; + const handleSave = async () => { if (!url.trim()) { setErrorMessage(t($ => $.properties.enterValidUrl)); @@ -857,6 +1105,127 @@ export const PropertiesModal = () => {
{t($ => $.properties.torrentPeerOptionsSavedHint)}
+
+
+
+ {t($ => $.properties.torrentPieceProgress)} +
+ +
+

+ {t($ => $.properties.torrentPieceProgressHint)} +

+ {!isTorrentFileProgressStatus(item.status) && ( +

+ {t($ => $.properties.torrentPieceProgressUnavailable)} +

+ )} + {torrentPieceProgressError && ( +

+ {t($ => $.properties.torrentPieceProgressFailed)} +

+ )} + {torrentPieceProgress && ( + <> +
+ {t($ => $.properties.torrentPieceProgressSummary, { + completed: torrentPieceProgress.completedPieces, + total: torrentPieceProgress.numPieces, + size: formatDownloadBytes(torrentPieceProgress.pieceLength), + })} +
+
$.properties.torrentPieceProgressMap)} + > + {torrentPieceProgress.buckets.map((percentage, index) => ( + + ))} +
+ + )} +
+
+
+
+ {t($ => $.properties.torrentFileProgress)} +
+ +
+

+ {t($ => $.properties.torrentFileProgressHint)} +

+ {!isTorrentFileProgressStatus(item.status) && ( +

+ {t($ => $.properties.torrentFileProgressUnavailable)} +

+ )} + {torrentFileProgressError && ( +

+ {t($ => $.properties.torrentFileProgressFailed)} +

+ )} + {torrentFileProgress && ( +
+ + + + + + + + + + + {torrentFileProgress.files.map(file => { + const percentage = file.length === 0 + ? 100 + : Math.round((file.completedLength / file.length) * 100); + return ( + + + + + + + ); + })} + +
#{t($ => $.properties.torrentFileProgressPath)}{t($ => $.properties.torrentFileProgressCompleted)}{t($ => $.properties.torrentFileProgressSelected)}
{file.index}{file.relativePath} + {formatDownloadBytes(file.completedLength)} / {formatDownloadBytes(file.length)} ({percentage}%) + + {file.selected + ? t($ => $.properties.torrentFileProgressSelected) + : t($ => $.properties.torrentFileProgressUnselected)} +
+
+ )} +
@@ -1313,6 +1682,36 @@ export const PropertiesModal = () => {
+ {item.isTorrent && ( +
+

+ {t($ => $.properties.torrentWebSeeds)} +

+

{t($ => $.properties.torrentWebSeedsHint)}

+