feat(torrents): add peer diagnostics

This commit is contained in:
NimBold
2026-08-02 01:04:04 +03:30
parent 2474d2c1cb
commit ab9f0507d0
14 changed files with 457 additions and 6 deletions
+25
View File
@@ -202,6 +202,31 @@ pub struct DownloadItem {
pub torrent_stop_timeout: Option<u32>,
}
#[derive(Clone, Debug, Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
pub struct TorrentPeer {
#[ts(type = "number")]
pub download_speed: u64,
#[ts(type = "number")]
pub upload_speed: u64,
pub seeder: bool,
pub am_choking: bool,
pub peer_choking: bool,
}
#[derive(Clone, Debug, Serialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
pub struct TorrentPeerDiagnostics {
#[ts(type = "number")]
pub total_peers: u32,
#[ts(type = "number")]
pub total_seeders: u32,
pub peers: Vec<TorrentPeer>,
pub truncated: bool,
}
#[derive(Clone, Debug, Serialize, Deserialize, TS)]
#[serde(rename_all = "camelCase")]
#[ts(export, export_to = "../../src/bindings/")]
+9 -1
View File
@@ -6452,6 +6452,14 @@ async fn set_torrent_peer_options(
.await
}
#[tauri::command]
async fn get_torrent_peers(
state: tauri::State<'_, AppState>,
id: String,
) -> Result<crate::ipc::TorrentPeerDiagnostics, String> {
state.queue_manager.get_aria2_torrent_peers(&id).await
}
pub(crate) fn normalize_speed_limit_for_aria2(limit: &str) -> Option<String> {
let trimmed = limit.trim();
if trimmed.is_empty() {
@@ -10884,7 +10892,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, 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, 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,
+190
View File
@@ -965,6 +965,72 @@ impl<R: tauri::Runtime> QueueManager<R> {
Ok(())
}
/// Return redacted, bounded peer diagnostics for the current Torrent GID.
/// The control lock and post-RPC mapping check prevent a late response from
/// being attributed to a replaced or terminal lifecycle.
pub async fn get_aria2_torrent_peers(
&self,
id: &str,
) -> Result<crate::ipc::TorrentPeerDiagnostics, String> {
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("download is not an active aria2 transfer".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(|| "active Torrent transfer has no gid".to_string())?;
let expected_mapping = self
.aria2_gid_mapping(&gid)
.ok_or_else(|| "active Torrent transfer has no current gid mapping".to_string())?;
if expected_mapping.id != id
|| !self
.is_aria2_control_epoch_current(id, expected_mapping.epoch)
.await
{
return Err("active Torrent transfer has a stale control epoch".to_string());
}
let state = self.app_handle.state::<crate::AppState>();
let result = crate::rpc_call(
state.aria2_port.load(std::sync::atomic::Ordering::Relaxed),
&state.aria2_secret,
"aria2.getPeers",
serde_json::json!([gid]),
)
.await
.map_err(|error| {
format!(
"aria2.getPeers failed: {}",
crate::redact_sensitive_text(&error)
)
})?;
let diagnostics = parse_torrent_peer_diagnostics(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 peer diagnostics".to_string());
}
Ok(diagnostics)
}
/// Pop the next task, or None if empty.
pub async fn pop_front(&self) -> Option<QueuedTask> {
self.pending.lock().await.pop_front()
@@ -3050,6 +3116,8 @@ const ARIA2_DEFAULT_TORRENT_MAX_PEERS: u32 = 55;
const ARIA2_DEFAULT_TORRENT_PEER_SPEED_LIMIT: &str = "50K";
const MAX_TORRENT_MAX_PEERS: u32 = 1000;
pub(crate) const MAX_TORRENT_STOP_TIMEOUT: u32 = 7 * 24 * 60 * 60;
pub(crate) const MAX_TORRENT_PEER_DIAGNOSTICS: usize = 128;
const MAX_TORRENT_PEER_RESPONSE: usize = 4096;
const MAX_TORRENT_TRACKERS: usize = 64;
const MAX_TORRENT_TRACKER_BYTES: usize = 16 * 1024;
@@ -3123,6 +3191,77 @@ pub(crate) fn normalize_torrent_stop_timeout(value: Option<u32>) -> Result<Optio
Ok(Some(value))
}
fn aria2_peer_number(value: Option<&serde_json::Value>) -> u64 {
match value {
Some(serde_json::Value::String(value)) => value.parse().unwrap_or_default(),
Some(serde_json::Value::Number(value)) => value.as_u64().unwrap_or_default(),
_ => 0,
}
}
fn aria2_peer_bool(value: Option<&serde_json::Value>) -> bool {
match value {
Some(serde_json::Value::Bool(value)) => *value,
Some(serde_json::Value::String(value)) => {
value.eq_ignore_ascii_case("true") || value == "1"
}
_ => false,
}
}
pub(crate) fn parse_torrent_peer_diagnostics(
result: serde_json::Value,
) -> Result<crate::ipc::TorrentPeerDiagnostics, String> {
let peers = result
.as_array()
.ok_or_else(|| "aria2.getPeers returned a non-array result".to_string())?;
if peers.len() > MAX_TORRENT_PEER_RESPONSE {
return Err("aria2.getPeers returned too many peers".to_string());
}
if peers.iter().any(|peer| !peer.is_object()) {
return Err("aria2.getPeers returned malformed peer data".to_string());
}
let total_peers = u32::try_from(peers.len()).unwrap_or(u32::MAX);
let mut total_seeders = 0u32;
let mut sanitized = Vec::with_capacity(peers.len().min(MAX_TORRENT_PEER_DIAGNOSTICS));
for peer in peers.iter().take(MAX_TORRENT_PEER_DIAGNOSTICS) {
let peer = peer
.as_object()
.ok_or_else(|| "aria2.getPeers returned malformed peer data".to_string())?;
let seeder = aria2_peer_bool(peer.get("seeder"));
if seeder {
total_seeders = total_seeders.saturating_add(1);
}
sanitized.push(crate::ipc::TorrentPeer {
download_speed: aria2_peer_number(peer.get("downloadSpeed")),
upload_speed: aria2_peer_number(peer.get("uploadSpeed")),
seeder,
am_choking: aria2_peer_bool(peer.get("amChoking")),
peer_choking: aria2_peer_bool(peer.get("peerChoking")),
});
}
// Count seeders beyond the display cap without retaining any identifying
// peer data. The response is bounded by Aria2's per-Torrent peer limit,
// while the UI receives at most MAX_TORRENT_PEER_DIAGNOSTICS rows.
for peer in peers.iter().skip(MAX_TORRENT_PEER_DIAGNOSTICS) {
let peer = peer
.as_object()
.ok_or_else(|| "aria2.getPeers returned malformed peer data".to_string())?;
if aria2_peer_bool(peer.get("seeder")) {
total_seeders = total_seeders.saturating_add(1);
}
}
Ok(crate::ipc::TorrentPeerDiagnostics {
total_peers,
total_seeders,
peers: sanitized,
truncated: peers.len() > MAX_TORRENT_PEER_DIAGNOSTICS,
})
}
pub(crate) fn normalize_torrent_trackers(value: Option<&str>) -> Result<Option<String>, String> {
let Some(raw) = value.map(str::trim).filter(|value| !value.is_empty()) else {
return Ok(None);
@@ -4210,6 +4349,57 @@ mod tests {
assert!(!options.contains_key("bt-stop-timeout"));
}
#[test]
fn torrent_peer_diagnostics_are_redacted_and_bounded() {
let mut result = vec![serde_json::json!({
"peerId": "secret-peer-id",
"ip": "192.0.2.10",
"port": "6881",
"bitfield": "ffffffff",
"downloadSpeed": "10602",
"uploadSpeed": "6890",
"seeder": "true",
"amChoking": "false",
"peerChoking": "true"
})];
result.extend((1..MAX_TORRENT_PEER_DIAGNOSTICS + 2).map(|index| {
serde_json::json!({
"peerId": format!("peer-{index}"),
"ip": format!("192.0.2.{index}"),
"downloadSpeed": index.to_string(),
"uploadSpeed": 0,
"seeder": index == MAX_TORRENT_PEER_DIAGNOSTICS + 1,
"amChoking": false,
"peerChoking": false
})
}));
let diagnostics = parse_torrent_peer_diagnostics(serde_json::Value::Array(result)).unwrap();
assert_eq!(diagnostics.total_peers, (MAX_TORRENT_PEER_DIAGNOSTICS + 2) as u32);
assert_eq!(diagnostics.total_seeders, 2);
assert_eq!(diagnostics.peers.len(), MAX_TORRENT_PEER_DIAGNOSTICS);
assert!(diagnostics.truncated);
let serialized = serde_json::to_string(&diagnostics).unwrap();
assert!(!serialized.contains("peerId"));
assert!(!serialized.contains("192.0.2."));
assert!(!serialized.contains("bitfield"));
}
#[test]
fn torrent_peer_diagnostics_reject_non_array_results() {
let error = parse_torrent_peer_diagnostics(serde_json::json!({"peers": []})).unwrap_err();
assert!(error.contains("non-array"));
}
#[test]
fn torrent_peer_diagnostics_reject_malformed_peer_entries() {
let error = parse_torrent_peer_diagnostics(serde_json::json!([{
"seeder": true
}, "not-a-peer"]))
.unwrap_err();
assert!(error.contains("malformed"));
}
#[test]
fn enqueue_item_carries_torrent_trackers_into_the_spawn_payload() {
let item: EnqueueItem = serde_json::from_value(serde_json::json!({