From d621c0eb308fa34d93cf1d6530fdd91ace6130b8 Mon Sep 17 00:00:00 2001 From: TheDancingDeveloper Date: Thu, 30 Jul 2026 00:04:37 +0000 Subject: [PATCH] feat: restore trackers and add torrent recheck --- crates/librtbit/src/api.rs | 28 ++- crates/librtbit/src/http_api/handlers/mod.rs | 10 + .../src/http_api/handlers/qbit_compat.rs | 70 +++++- .../src/http_api/handlers/torrents.rs | 142 +++++++++++ crates/librtbit/src/http_api/mod.rs | 3 + crates/librtbit/src/session/mod.rs | 71 +++++- crates/librtbit/src/session/peer_sources.rs | 2 +- .../librtbit/src/session_persistence/json.rs | 1 + .../librtbit/src/session_persistence/mod.rs | 11 +- .../src/session_persistence/postgres.rs | 48 ++-- .../librtbit/src/tests/read_only_storage.rs | 226 ++++++++++++++++++ crates/librtbit/src/torrent_state/mod.rs | 75 +++++- crates/librtbit/webui/e2e/mock-ui.spec.ts | 32 +++ crates/librtbit/webui/src/api-types.ts | 2 + .../src/components/TorrentContextMenu.tsx | 32 ++- crates/librtbit/webui/src/context.tsx | 6 + crates/librtbit/webui/src/http-api.ts | 13 + crates/librtbit/webui/src/mock-api.ts | 9 + 18 files changed, 746 insertions(+), 35 deletions(-) diff --git a/crates/librtbit/src/api.rs b/crates/librtbit/src/api.rs index 30f5bb2..bf00ee9 100644 --- a/crates/librtbit/src/api.rs +++ b/crates/librtbit/src/api.rs @@ -259,6 +259,7 @@ impl Api { trackers: Some( mgr.shared() .trackers + .read() .iter() .map(|t| t.to_string()) .collect(), @@ -352,6 +353,31 @@ impl Api { Ok(Default::default()) } + pub async fn api_torrent_action_recheck( + &self, + idx: TorrentIdOrHash, + ) -> Result { + let handle = self.mgr_handle(idx)?; + self.session + .force_recheck(&handle) + .await + .with_status(StatusCode::BAD_REQUEST)?; + Ok(Default::default()) + } + + pub async fn api_torrent_action_add_trackers( + &self, + idx: TorrentIdOrHash, + trackers: Vec, + ) -> Result { + let handle = self.mgr_handle(idx)?; + self.session + .add_trackers(&handle, trackers) + .await + .with_status(StatusCode::BAD_REQUEST)?; + Ok(Default::default()) + } + pub async fn api_torrent_action_forget( &self, idx: TorrentIdOrHash, @@ -589,7 +615,7 @@ impl Api { let mgr = self.mgr_handle(idx)?; let registry = &mgr.shared().tracker_status; // Make sure every configured tracker shows up, even before any announce. - for t in mgr.shared().trackers.iter() { + for t in mgr.shared().trackers.read().iter() { registry.ensure(t.as_str()); } Ok(TrackerStatusResponse { diff --git a/crates/librtbit/src/http_api/handlers/mod.rs b/crates/librtbit/src/http_api/handlers/mod.rs index ee3eaaf..6bd407e 100644 --- a/crates/librtbit/src/http_api/handlers/mod.rs +++ b/crates/librtbit/src/http_api/handlers/mod.rs @@ -67,6 +67,8 @@ async fn h_api_root(parts: Parts) -> impl IntoResponse { "POST /torrents/resolve_magnet": "Resolve a magnet to torrent file bytes", "POST /torrents/{id_or_infohash}/pause": "Pause torrent", "POST /torrents/{id_or_infohash}/start": "Resume torrent", + "POST /torrents/{id_or_infohash}/recheck": "Force a full integrity check", + "POST /torrents/{id_or_infohash}/trackers": "Attach tracker announce URLs", "POST /torrents/{id_or_infohash}/forget": "Forget about the torrent, keep the files", "POST /torrents/{id_or_infohash}/delete": "Forget about the torrent, remove the files", "POST /torrents/{id_or_infohash}/add_peers": "Add peers (newline-delimited)", @@ -137,6 +139,14 @@ pub fn make_api_router(state: ApiState) -> Router { "/torrents/{id}/start", post(torrents::h_torrent_action_start), ) + .route( + "/torrents/{id}/recheck", + post(torrents::h_torrent_action_recheck), + ) + .route( + "/torrents/{id}/trackers", + post(torrents::h_torrent_action_add_trackers), + ) .route( "/torrents/{id}/forget", post(torrents::h_torrent_action_forget), diff --git a/crates/librtbit/src/http_api/handlers/qbit_compat.rs b/crates/librtbit/src/http_api/handlers/qbit_compat.rs index 3be4804..a2482a9 100644 --- a/crates/librtbit/src/http_api/handlers/qbit_compat.rs +++ b/crates/librtbit/src/http_api/handlers/qbit_compat.rs @@ -589,12 +589,13 @@ async fn h_torrents_info( let tracker = handle .shared() .trackers + .read() .iter() .next() .map(|u| u.to_string()) .unwrap_or_default(); - let trackers_count = handle.shared().trackers.len(); + let trackers_count = handle.shared().trackers.read().len(); Some(QbitTorrentInfo { added_on, @@ -1062,6 +1063,17 @@ async fn h_torrents_resume(State(state): State>, body: Bytes) -> "Ok." } +async fn h_torrents_recheck(State(state): State>, body: Bytes) -> &'static str { + let form: HashesForm = serde_urlencoded::from_bytes(&body).unwrap_or_default(); + let api = &state.api_state.api; + for idx in resolve_hashes(api, &form.hashes) { + if let Err(error) = api.api_torrent_action_recheck(idx).await { + warn!(%error, "qbit compat: error rechecking torrent"); + } + } + "Ok." +} + #[derive(Deserialize, Default)] struct DeleteForm { #[serde(default)] @@ -1243,6 +1255,7 @@ pub(crate) fn make_qbit_router(api_state: ApiState) -> Router { .route("/add", post(h_torrents_add)) .route("/pause", post(h_torrents_pause)) .route("/resume", post(h_torrents_resume)) + .route("/recheck", post(h_torrents_recheck)) .route("/delete", post(h_torrents_delete)) .route("/setCategory", post(h_torrents_set_category)) .route("/categories", get(h_categories)) @@ -1304,14 +1317,17 @@ mod tests { use http::StatusCode; use crate::{ - Api, ListenerMode, Session, SessionOptions, + AddTorrent, AddTorrentOptions, Api, CreateTorrentOptions, ListenerMode, Session, + SessionOptions, create_torrent, http_api::{HttpApi, HttpApiOptions}, listen::ListenerOptions, + spawn_utils::BlockingSpawner, + torrent_state::TorrentStatsState, }; use super::{ - QbitSessions, QbitState, h_app_preferences, h_app_set_preferences, matches_category, - qbit_file_name, + QbitSessions, QbitState, h_app_preferences, h_app_set_preferences, h_torrents_recheck, + matches_category, qbit_file_name, }; async fn qbit_state() -> (Arc, Arc, tempfile::TempDir) { @@ -1409,6 +1425,52 @@ mod tests { assert_eq!(session.announce_port(), Some(4241)); } + #[tokio::test] + async fn recheck_endpoint_accepts_qbittorrent_hash_form() { + let (state, session, output) = qbit_state().await; + std::fs::write(output.path().join("payload.bin"), vec![0x71; 32 * 1024]).unwrap(); + let torrent = create_torrent( + output.path(), + CreateTorrentOptions { + piece_length: Some(16_384), + ..Default::default() + }, + &BlockingSpawner::new(1), + ) + .await + .unwrap() + .as_bytes() + .unwrap() + .to_vec(); + let handle = session + .add_torrent( + AddTorrent::from_bytes(torrent), + Some(AddTorrentOptions { + paused: true, + overwrite: true, + output_folder: Some(output.path().to_string_lossy().into_owned()), + ..Default::default() + }), + ) + .await + .unwrap() + .into_handle() + .unwrap(); + handle.wait_until_initialized().await.unwrap(); + + let body = Bytes::from(format!("hashes={}", handle.info_hash().as_string())); + assert_eq!( + h_torrents_recheck(axum::extract::State(state), body).await, + "Ok." + ); + assert!(matches!( + handle.stats().state, + TorrentStatsState::Initializing + )); + handle.wait_until_initialized().await.unwrap(); + assert!(matches!(handle.stats().state, TorrentStatsState::Paused)); + } + #[test] fn category_filter_supports_qbittorrent_special_values() { assert!(matches_category("all", "Linux ISOs")); diff --git a/crates/librtbit/src/http_api/handlers/torrents.rs b/crates/librtbit/src/http_api/handlers/torrents.rs index 377bc0d..74b9fda 100644 --- a/crates/librtbit/src/http_api/handlers/torrents.rs +++ b/crates/librtbit/src/http_api/handlers/torrents.rs @@ -304,6 +304,52 @@ pub async fn h_torrent_action_start( .map(axum::Json) } +#[cfg_attr(feature = "swagger", utoipa::path( + post, + path = "/torrents/{id}/recheck", + params(("id" = String, Path, description = "Torrent ID or info hash")), + responses( + (status = 200, description = "Full integrity check started", body = EmptyJsonResponse) + ) +))] +pub async fn h_torrent_action_recheck( + State(state): State, + Path(idx): Path, +) -> Result { + state + .api + .api_torrent_action_recheck(idx) + .await + .map(axum::Json) +} + +#[derive(Deserialize)] +#[cfg_attr(feature = "swagger", derive(utoipa::ToSchema))] +pub struct AddTrackersRequest { + trackers: Vec, +} + +#[cfg_attr(feature = "swagger", utoipa::path( + post, + path = "/torrents/{id}/trackers", + params(("id" = String, Path, description = "Torrent ID or info hash")), + request_body(content = AddTrackersRequest, description = "Tracker announce URLs to merge"), + responses( + (status = 200, description = "Trackers attached", body = EmptyJsonResponse) + ) +))] +pub async fn h_torrent_action_add_trackers( + State(state): State, + Path(idx): Path, + axum::Json(req): axum::Json, +) -> Result { + state + .api + .api_torrent_action_add_trackers(idx, req.trackers) + .await + .map(axum::Json) +} + #[cfg_attr(feature = "swagger", utoipa::path( post, path = "/torrents/{id}/forget", @@ -636,3 +682,99 @@ pub async fn h_set_torrent_category( .await .map(axum::Json) } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use anyhow::Context; + use axum::extract::{Path, State}; + use tempfile::TempDir; + + use super::{AddTrackersRequest, h_torrent_action_add_trackers, h_torrent_action_recheck}; + use crate::{ + AddTorrent, AddTorrentOptions, Api, CreateTorrentOptions, Session, SessionOptions, + create_torrent, + http_api::{HttpApi, HttpApiOptions}, + spawn_utils::BlockingSpawner, + torrent_state::{ManagedTorrentHandle, TorrentStatsState}, + }; + + async fn torrent_api() -> anyhow::Result<(Arc, ManagedTorrentHandle, TempDir)> { + let payload = TempDir::with_prefix("rtbit_http_action_payload")?; + std::fs::write(payload.path().join("payload.bin"), vec![0x5e; 32 * 1024])?; + let torrent = create_torrent( + payload.path(), + CreateTorrentOptions { + piece_length: Some(16_384), + ..Default::default() + }, + &BlockingSpawner::new(1), + ) + .await? + .as_bytes()? + .to_vec(); + let session = Session::new_with_opts( + payload.path().to_owned(), + SessionOptions { + disable_dht: true, + disable_trackers: true, + disable_local_service_discovery: true, + ..Default::default() + }, + ) + .await?; + let handle = session + .add_torrent( + AddTorrent::from_bytes(torrent), + Some(AddTorrentOptions { + paused: true, + overwrite: true, + output_folder: Some(payload.path().to_string_lossy().into_owned()), + ..Default::default() + }), + ) + .await? + .into_handle() + .context("expected torrent handle")?; + handle.wait_until_initialized().await?; + let api = Api::new( + session, + None, + #[cfg(feature = "tracing-subscriber-utils")] + None, + ); + Ok(( + Arc::new(HttpApi::new(api, Some(HttpApiOptions::default()))), + handle, + payload, + )) + } + + #[tokio::test] + async fn add_trackers_handler_merges_urls() -> anyhow::Result<()> { + let (state, handle, _payload) = torrent_api().await?; + h_torrent_action_add_trackers( + State(state), + Path(handle.id().into()), + axum::Json(AddTrackersRequest { + trackers: vec!["https://tracker.example.test/announce".to_string()], + }), + ) + .await + .map_err(|error| anyhow::anyhow!("{error:?}"))?; + assert_eq!(handle.shared().trackers.read().len(), 1); + Ok(()) + } + + #[tokio::test] + async fn recheck_handler_starts_full_check() -> anyhow::Result<()> { + let (state, handle, _payload) = torrent_api().await?; + h_torrent_action_recheck(State(state), Path(handle.id().into())) + .await + .map_err(|error| anyhow::anyhow!("{error:?}"))?; + handle.wait_until_initialized().await?; + assert!(matches!(handle.stats().state, TorrentStatsState::Paused)); + Ok(()) + } +} diff --git a/crates/librtbit/src/http_api/mod.rs b/crates/librtbit/src/http_api/mod.rs index ffd1621..72bc502 100644 --- a/crates/librtbit/src/http_api/mod.rs +++ b/crates/librtbit/src/http_api/mod.rs @@ -39,6 +39,8 @@ mod webui; handlers::torrents::h_peer_stats, handlers::torrents::h_torrent_action_pause, handlers::torrents::h_torrent_action_start, + handlers::torrents::h_torrent_action_recheck, + handlers::torrents::h_torrent_action_add_trackers, handlers::torrents::h_torrent_action_forget, handlers::torrents::h_torrent_action_delete, handlers::torrents::h_torrent_action_update_only_files, @@ -66,6 +68,7 @@ mod webui; crate::api::ApiAddTorrentResponse, crate::limits::LimitsConfig, handlers::torrents::UpdateOnlyFilesRequest, + handlers::torrents::AddTrackersRequest, )) )] struct ApiDoc; diff --git a/crates/librtbit/src/session/mod.rs b/crates/librtbit/src/session/mod.rs index a98639c..9cf751b 100644 --- a/crates/librtbit/src/session/mod.rs +++ b/crates/librtbit/src/session/mod.rs @@ -943,7 +943,7 @@ impl Session { id, span, info_hash, - trackers: trackers.into_iter().collect(), + trackers: RwLock::new(trackers.into_iter().collect()), spawner: self.spawner.clone(), peer_id: self.peer_id, storage_factory, @@ -1143,6 +1143,75 @@ impl Session { Ok(()) } + pub async fn force_recheck( + self: &Arc, + handle: &ManagedTorrentHandle, + ) -> anyhow::Result<()> { + let resume_after_check = handle.prepare_force_recheck()?; + let peer_rx = resume_after_check + .then(|| self.make_peer_rx_managed_torrent(handle, true)) + .flatten(); + handle.start(peer_rx, !resume_after_check)?; + self.try_update_persistence_metadata(handle).await; + Ok(()) + } + + pub async fn add_trackers( + self: &Arc, + handle: &ManagedTorrentHandle, + trackers: Vec, + ) -> anyhow::Result<()> { + if trackers.is_empty() { + bail!("at least one tracker URL is required"); + } + let trackers = trackers + .into_iter() + .map(|tracker| { + let url = url::Url::parse(&tracker) + .with_context(|| format!("invalid tracker URL {tracker:?}"))?; + match url.scheme() { + "http" | "https" | "udp" => Ok(url), + scheme => bail!("unsupported tracker URL scheme {scheme:?}"), + } + }) + .collect::>>()?; + + let added = { + let mut configured = handle.shared().trackers.write(); + let mut added = Vec::new(); + for tracker in trackers { + handle.shared().tracker_status.ensure(tracker.as_str()); + if configured.insert(tracker.clone()) { + added.push(tracker); + } + } + added + }; + + self.try_update_persistence_metadata(handle).await; + + if !added.is_empty() + && matches!( + handle.stats().state, + crate::torrent_state::TorrentStatsState::Live + ) + && let Some(peer_rx) = self.make_peer_rx( + handle.info_hash(), + added, + true, + handle.shared().options.force_tracker_interval, + Vec::new(), + handle + .with_metadata(|metadata| metadata.info.info().private) + .unwrap_or(false), + Some(handle.shared().tracker_status.clone()), + ) + { + handle.add_peer_stream(peer_rx)?; + } + Ok(()) + } + pub async fn update_only_files( self: &Arc, handle: &ManagedTorrentHandle, diff --git a/crates/librtbit/src/session/peer_sources.rs b/crates/librtbit/src/session/peer_sources.rs index f6ff6e9..f994d05 100644 --- a/crates/librtbit/src/session/peer_sources.rs +++ b/crates/librtbit/src/session/peer_sources.rs @@ -36,7 +36,7 @@ impl Session { let is_private = t.with_metadata(|m| m.info.info().private).unwrap_or(false); self.make_peer_rx( t.info_hash(), - t.shared().trackers.iter().cloned().collect(), + t.shared().trackers.read().iter().cloned().collect(), announce, t.shared().options.force_tracker_interval, t.shared().options.initial_peers.clone(), diff --git a/crates/librtbit/src/session_persistence/json.rs b/crates/librtbit/src/session_persistence/json.rs index b142bca..6e3296d 100644 --- a/crates/librtbit/src/session_persistence/json.rs +++ b/crates/librtbit/src/session_persistence/json.rs @@ -146,6 +146,7 @@ impl JsonSessionPersistenceStore { trackers: torrent .shared() .trackers + .read() .iter() .map(|u| u.to_string()) .collect(), diff --git a/crates/librtbit/src/session_persistence/mod.rs b/crates/librtbit/src/session_persistence/mod.rs index 4fcc27a..cff21aa 100644 --- a/crates/librtbit/src/session_persistence/mod.rs +++ b/crates/librtbit/src/session_persistence/mod.rs @@ -65,15 +65,13 @@ impl SerializedTorrent { pub fn into_add_torrent(self) -> anyhow::Result<(AddTorrent<'static>, AddTorrentOptions)> { let category = self.category.clone(); + let trackers = self.trackers.into_iter().collect::>(); let add_torrent = if !self.torrent_bytes.is_empty() { AddTorrent::TorrentFileBytes(self.torrent_bytes) } else { - let magnet = Magnet::from_id20( - self.info_hash, - self.trackers.into_iter().collect(), - self.only_files.clone(), - ) - .to_string(); + let magnet = + Magnet::from_id20(self.info_hash, trackers.clone(), self.only_files.clone()) + .to_string(); AddTorrent::from_url(magnet) }; @@ -102,6 +100,7 @@ impl SerializedTorrent { category, added_on: self.added_on, completion_on: self.completion_on, + trackers: Some(trackers), ..Default::default() }; diff --git a/crates/librtbit/src/session_persistence/postgres.rs b/crates/librtbit/src/session_persistence/postgres.rs index 982689b..13b6562 100644 --- a/crates/librtbit/src/session_persistence/postgres.rs +++ b/crates/librtbit/src/session_persistence/postgres.rs @@ -109,6 +109,13 @@ impl SessionPersistenceStore for PostgresSessionStorage { .as_ref() .map(|i| i.torrent_bytes.clone()) .unwrap_or_default(); + let trackers = torrent + .shared() + .trackers + .read() + .iter() + .map(|tracker| tracker.to_string()) + .collect::>(); let q = "INSERT INTO torrents (id, info_hash, torrent_bytes, trackers, output_folder, output_folder_root, only_files, is_paused) VALUES($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT(id) DO NOTHING"; @@ -116,14 +123,7 @@ impl SessionPersistenceStore for PostgresSessionStorage { .bind::(id.try_into()?) .bind(&torrent.info_hash().0[..]) .bind(torrent_bytes.as_ref()) - .bind( - torrent - .shared() - .trackers - .iter() - .map(|t| t.to_string()) - .collect::>(), - ) + .bind(trackers) .bind( torrent .shared() @@ -179,17 +179,27 @@ impl SessionPersistenceStore for PostgresSessionStorage { id: TorrentId, torrent: &ManagedTorrentHandle, ) -> anyhow::Result<()> { - sqlx::query("UPDATE torrents SET only_files = $1, is_paused = $2 WHERE id = $3") - .bind(torrent.only_files().map(|v| { - v.into_iter() - .filter_map(|f| f.try_into().ok()) - .collect::>() - })) - .bind(torrent.is_paused()) - .bind::(id.try_into()?) - .execute(&self.pool) - .await - .context("error executing UPDATE torrents")?; + let trackers = torrent + .shared() + .trackers + .read() + .iter() + .map(|tracker| tracker.to_string()) + .collect::>(); + sqlx::query( + "UPDATE torrents SET trackers = $1, only_files = $2, is_paused = $3 WHERE id = $4", + ) + .bind(trackers) + .bind(torrent.only_files().map(|v| { + v.into_iter() + .filter_map(|f| f.try_into().ok()) + .collect::>() + })) + .bind(torrent.is_paused()) + .bind::(id.try_into()?) + .execute(&self.pool) + .await + .context("error executing UPDATE torrents")?; Ok(()) } diff --git a/crates/librtbit/src/tests/read_only_storage.rs b/crates/librtbit/src/tests/read_only_storage.rs index bcb591f..9e9b433 100644 --- a/crates/librtbit/src/tests/read_only_storage.rs +++ b/crates/librtbit/src/tests/read_only_storage.rs @@ -9,6 +9,7 @@ use crate::{ SessionPersistenceConfig, create_torrent, spawn_utils::BlockingSpawner, storage::{StorageFactoryExt, filesystem::FilesystemStorageFactory}, + tests::{mock_tracker::MockUdpTracker, test_util::wait_until}, }; async fn create_read_only_session( @@ -240,3 +241,228 @@ async fn read_only_session_verifies_single_entry_multi_file_save_path() -> anyho ); Ok(()) } + +#[tokio::test(flavor = "multi_thread")] +async fn force_recheck_hashes_payload_and_keeps_paused_torrent_paused() -> anyhow::Result<()> { + let payload_dir = TempDir::with_prefix("rtbit_recheck_paused_payload")?; + let payload_path = payload_dir.path().join("payload.bin"); + std::fs::write(&payload_path, vec![0x6a; 64 * 1024])?; + let torrent = create_single_file_torrent(payload_dir.path()).await?; + + let session_dir = TempDir::with_prefix("rtbit_recheck_paused_session")?; + let session = create_read_only_session(session_dir.path()).await?; + let handle = session + .add_torrent( + AddTorrent::from_bytes(torrent), + Some(AddTorrentOptions { + paused: true, + overwrite: true, + output_folder: Some(payload_dir.path().to_string_lossy().into_owned()), + ..Default::default() + }), + ) + .await? + .into_handle() + .context("expected a torrent handle")?; + timeout(Duration::from_secs(10), handle.wait_until_initialized()) + .await + .context("timed out waiting for initial validation")??; + assert!(handle.stats().finished); + + std::fs::write(&payload_path, vec![0x7b; 64 * 1024])?; + session.force_recheck(&handle).await?; + assert!(matches!( + handle.stats().state, + crate::torrent_state::TorrentStatsState::Initializing + )); + timeout(Duration::from_secs(10), handle.wait_until_initialized()) + .await + .context("timed out waiting for forced recheck")??; + + let stats = handle.stats(); + assert!(matches!( + stats.state, + crate::torrent_state::TorrentStatsState::Paused + )); + assert!(!stats.finished); + assert_eq!(stats.progress_bytes, 0); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn force_recheck_resumes_torrent_that_was_live() -> anyhow::Result<()> { + let payload_dir = TempDir::with_prefix("rtbit_recheck_live_payload")?; + std::fs::write( + payload_dir.path().join("payload.bin"), + vec![0x8c; 64 * 1024], + )?; + let torrent = create_single_file_torrent(payload_dir.path()).await?; + + let session_dir = TempDir::with_prefix("rtbit_recheck_live_session")?; + let session = create_read_only_session(session_dir.path()).await?; + let handle = session + .add_torrent( + AddTorrent::from_bytes(torrent), + Some(AddTorrentOptions { + overwrite: true, + output_folder: Some(payload_dir.path().to_string_lossy().into_owned()), + ..Default::default() + }), + ) + .await? + .into_handle() + .context("expected a torrent handle")?; + timeout(Duration::from_secs(10), handle.wait_until_initialized()) + .await + .context("timed out waiting for initial validation")??; + assert!(matches!( + handle.stats().state, + crate::torrent_state::TorrentStatsState::Live + )); + + session.force_recheck(&handle).await?; + timeout(Duration::from_secs(10), handle.wait_until_initialized()) + .await + .context("timed out waiting for forced recheck")??; + assert!(matches!( + handle.stats().state, + crate::torrent_state::TorrentStatsState::Live + )); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn added_trackers_are_validated_deduplicated_and_restored() -> anyhow::Result<()> { + let payload_dir = TempDir::with_prefix("rtbit_tracker_restore_payload")?; + std::fs::write( + payload_dir.path().join("payload.bin"), + vec![0x9d; 32 * 1024], + )?; + let torrent = create_single_file_torrent(payload_dir.path()).await?; + + let persistence_dir = TempDir::with_prefix("rtbit_tracker_restore_session")?; + let new_session = || { + Session::new_with_opts( + payload_dir.path().to_owned(), + SessionOptions { + disable_dht: true, + disable_trackers: true, + disable_local_service_discovery: true, + default_storage_factory: Some(FilesystemStorageFactory::read_only().boxed()), + persistence: Some(SessionPersistenceConfig::Json { + folder: Some(persistence_dir.path().to_owned()), + }), + ..Default::default() + }, + ) + }; + + let session = new_session().await?; + let handle = session + .add_torrent( + AddTorrent::from_bytes(torrent), + Some(AddTorrentOptions { + paused: true, + overwrite: true, + output_folder: Some(payload_dir.path().to_string_lossy().into_owned()), + ..Default::default() + }), + ) + .await? + .into_handle() + .context("expected a torrent handle")?; + timeout(Duration::from_secs(10), handle.wait_until_initialized()).await??; + assert!(handle.shared().trackers.read().is_empty()); + + let empty = session.add_trackers(&handle, Vec::new()).await; + assert!(empty.is_err()); + assert!(handle.shared().trackers.read().is_empty()); + let invalid = session + .add_trackers(&handle, vec!["not a tracker".to_string()]) + .await; + assert!(invalid.is_err()); + assert!(handle.shared().trackers.read().is_empty()); + let unsupported = session + .add_trackers( + &handle, + vec!["ftp://tracker.example.test/announce".to_string()], + ) + .await; + assert!(unsupported.is_err()); + assert!(handle.shared().trackers.read().is_empty()); + + let tracker = "udp://tracker.example.test:6969/announce".to_string(); + session + .add_trackers(&handle, vec![tracker.clone(), tracker.clone()]) + .await?; + assert_eq!(handle.shared().trackers.read().len(), 1); + assert_eq!(handle.shared().tracker_status.snapshot().len(), 1); + + let info_hash = handle.info_hash(); + drop(handle); + drop(session); + + let restored = new_session().await?; + let restored_handle = restored + .get(info_hash.into()) + .context("expected persisted torrent")?; + timeout( + Duration::from_secs(10), + restored_handle.wait_until_initialized(), + ) + .await??; + let restored_trackers = restored_handle.shared().trackers.read(); + assert_eq!(restored_trackers.len(), 1); + assert!(restored_trackers.iter().any(|url| url.as_str() == tracker)); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn adding_tracker_to_live_torrent_starts_announcing() -> anyhow::Result<()> { + let tracker = MockUdpTracker::start().await; + let payload_dir = TempDir::with_prefix("rtbit_tracker_live_payload")?; + std::fs::write( + payload_dir.path().join("payload.bin"), + vec![0xae; 32 * 1024], + )?; + let torrent = create_single_file_torrent(payload_dir.path()).await?; + let session_dir = TempDir::with_prefix("rtbit_tracker_live_session")?; + let session = Session::new_with_opts( + session_dir.path().to_owned(), + SessionOptions { + disable_dht: true, + disable_local_service_discovery: true, + ..Default::default() + }, + ) + .await?; + let handle = session + .add_torrent( + AddTorrent::from_bytes(torrent), + Some(AddTorrentOptions { + overwrite: true, + output_folder: Some(payload_dir.path().to_string_lossy().into_owned()), + ..Default::default() + }), + ) + .await? + .into_handle() + .context("expected a torrent handle")?; + timeout(Duration::from_secs(10), handle.wait_until_initialized()).await??; + + session.add_trackers(&handle, vec![tracker.url()]).await?; + wait_until( + || { + anyhow::ensure!( + tracker.stats().announces > 0, + "tracker has not received announce" + ); + Ok(()) + }, + Duration::from_secs(5), + ) + .await?; + + tracker.shutdown().await; + Ok(()) +} diff --git a/crates/librtbit/src/torrent_state/mod.rs b/crates/librtbit/src/torrent_state/mod.rs index 1258cc8..5c8c74e 100644 --- a/crates/librtbit/src/torrent_state/mod.rs +++ b/crates/librtbit/src/torrent_state/mod.rs @@ -183,7 +183,7 @@ pub struct ManagedTorrentShared { pub id: TorrentId, pub info_hash: Id20, pub(crate) spawner: BlockingSpawner, - pub trackers: HashSet, + pub trackers: RwLock>, pub peer_id: Id20, pub span: tracing::Span, pub(crate) options: ManagedTorrentOptions, @@ -496,6 +496,77 @@ impl ManagedTorrent { self.locked.read().paused } + /// Replace the current state with a fresh initializer that ignores any + /// persisted bitfield and hashes every selected piece from storage. + /// Returns whether the torrent was live and should resume after checking. + pub(crate) fn prepare_force_recheck(&self) -> anyhow::Result { + let metadata = self + .metadata + .load_full() + .context("torrent metadata was not loaded")?; + let mut locked = self.locked.write(); + let (files, resume_after_check) = match locked.state.take() { + ManagedTorrentState::Initializing(initializing) => { + locked.state = ManagedTorrentState::Initializing(initializing); + bail!("torrent is already checking") + } + ManagedTorrentState::Live(live) => { + // Cancels all live peer and tracker tasks and returns ownership + // of the existing storage for the checker. + let paused = match live.pause() { + Ok(paused) => paused, + Err(error) => { + locked.state = ManagedTorrentState::Error(anyhow::anyhow!( + "failed to pause torrent for integrity check: {error:#}" + )); + return Err(error); + } + }; + (paused.files, true) + } + ManagedTorrentState::Paused(paused) => (paused.files, false), + ManagedTorrentState::Error(error) => { + match self + .shared + .storage_factory + .create_and_init(self.shared(), &metadata) + { + Ok(files) => (files, false), + Err(create_error) => { + locked.state = ManagedTorrentState::Error(error); + return Err(create_error); + } + } + } + ManagedTorrentState::None => { + locked.state = ManagedTorrentState::None; + bail!("bug: torrent is in empty state") + } + }; + + locked.state = ManagedTorrentState::Initializing(Arc::new(TorrentStateInitializing::new( + self.shared.clone(), + metadata, + locked.only_files.clone(), + files, + true, + ))); + locked.paused = !resume_after_check; + self.state_change_notify.notify_waiters(); + Ok(resume_after_check) + } + + pub(crate) fn add_peer_stream(&self, peer_rx: PeerStream) -> anyhow::Result<()> { + let live = self.live().context("torrent is not live")?; + let weak = Arc::downgrade(&live); + live.spawn( + debug_span!(parent: live.shared.span.clone(), "added_peer_source"), + format!("[{}]added_peer_source", live.shared.id), + async move { drain_peer_stream(&weak, peer_rx).await }, + ); + Ok(()) + } + /// Pause the torrent if it's live or initializing. pub(crate) fn pause(&self) -> anyhow::Result<()> { let mut g = self.locked.write(); @@ -756,7 +827,7 @@ fn spawn_peer_adder(live: &Arc, peer_rx: PeerStream) { let is_private = state.metadata.info.info().private; let new_peer_rx = session.make_peer_rx( state.shared.info_hash, - state.shared.trackers.iter().cloned().collect(), + state.shared.trackers.read().iter().cloned().collect(), true, // announce state.shared.options.force_tracker_interval, Vec::new(), // no initial peers on re-discovery diff --git a/crates/librtbit/webui/e2e/mock-ui.spec.ts b/crates/librtbit/webui/e2e/mock-ui.spec.ts index 506eeb9..15df708 100644 --- a/crates/librtbit/webui/e2e/mock-ui.spec.ts +++ b/crates/librtbit/webui/e2e/mock-ui.spec.ts @@ -69,6 +69,38 @@ test("selects a torrent and exposes safe bulk actions", async ({ page }) => { .toBe(pausedBefore + 1); }); +test("starts an integrity check from the torrent context menu", async ({ + page, +}) => { + const firstRow = page.locator("tbody tr").first(); + await firstRow.click({ button: "right" }); + const recheck = page.getByRole("button", { name: "Force Recheck" }); + await expect(recheck).toBeVisible(); + await recheck.click(); + await expect(firstRow).toContainText("Checking"); +}); + +test("disables recheck while a torrent is already checking", async ({ + page, +}) => { + const firstRow = page.locator("tbody tr").first(); + const torrentName = await firstRow + .locator("td div[title]") + .first() + .getAttribute("title"); + expect(torrentName).toBeTruthy(); + await firstRow.click({ button: "right" }); + await page.getByRole("button", { name: "Force Recheck" }).click(); + const checkingRow = page.locator("tbody tr").filter({ + has: page.getByTitle(torrentName!, { exact: true }), + }); + await expect(checkingRow).toContainText("Checking"); + await checkingRow.click({ button: "right" }); + await expect( + page.getByRole("button", { name: "Force Recheck" }), + ).toBeDisabled(); +}); + test("supports configuration and responsive card layout", async ({ page }) => { await page.getByTitle("Configure").click(); await expect( diff --git a/crates/librtbit/webui/src/api-types.ts b/crates/librtbit/webui/src/api-types.ts index 5a7cd87..3784062 100644 --- a/crates/librtbit/webui/src/api-types.ts +++ b/crates/librtbit/webui/src/api-types.ts @@ -509,6 +509,8 @@ export interface RtbitAPI { pause: (index: number) => Promise; updateOnlyFiles: (index: number, files: number[]) => Promise; start: (index: number) => Promise; + recheck: (index: number) => Promise; + addTrackers: (index: number, trackers: string[]) => Promise; forget: (index: number) => Promise; delete: (index: number) => Promise; stats: () => Promise; diff --git a/crates/librtbit/webui/src/components/TorrentContextMenu.tsx b/crates/librtbit/webui/src/components/TorrentContextMenu.tsx index 9cdcff4..2373b5c 100644 --- a/crates/librtbit/webui/src/components/TorrentContextMenu.tsx +++ b/crates/librtbit/webui/src/components/TorrentContextMenu.tsx @@ -1,5 +1,5 @@ import { useContext, useEffect, useRef, useState } from "react"; -import { TorrentListItem } from "../api-types"; +import { ErrorDetails, TorrentListItem } from "../api-types"; import { APIContext } from "../context"; import { useErrorStore } from "../stores/errorStore"; import { useTorrentStore } from "../stores/torrentStore"; @@ -19,6 +19,7 @@ import { FaAngleDoubleDown, FaTachometerAlt, FaSeedling, + FaSyncAlt, } from "react-icons/fa"; import { BsCheckSquareFill, BsSquare } from "react-icons/bs"; @@ -76,6 +77,7 @@ export const TorrentContextMenu: React.FC = ({ const hasFinished = targets.some((t) => t.stats?.finished); const hasQueueState = targets.some((t) => t.stats?.queue_state != null); + const canRecheck = targets.some((t) => t.stats?.state !== "initializing"); // For single target toggles const isSequential = singleTarget?.stats?.sequential ?? false; @@ -184,6 +186,23 @@ export const TorrentContextMenu: React.FC = ({ onClose(); }; + const handleRecheck = async () => { + setActionInProgress(true); + for (const t of targets) { + if (t.stats?.state === "initializing") continue; + try { + await API.recheck(t.id); + } catch (e: unknown) { + setCloseableError({ + text: `Error checking torrent "${t.name ?? t.id}"`, + details: e as ErrorDetails, + }); + } + } + refreshTorrents(); + onClose(); + }; + const handleCopyName = () => { const text = targets .map((t) => t.name ?? "") @@ -400,6 +419,17 @@ export const TorrentContextMenu: React.FC = ({ {(hasResumable || hasLive) && separator} + + + {separator} + {canConfigure && ( <>