mirror of
https://github.com/nimbold/Firelink.git
synced 2026-09-02 14:07:57 +00:00
fix(torrent): harden RPC and seeding lifecycle
- Retry ambiguous seed resumes without stranding queue permits. - Bound cached Torrent metadata rereads at every native consumer. - Validate Aria2 global-option results and retain upload telemetry during lost-event reconciliation.
This commit is contained in:
+63
-15
@@ -6435,6 +6435,30 @@ fn ensure_aria2_gid_result(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn ensure_aria2_ok_result(method: &str, result: &serde_json::Value) -> Result<(), String> {
|
||||||
|
match result.as_str() {
|
||||||
|
Some("OK") => Ok(()),
|
||||||
|
Some(value) => Err(format!(
|
||||||
|
"aria2.{method} returned unexpected result {value}"
|
||||||
|
)),
|
||||||
|
None => Err(format!("aria2.{method} returned a non-string result")),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const ARIA2_RECONCILIATION_STATUS_FIELDS: &[&str] = &[
|
||||||
|
"status",
|
||||||
|
"seeder",
|
||||||
|
"errorCode",
|
||||||
|
"errorMessage",
|
||||||
|
"completedLength",
|
||||||
|
"totalLength",
|
||||||
|
"uploadLength",
|
||||||
|
];
|
||||||
|
|
||||||
|
fn aria2_reconciliation_status_params(gid: &str) -> serde_json::Value {
|
||||||
|
serde_json::json!([gid, ARIA2_RECONCILIATION_STATUS_FIELDS])
|
||||||
|
}
|
||||||
|
|
||||||
async fn aria2_download_status(port: u16, secret: &str, gid: &str) -> Result<String, String> {
|
async fn aria2_download_status(port: u16, secret: &str, gid: &str) -> Result<String, String> {
|
||||||
let result = rpc_call(
|
let result = rpc_call(
|
||||||
port,
|
port,
|
||||||
@@ -6543,7 +6567,7 @@ async fn reconcile_aria2_downloads(app_handle: &tauri::AppHandle) {
|
|||||||
port,
|
port,
|
||||||
&secret,
|
&secret,
|
||||||
"aria2.tellStatus",
|
"aria2.tellStatus",
|
||||||
serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]),
|
aria2_reconciliation_status_params(&gid),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
@@ -7005,7 +7029,7 @@ async fn validate_torrent_enqueue(
|
|||||||
item.mirrors = (!legacy_web_seeds.is_empty()).then_some(legacy_web_seeds.join("\n"));
|
item.mirrors = (!legacy_web_seeds.is_empty()).then_some(legacy_web_seeds.join("\n"));
|
||||||
if let Some(path) = item.torrent_path.as_deref() {
|
if let Some(path) = item.torrent_path.as_deref() {
|
||||||
let path = crate::torrent::validate_managed_torrent_path(app_handle, &item.id, path)?;
|
let path = crate::torrent::validate_managed_torrent_path(app_handle, &item.id, path)?;
|
||||||
let bytes = std::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes_sync(&path)
|
||||||
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
let normalized_user_web_seeds = queue::normalize_torrent_web_seeds(
|
let normalized_user_web_seeds = queue::normalize_torrent_web_seeds(
|
||||||
@@ -7120,7 +7144,7 @@ fn expected_torrent_output_paths(
|
|||||||
return Ok(None);
|
return Ok(None);
|
||||||
};
|
};
|
||||||
let torrent_path = crate::torrent::validate_managed_torrent_path(app_handle, id, torrent_path)?;
|
let torrent_path = crate::torrent::validate_managed_torrent_path(app_handle, id, torrent_path)?;
|
||||||
let bytes = std::fs::read(torrent_path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes_sync(&torrent_path)
|
||||||
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
let selected = crate::torrent::validate_selected_indices(
|
let selected = crate::torrent::validate_selected_indices(
|
||||||
@@ -7585,7 +7609,7 @@ async fn rekey_torrent_metadata(
|
|||||||
&source.to_string_lossy(),
|
&source.to_string_lossy(),
|
||||||
)
|
)
|
||||||
.map_err(AppError::Internal)?;
|
.map_err(AppError::Internal)?;
|
||||||
let bytes = tokio::fs::read(&source)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&source)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| {
|
.map_err(|error| {
|
||||||
AppError::Internal(format!("could not read cached torrent metadata: {error}"))
|
AppError::Internal(format!("could not read cached torrent metadata: {error}"))
|
||||||
@@ -8237,7 +8261,7 @@ async fn get_torrent_file_selection(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
||||||
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
@@ -8279,7 +8303,7 @@ async fn set_torrent_file_selection(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
||||||
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
@@ -8419,7 +8443,7 @@ async fn get_torrent_details(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
||||||
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
||||||
crate::torrent::torrent_details_from_bytes(&bytes)
|
crate::torrent::torrent_details_from_bytes(&bytes)
|
||||||
@@ -8457,7 +8481,7 @@ async fn get_torrent_magnet_link(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
||||||
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|_| "Torrent metadata is unavailable".to_string())?;
|
.map_err(|_| "Torrent metadata is unavailable".to_string())?;
|
||||||
let details = crate::torrent::torrent_details_from_bytes(&bytes)?;
|
let details = crate::torrent::torrent_details_from_bytes(&bytes)?;
|
||||||
@@ -8510,7 +8534,7 @@ async fn export_torrent_metadata(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
||||||
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|_| "Torrent metadata is unavailable".to_string())?;
|
.map_err(|_| "Torrent metadata is unavailable".to_string())?;
|
||||||
let details = crate::torrent::torrent_details_from_bytes(&bytes)?;
|
let details = crate::torrent::torrent_details_from_bytes(&bytes)?;
|
||||||
@@ -9131,7 +9155,7 @@ async fn move_torrent_data(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
||||||
let torrent_path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, torrent_path)?;
|
let torrent_path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, torrent_path)?;
|
||||||
let bytes = tokio::fs::read(&torrent_path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&torrent_path)
|
||||||
.await
|
.await
|
||||||
.map_err(|_| "Torrent metadata is unavailable".to_string())?;
|
.map_err(|_| "Torrent metadata is unavailable".to_string())?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
@@ -9635,7 +9659,7 @@ async fn verify_torrent_data(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
.ok_or_else(|| "Torrent metadata is not resolved yet".to_string())?;
|
||||||
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
let path = crate::torrent::validate_managed_torrent_path(&app_handle, &id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
@@ -9899,7 +9923,7 @@ async fn normalize_persisted_torrent_web_seeds(
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is unavailable for web-seed management".to_string())?;
|
.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 path = crate::torrent::validate_managed_torrent_path(app_handle, id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| format!("could not read cached Torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached Torrent metadata: {error}"))?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
@@ -10228,7 +10252,7 @@ async fn set_torrent_max_open_files(
|
|||||||
serde_json::json!([{"bt-max-open-files": max_open_files.to_string()}]),
|
serde_json::json!([{"bt-max-open-files": max_open_files.to_string()}]),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
.map(|_| ())
|
.and_then(|result| ensure_aria2_ok_result("changeGlobalOption", &result))
|
||||||
.map_err(|error| format!("Failed to set Torrent maximum open files: {error}"))
|
.map_err(|error| format!("Failed to set Torrent maximum open files: {error}"))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -10248,7 +10272,7 @@ async fn set_torrent_overall_upload_limit(
|
|||||||
serde_json::json!([{"max-overall-upload-limit": limit_str}]),
|
serde_json::json!([{"max-overall-upload-limit": limit_str}]),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
.map(|_| ())
|
.and_then(|result| ensure_aria2_ok_result("changeGlobalOption", &result))
|
||||||
.map_err(|error| format!("Failed to set Torrent overall upload limit: {error}"))
|
.map_err(|error| format!("Failed to set Torrent overall upload limit: {error}"))
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -10274,6 +10298,7 @@ async fn set_global_speed_limit(
|
|||||||
serde_json::json!([{"max-overall-download-limit": limit_str}]),
|
serde_json::json!([{"max-overall-download-limit": limit_str}]),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
.and_then(|result| ensure_aria2_ok_result("changeGlobalOption", &result))
|
||||||
.map(|_| {
|
.map(|_| {
|
||||||
state
|
state
|
||||||
.queue_manager
|
.queue_manager
|
||||||
@@ -11965,6 +11990,29 @@ mod tests {
|
|||||||
assert!(!magnet.contains("passkey"));
|
assert!(!magnet.contains("passkey"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn aria2_global_option_results_require_the_documented_ok_value() {
|
||||||
|
assert!(super::ensure_aria2_ok_result("changeGlobalOption", &json!("OK")).is_ok());
|
||||||
|
let unexpected = super::ensure_aria2_ok_result("changeGlobalOption", &json!("accepted"))
|
||||||
|
.expect_err("unexpected global-option results must not claim success");
|
||||||
|
assert!(unexpected.contains("unexpected result accepted"));
|
||||||
|
let malformed = super::ensure_aria2_ok_result("changeGlobalOption", &json!({}))
|
||||||
|
.expect_err("non-string global-option results must not claim success");
|
||||||
|
assert!(malformed.contains("non-string result"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn aria2_reconciliation_requests_upload_length_for_torrent_telemetry() {
|
||||||
|
let params = super::aria2_reconciliation_status_params("gid-1");
|
||||||
|
let fields = params
|
||||||
|
.get(1)
|
||||||
|
.and_then(serde_json::Value::as_array)
|
||||||
|
.expect("reconciliation params should contain a status field list");
|
||||||
|
assert!(fields
|
||||||
|
.iter()
|
||||||
|
.any(|field| field.as_str() == Some("uploadLength")));
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn torrent_move_root_mapping_handles_single_file_output() {
|
fn torrent_move_root_mapping_handles_single_file_output() {
|
||||||
let root = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
|
let root = tempfile::tempdir_in(std::env::current_dir().unwrap()).unwrap();
|
||||||
@@ -16460,7 +16508,7 @@ pub fn run() {
|
|||||||
poll_port.load(std::sync::atomic::Ordering::Relaxed),
|
poll_port.load(std::sync::atomic::Ordering::Relaxed),
|
||||||
&poll_secret,
|
&poll_secret,
|
||||||
"aria2.tellStatus",
|
"aria2.tellStatus",
|
||||||
serde_json::json!([gid, ["status", "seeder", "errorCode", "errorMessage", "completedLength", "totalLength"]]),
|
aria2_reconciliation_status_params(&gid),
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
{
|
{
|
||||||
|
|||||||
+222
-5
@@ -1517,6 +1517,12 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn seed_starting(&self, id: &str) -> bool {
|
||||||
|
self.seed_capacity
|
||||||
|
.lock()
|
||||||
|
.map_or(false, |state| state.starting.contains(id))
|
||||||
|
}
|
||||||
|
|
||||||
fn add_seed_waiter(&self, waiter: SeedWaiter) {
|
fn add_seed_waiter(&self, waiter: SeedWaiter) {
|
||||||
if let Ok(mut state) = self.seed_capacity.lock() {
|
if let Ok(mut state) = self.seed_capacity.lock() {
|
||||||
if !state.owners.contains(&waiter.id)
|
if !state.owners.contains(&waiter.id)
|
||||||
@@ -1825,14 +1831,113 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
// An unverified unpause must not release either the seed
|
// An unverified unpause must not release either the seed
|
||||||
// owner or its download permit: the daemon may already be
|
// owner or its download permit: the daemon may already be
|
||||||
// active even though the RPC/status check was unavailable.
|
// active even though the RPC/status check was unavailable.
|
||||||
// Keep the item fenced in the starting state until a
|
// Keep the item fenced in the starting state and retry the
|
||||||
// terminal event or an explicit user pause resolves it.
|
// status-verified control until the daemon or a newer
|
||||||
|
// lifecycle resolves it.
|
||||||
log::warn!(
|
log::warn!(
|
||||||
"Torrent seed resume [{}] could not be verified; retaining seed ownership and permit: {}",
|
"Torrent seed resume [{}] could not be verified; retaining seed ownership and permit: {}",
|
||||||
id,
|
id,
|
||||||
error
|
error
|
||||||
);
|
);
|
||||||
self.emit_waiting_to_seed(&id, None);
|
self.emit_waiting_to_seed(&id, None);
|
||||||
|
let retry_waiter = SeedWaiter {
|
||||||
|
id: id.clone(),
|
||||||
|
queue_id: waiter.queue_id.clone(),
|
||||||
|
lifecycle_generation: waiter.lifecycle_generation,
|
||||||
|
};
|
||||||
|
let manager = Arc::clone(&self);
|
||||||
|
tauri::async_runtime::spawn(async move {
|
||||||
|
manager
|
||||||
|
.retry_ambiguous_seed_resume(retry_waiter, gid, epoch)
|
||||||
|
.await;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn retry_ambiguous_seed_resume(
|
||||||
|
self: Arc<Self>,
|
||||||
|
waiter: SeedWaiter,
|
||||||
|
gid: String,
|
||||||
|
epoch: u64,
|
||||||
|
) {
|
||||||
|
let mut delay = Duration::from_millis(250);
|
||||||
|
loop {
|
||||||
|
tokio::select! {
|
||||||
|
_ = tokio::time::sleep(delay) => {}
|
||||||
|
_ = self.notify.notified() => {}
|
||||||
|
}
|
||||||
|
|
||||||
|
let _control_guard = self.acquire_aria2_control(&waiter.id).await;
|
||||||
|
let lifecycle_current = self
|
||||||
|
.is_registered_generation_or_legacy(&waiter.id, waiter.lifecycle_generation)
|
||||||
|
.await
|
||||||
|
&& self.seed_starting(&waiter.id)
|
||||||
|
&& matches!(self.active_kind(&waiter.id).await, Some(TaskKind::Aria2))
|
||||||
|
&& self.has_active_permit(&waiter.id).await
|
||||||
|
&& self.aria2_gid_for_download(&waiter.id).as_deref() == Some(gid.as_str())
|
||||||
|
&& self.is_aria2_control_epoch_current(&waiter.id, epoch).await
|
||||||
|
&& self.is_current_aria2_gid_mapping(
|
||||||
|
&gid,
|
||||||
|
&Aria2GidMapping {
|
||||||
|
id: waiter.id.clone(),
|
||||||
|
epoch,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
if !lifecycle_current {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
match self.spawner.resume_for_seed(&gid).await {
|
||||||
|
Ok(Aria2SeedControlOutcome::Resumed) => {
|
||||||
|
if !self.is_aria2_control_epoch_current(&waiter.id, epoch).await
|
||||||
|
|| !self.is_current_aria2_gid_mapping(
|
||||||
|
&gid,
|
||||||
|
&Aria2GidMapping {
|
||||||
|
id: waiter.id.clone(),
|
||||||
|
epoch,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
{
|
||||||
|
let _ = self.spawner.pause_for_seed(&gid).await;
|
||||||
|
let _ = self.spawner.pause_for_seed(&gid).await;
|
||||||
|
self.release_download_permit_for_seed(&waiter.id).await;
|
||||||
|
self.abandon_seed_start(&waiter.id);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
self.record_seed_started(&waiter.id).await;
|
||||||
|
self.release_download_permit_for_seed(&waiter.id).await;
|
||||||
|
self.finish_seed_start(&waiter.id);
|
||||||
|
self.emit_state(&waiter.id, DownloadStatus::Seeding);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Ok(Aria2SeedControlOutcome::Complete) => {
|
||||||
|
self.finish_seed_start(&waiter.id);
|
||||||
|
self.apply_completion_locked(&waiter.id, PendingOutcome::Complete)
|
||||||
|
.await;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Ok(Aria2SeedControlOutcome::Paused) => {
|
||||||
|
let remaining = self.capture_seed_remaining(&waiter.id).await;
|
||||||
|
self.release_download_permit_for_seed(&waiter.id).await;
|
||||||
|
self.finish_seed_start(&waiter.id);
|
||||||
|
self.abandon_seed_start(&waiter.id);
|
||||||
|
self.add_seed_waiter(SeedWaiter {
|
||||||
|
id: waiter.id.clone(),
|
||||||
|
queue_id: waiter.queue_id.clone(),
|
||||||
|
lifecycle_generation: waiter.lifecycle_generation,
|
||||||
|
});
|
||||||
|
self.emit_waiting_to_seed(&waiter.id, remaining);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
Err(error) => {
|
||||||
|
log::warn!(
|
||||||
|
"Torrent seed resume [{}] remains unverified; retaining seed ownership and permit: {}",
|
||||||
|
waiter.id,
|
||||||
|
error
|
||||||
|
);
|
||||||
|
delay = (delay * 2).min(Duration::from_secs(5));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -2333,7 +2438,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
.as_deref()
|
.as_deref()
|
||||||
.ok_or_else(|| "Torrent metadata is unavailable for web-seed management".to_string())?;
|
.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 path = crate::torrent::validate_managed_torrent_path(&self.app_handle, id, path)?;
|
||||||
let bytes = tokio::fs::read(path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| format!("could not read cached Torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached Torrent metadata: {error}"))?;
|
||||||
crate::torrent::parse_torrent_bytes(&bytes)
|
crate::torrent::parse_torrent_bytes(&bytes)
|
||||||
@@ -3178,7 +3283,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
|
|||||||
id,
|
id,
|
||||||
torrent_path,
|
torrent_path,
|
||||||
)?;
|
)?;
|
||||||
let bytes = tokio::fs::read(&torrent_path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&torrent_path)
|
||||||
.await
|
.await
|
||||||
.map_err(|_| "live Torrent file progress is unavailable".to_string())?;
|
.map_err(|_| "live Torrent file progress is unavailable".to_string())?;
|
||||||
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
let metadata = crate::torrent::parse_torrent_bytes(&bytes)?;
|
||||||
@@ -7815,7 +7920,7 @@ impl SidecarSpawner for ProductionSpawner {
|
|||||||
id,
|
id,
|
||||||
path,
|
path,
|
||||||
)?;
|
)?;
|
||||||
let bytes = tokio::fs::read(&path)
|
let bytes = crate::torrent::read_bounded_torrent_bytes(&path)
|
||||||
.await
|
.await
|
||||||
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
.map_err(|error| format!("could not read cached torrent metadata: {error}"))?;
|
||||||
let (sanitized_bytes, embedded_web_seeds) =
|
let (sanitized_bytes, embedded_web_seeds) =
|
||||||
@@ -8902,6 +9007,42 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
struct AmbiguousResumeSeedSpawner {
|
||||||
|
resume_calls: Arc<AtomicUsize>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait::async_trait]
|
||||||
|
impl SidecarSpawner for AmbiguousResumeSeedSpawner {
|
||||||
|
async fn add_uri(&self, id: &str, _payload: &SpawnPayload) -> Result<String, String> {
|
||||||
|
Ok(format!("gid-{id}"))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn remove_uri(&self, _gid: &str) -> Result<(), String> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn pause_for_seed(&self, _gid: &str) -> Result<Aria2SeedControlOutcome, String> {
|
||||||
|
Ok(Aria2SeedControlOutcome::Paused)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn resume_for_seed(&self, _gid: &str) -> Result<Aria2SeedControlOutcome, String> {
|
||||||
|
if self.resume_calls.fetch_add(1, Ordering::SeqCst) == 0 {
|
||||||
|
Err("seed resume status verification unavailable".to_string())
|
||||||
|
} else {
|
||||||
|
Ok(Aria2SeedControlOutcome::Resumed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn run_media(
|
||||||
|
&self,
|
||||||
|
_id: &str,
|
||||||
|
_payload: &SpawnPayload,
|
||||||
|
_lifecycle_generation: u64,
|
||||||
|
) -> Result<(), String> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn aria2_connection_options_enable_requested_ranges_for_small_release_assets() {
|
fn aria2_connection_options_enable_requested_ranges_for_small_release_assets() {
|
||||||
let mut options = serde_json::Map::new();
|
let mut options = serde_json::Map::new();
|
||||||
@@ -10871,6 +11012,82 @@ mod tests {
|
|||||||
assert_eq!(manager.current_aria2_control_epoch("second").await, 1);
|
assert_eq!(manager.current_aria2_control_epoch("second").await, 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn ambiguous_seed_resume_retries_without_stranding_the_seed_permit() {
|
||||||
|
let app = tauri::test::mock_builder()
|
||||||
|
.build(tauri::test::mock_context(tauri::test::noop_assets()))
|
||||||
|
.expect("mock app");
|
||||||
|
let spawner = Arc::new(AmbiguousResumeSeedSpawner {
|
||||||
|
resume_calls: Arc::new(AtomicUsize::new(0)),
|
||||||
|
});
|
||||||
|
let manager = Arc::new(QueueManager::test_new(
|
||||||
|
app.handle().clone(),
|
||||||
|
1,
|
||||||
|
spawner.clone(),
|
||||||
|
));
|
||||||
|
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;
|
||||||
|
manager
|
||||||
|
.apply_completion(id, PendingOutcome::Seeding)
|
||||||
|
.await;
|
||||||
|
}
|
||||||
|
|
||||||
|
assert!(manager.is_waiting_to_seed("second"));
|
||||||
|
manager.release_registered_id("first").await;
|
||||||
|
assert!(manager.try_start_waiting_seed().await);
|
||||||
|
|
||||||
|
let first_attempt = tokio::time::timeout(Duration::from_secs(1), async {
|
||||||
|
loop {
|
||||||
|
if spawner.resume_calls.load(Ordering::SeqCst) >= 1
|
||||||
|
&& manager.has_active_permit("second").await
|
||||||
|
&& manager.is_waiting_to_seed("second")
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
tokio::task::yield_now().await;
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
assert!(first_attempt.is_ok(), "ambiguous seed resume was not retained safely");
|
||||||
|
|
||||||
|
manager.wake_seed_waiters();
|
||||||
|
let recovered = tokio::time::timeout(Duration::from_secs(1), async {
|
||||||
|
loop {
|
||||||
|
if spawner.resume_calls.load(Ordering::SeqCst) >= 2
|
||||||
|
&& manager.is_seed_owner("second")
|
||||||
|
&& !manager.is_waiting_to_seed("second")
|
||||||
|
&& !manager.has_active_permit("second").await
|
||||||
|
{
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
tokio::task::yield_now().await;
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
assert!(
|
||||||
|
recovered.is_ok(),
|
||||||
|
"ambiguous seed resume did not reconcile after the daemon became reachable"
|
||||||
|
);
|
||||||
|
assert_eq!(manager.current_aria2_control_epoch("second").await, 1);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn unexpected_seeding_outcome_without_policy_is_reconciled_as_complete() {
|
async fn unexpected_seeding_outcome_without_policy_is_reconciled_as_complete() {
|
||||||
let app = tauri::test::mock_builder()
|
let app = tauri::test::mock_builder()
|
||||||
|
|||||||
@@ -1,5 +1,6 @@
|
|||||||
use sha1::{Digest, Sha1};
|
use sha1::{Digest, Sha1};
|
||||||
use std::collections::{BTreeMap, HashSet};
|
use std::collections::{BTreeMap, HashSet};
|
||||||
|
use std::io::Read;
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
use tauri::Manager;
|
use tauri::Manager;
|
||||||
@@ -788,7 +789,8 @@ pub fn inspect_source(source: &str) -> Result<ParsedTorrent, String> {
|
|||||||
return magnet_metadata(source.trim());
|
return magnet_metadata(source.trim());
|
||||||
}
|
}
|
||||||
let path = local_torrent_path(source)?;
|
let path = local_torrent_path(source)?;
|
||||||
let bytes = std::fs::read(&path).map_err(|error| format!("could not read torrent file: {error}"))?;
|
let bytes = read_bounded_torrent_bytes_sync(&path)
|
||||||
|
.map_err(|error| format!("could not read torrent file: {error}"))?;
|
||||||
parse_torrent_bytes(&bytes)
|
parse_torrent_bytes(&bytes)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1148,7 +1150,21 @@ fn is_canonical_torrent_temp_file(name: &str) -> bool {
|
|||||||
&& temporary_id.bytes().all(|byte| byte.is_ascii_hexdigit())
|
&& temporary_id.bytes().all(|byte| byte.is_ascii_hexdigit())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn read_bounded_torrent_bytes(path: &Path) -> std::io::Result<Vec<u8>> {
|
pub(crate) fn read_bounded_torrent_bytes_sync(path: &Path) -> std::io::Result<Vec<u8>> {
|
||||||
|
let file = std::fs::File::open(path)?;
|
||||||
|
let mut bytes = Vec::with_capacity(std::cmp::min(MAX_TORRENT_BYTES, 64 * 1024));
|
||||||
|
file.take((MAX_TORRENT_BYTES + 1) as u64)
|
||||||
|
.read_to_end(&mut bytes)?;
|
||||||
|
if bytes.len() > MAX_TORRENT_BYTES {
|
||||||
|
return Err(std::io::Error::new(
|
||||||
|
std::io::ErrorKind::InvalidData,
|
||||||
|
format!("torrent metadata exceeds {MAX_TORRENT_BYTES} bytes"),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
Ok(bytes)
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) async fn read_bounded_torrent_bytes(path: &Path) -> std::io::Result<Vec<u8>> {
|
||||||
let file = tokio::fs::File::open(path).await?;
|
let file = tokio::fs::File::open(path).await?;
|
||||||
let mut bytes = Vec::with_capacity(std::cmp::min(MAX_TORRENT_BYTES, 64 * 1024));
|
let mut bytes = Vec::with_capacity(std::cmp::min(MAX_TORRENT_BYTES, 64 * 1024));
|
||||||
file.take((MAX_TORRENT_BYTES + 1) as u64)
|
file.take((MAX_TORRENT_BYTES + 1) as u64)
|
||||||
@@ -1687,6 +1703,19 @@ mod tests {
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn bounded_local_torrent_reads_reject_oversized_files() {
|
||||||
|
let temporary = tempfile::tempdir().expect("temporary torrent directory should exist");
|
||||||
|
let path = temporary.path().join("oversized.torrent");
|
||||||
|
let file = std::fs::File::create(&path).expect("oversized torrent fixture should exist");
|
||||||
|
file.set_len((MAX_TORRENT_BYTES + 1) as u64)
|
||||||
|
.expect("oversized torrent fixture should be sparse-writable");
|
||||||
|
|
||||||
|
let error = read_bounded_torrent_bytes_sync(&path)
|
||||||
|
.expect_err("oversized torrent metadata must be rejected before parsing");
|
||||||
|
assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn rejects_noncanonical_lengths_and_invalid_files_field() {
|
fn rejects_noncanonical_lengths_and_invalid_files_field() {
|
||||||
assert!(parse_torrent_bytes(b"d4:infod6:lengthi5e4:name04:testee").is_err());
|
assert!(parse_torrent_bytes(b"d4:infod6:lengthi5e4:name04:testee").is_err());
|
||||||
|
|||||||
Reference in New Issue
Block a user