From 1cfda45334976460de925f4ba54c5fbd8f8024ff Mon Sep 17 00:00:00 2001 From: NimBold Date: Tue, 8 Sep 2026 10:40:57 +0330 Subject: [PATCH] fix(download): recover duplicate ownership and fail fast on aria2 startup - Repair ownership-backed and malformed download rows from migration backups. - Preserve nonterminal removal intent and completed tombstones during recovery. - Reject unrelated backup names and keep incomplete repairs retryable. - Surface known Aria2 startup failures before range probes and retry loopback RPC port conflicts. - Add regression coverage for duplicate recovery and startup failure paths. --- src-tauri/src/db.rs | 578 +++++++++++++++++++++++++++++++++++++---- src-tauri/src/lib.rs | 62 ++++- src-tauri/src/queue.rs | 69 ++++- 3 files changed, 655 insertions(+), 54 deletions(-) diff --git a/src-tauri/src/db.rs b/src-tauri/src/db.rs index 219fbe9..fb57736 100644 --- a/src-tauri/src/db.rs +++ b/src-tauri/src/db.rs @@ -329,10 +329,14 @@ fn recover_downloads_from_migration_backup( if !table_exists(connection, "metadata")? { return Ok(()); } - if metadata_exists(connection, RECOVERY_MARKER)? { - return Ok(()); - } - + let recovery_status = connection + .query_row( + "SELECT value FROM metadata WHERE key = ?1", + params![RECOVERY_MARKER], + |row| row.get::<_, String>(0), + ) + .optional() + .map_err(|error| format!("failed to read recovery status: {error}"))?; if !table_exists(connection, "downloads")? { return Ok(()); } @@ -340,24 +344,148 @@ fn recover_downloads_from_migration_backup( let current_downloads_count: i64 = connection .query_row("SELECT COUNT(*) FROM downloads", [], |row| row.get(0)) .map_err(|error| format!("failed to count downloads for recovery: {error}"))?; - if current_downloads_count > 0 { + + let mut malformed_download_ids = std::collections::HashSet::new(); + { + let mut statement = connection + .prepare("SELECT id, data FROM downloads") + .map_err(|error| format!("failed to inspect downloads for recovery: {error}"))?; + let rows = statement + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + }) + .map_err(|error| format!("failed to query downloads for recovery: {error}"))?; + for row in rows { + let (id, data) = row + .map_err(|error| format!("failed to read download for recovery: {error}"))?; + let valid_record = serde_json::from_str::(&data) + .ok() + .and_then(|value| value.get("id").and_then(Value::as_str).map(str::to_owned)) + .is_some_and(|stored_id| stored_id == id && !stored_id.is_empty()); + if !valid_record { + malformed_download_ids.insert(id); + } + } + } + + // A previous startup could have persisted a partial renderer snapshot: + // the downloads table was replaced, but the native ownership registry was + // deliberately retained. In that state an owned target is correctly + // protected from replacement, yet its download row is invisible to the + // renderer and the duplicate modal has no safe Replace action. Repair + // only IDs with current durable evidence: ownership-backed missing rows, + // nonterminal removal jobs, or an existing row whose persisted document + // is malformed. They must not be covered by a completed removal job. + // Never reconstruct an unknown ID from a backup without one of those + // durable records. + // A completed full recovery does not make the ownership registry + // self-healing: a later stale renderer snapshot can still remove rows + // while leaving native ownership intact. Any existing marker therefore + // suppresses another unscoped full restore, but still permits this narrow + // ownership-backed repair pass. + let has_removal_jobs_table = table_exists(connection, "download_removal_jobs")?; + let has_download_ownership_table = table_exists(connection, "download_ownership")?; + let ownership_count: i64 = if has_download_ownership_table { connection - .execute( - "INSERT INTO metadata (key, value) VALUES (?1, 'skipped') - ON CONFLICT(key) DO UPDATE SET value = excluded.value", - params![RECOVERY_MARKER], + .query_row("SELECT COUNT(*) FROM download_ownership", [], |row| { + row.get(0) + }) + .map_err(|error| format!("failed to count download ownership for recovery: {error}"))? + } else { + 0 + }; + let mut tombstoned_removal_ids = std::collections::HashSet::new(); + let mut recoverable_removal_ids = std::collections::HashSet::new(); + if has_removal_jobs_table { + let mut statement = connection + .prepare("SELECT id, data FROM download_removal_jobs") + .map_err(|error| format!("failed to read removal jobs for recovery: {error}"))?; + let rows = statement + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + }) + .map_err(|error| format!("failed to query removal jobs for recovery: {error}"))?; + for row in rows { + let (id, data) = row + .map_err(|error| format!("failed to read removal job for recovery: {error}"))?; + let nonterminal = serde_json::from_str::(&data) + .ok() + .and_then(|value| value.get("phase").and_then(Value::as_str).map(str::to_owned)) + .is_some_and(|phase| matches!(phase.as_str(), "pending" | "running" | "failed")); + // Pending, running, and failed jobs retain recoverable download + // intent. Completed jobs are the durable deletion tombstone. An + // invalid or unknown phase remains protected conservatively. + if nonterminal { + recoverable_removal_ids.insert(id); + } else { + tombstoned_removal_ids.insert(id); + } + } + } + let orphan_recovery = current_downloads_count > 0 + || recovery_status.is_some() + || ownership_count > 0 + || !malformed_download_ids.is_empty() + || !recoverable_removal_ids.is_empty(); + let mut orphan_ids = std::collections::HashSet::new(); + if orphan_recovery && has_download_ownership_table { + let query = "SELECT ownership.id + FROM download_ownership AS ownership + LEFT JOIN downloads AS downloads ON downloads.id = ownership.id + WHERE downloads.id IS NULL"; + let mut statement = connection + .prepare(query) + .map_err(|error| format!("failed to find orphaned download ownership: {error}"))?; + let rows = statement + .query_map([], |row| row.get::<_, String>(0)) + .map_err(|error| format!("failed to query orphaned download ownership: {error}"))?; + for row in rows { + let id = row + .map_err(|error| format!("failed to read orphaned download ownership: {error}"))?; + if !tombstoned_removal_ids.contains(&id) { + orphan_ids.insert(id); + } + } + } + for id in malformed_download_ids { + if !tombstoned_removal_ids.contains(&id) { + orphan_ids.insert(id); + } + } + for id in recoverable_removal_ids { + let download_exists: bool = connection + .query_row( + "SELECT EXISTS(SELECT 1 FROM downloads WHERE id = ?1)", + [&id], + |row| row.get(0), ) - .map_err(|error| format!("failed to record recovery status: {error}"))?; + .map_err(|error| format!("failed to check removal download for recovery: {error}"))?; + if !download_exists { + orphan_ids.insert(id); + } + } + + if orphan_recovery && orphan_ids.is_empty() { + if current_downloads_count > 0 && recovery_status.is_none() { + connection + .execute( + "INSERT INTO metadata (key, value) VALUES (?1, 'skipped') + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + params![RECOVERY_MARKER], + ) + .map_err(|error| format!("failed to record recovery status: {error}"))?; + } return Ok(()); } + let backup_prefix = format!("{DATABASE_NAME}.backup-schema-v3-"); let entries = fs::read_dir(app_data_dir) .map_err(|error| format!("failed to read app data directory for recovery: {error}"))?; let mut backup_candidates: Vec = Vec::new(); for entry in entries.flatten() { let path = entry.path(); if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) { - if file_name.starts_with(&format!("{DATABASE_NAME}.backup-schema-v3")) { + if file_name.starts_with(&backup_prefix) { if let Ok(metadata) = fs::symlink_metadata(&path) { if metadata.is_file() && !metadata.file_type().is_symlink() { backup_candidates.push(path); @@ -368,6 +496,8 @@ fn recover_downloads_from_migration_backup( } backup_candidates.sort_by(|a, b| b.cmp(a)); + let mut processed_valid_candidate = false; + let mut total_restored_count = 0; for candidate in backup_candidates { let Ok(backup_conn) = Connection::open(&candidate) else { continue; @@ -391,6 +521,18 @@ fn recover_downloads_from_migration_backup( ); continue; }; + if !backup_downloads.iter().any(|data| { + serde_json::from_str::(data) + .ok() + .and_then(|value| value.get("id").and_then(Value::as_str).map(str::to_owned)) + .is_some_and(|id| !id.is_empty()) + }) { + log::warn!( + "Migration backup candidate '{}' contains no valid download records", + candidate.display() + ); + continue; + } if portable { if let Err(error) = sanitize_download_strings(&mut backup_downloads) { log::warn!( @@ -405,39 +547,48 @@ fn recover_downloads_from_migration_backup( .transaction() .map_err(|error| format!("failed to begin recovery transaction: {error}"))?; - let has_removal_jobs_table = table_exists(&transaction, "download_removal_jobs").unwrap_or(false); let mut restored_ids = std::collections::HashSet::new(); + let mut restored_queue_ids = std::collections::HashSet::new(); let mut restored_count = 0; for data in &backup_downloads { let Ok(value) = serde_json::from_str::(data) else { continue; }; - let Some(id) = value.get("id").and_then(Value::as_str) else { + let Some(id) = value + .get("id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + else { continue; }; - if has_removal_jobs_table - && transaction - .query_row( - "SELECT EXISTS(SELECT 1 FROM download_removal_jobs WHERE id = ?1)", - [id], - |row| row.get::<_, bool>(0), - ) - .unwrap_or(false) - { + if tombstoned_removal_ids.contains(id) { + continue; + } + if orphan_recovery && !orphan_ids.contains(id) { continue; } - let status = value.get("status").and_then(Value::as_str).unwrap_or("completed"); + let status = value + .get("status") + .and_then(Value::as_str) + .unwrap_or("completed"); let queue_id = value.get("queueId").and_then(Value::as_str); + let upsert = if orphan_recovery && orphan_ids.contains(id) { + "INSERT INTO downloads (id, status, queue_id, data) VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(id) DO UPDATE SET status = excluded.status, + queue_id = excluded.queue_id, data = excluded.data" + } else { + "INSERT INTO downloads (id, status, queue_id, data) VALUES (?1, ?2, ?3, ?4) + ON CONFLICT(id) DO NOTHING" + }; let inserted = transaction - .execute( - "INSERT INTO downloads (id, status, queue_id, data) VALUES (?1, ?2, ?3, ?4) - ON CONFLICT(id) DO NOTHING", - params![id, status, queue_id, data], - ) + .execute(upsert, params![id, status, queue_id, data]) .map_err(|error| format!("failed to restore download '{id}': {error}"))?; if inserted > 0 { restored_ids.insert(id.to_string()); + if let Some(queue_id) = queue_id { + restored_queue_ids.insert(queue_id.to_string()); + } restored_count += 1; } } @@ -496,32 +647,67 @@ fn recover_downloads_from_migration_backup( if let Ok(mut stmt) = backup_conn.prepare("SELECT id, data FROM queues") { if let Ok(rows) = stmt.query_map([], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))) { for (id, data) in rows.flatten() { - let _ = transaction.execute( - "INSERT INTO queues (id, data) VALUES (?1, ?2) ON CONFLICT(id) DO NOTHING", - params![id, data], - ); + if !orphan_recovery || restored_queue_ids.contains(&id) { + let _ = transaction.execute( + "INSERT INTO queues (id, data) VALUES (?1, ?2) ON CONFLICT(id) DO NOTHING", + params![id, data], + ); + } } } } } - transaction - .execute( - "INSERT INTO metadata (key, value) VALUES (?1, 'complete') - ON CONFLICT(key) DO UPDATE SET value = excluded.value", - params![RECOVERY_MARKER], - ) - .map_err(|error| format!("failed to record recovery completion: {error}"))?; + if !orphan_recovery { + transaction + .execute( + "INSERT INTO metadata (key, value) VALUES (?1, 'complete') + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + params![RECOVERY_MARKER], + ) + .map_err(|error| format!("failed to record recovery completion: {error}"))?; + } transaction .commit() .map_err(|error| format!("failed to commit recovery: {error}"))?; - log::info!( - "Restored {restored_count} download(s) from migration backup '{}'", - candidate.display() - ); - break; + processed_valid_candidate = true; + total_restored_count += restored_count; + if orphan_recovery { + for id in restored_ids { + orphan_ids.remove(&id); + } + if orphan_ids.is_empty() { + break; + } + } else { + log::info!( + "Restored {restored_count} download(s) from migration backup '{}'", + candidate.display() + ); + break; + } + } + + if orphan_recovery && processed_valid_candidate { + if orphan_ids.is_empty() { + connection + .execute( + "INSERT INTO metadata (key, value) VALUES (?1, 'complete') + ON CONFLICT(key) DO UPDATE SET value = excluded.value", + params![RECOVERY_MARKER], + ) + .map_err(|error| format!("failed to record recovery completion: {error}"))?; + log::info!( + "Reconciled {total_restored_count} orphaned download(s) from migration backups" + ); + } else { + log::warn!( + "Could not reconcile {} orphaned download(s) from migration backups", + orphan_ids.len() + ); + } } Ok(()) @@ -4215,12 +4401,316 @@ mod tests { ); } + #[test] + fn recover_downloads_from_migration_backup_repairs_orphaned_owned_rows() { + let root = TempDir::new().unwrap(); + let db = init_at_path(root.path()).unwrap(); + let mut connection = db.lock().unwrap(); + + let backup_path = root + .path() + .join("firelink.sqlite.backup-schema-v3-20260907T120000Z-partial"); + { + let backup_conn = Connection::open(&backup_path).unwrap(); + backup_conn + .execute_batch( + r#" + CREATE TABLE downloads (id TEXT PRIMARY KEY, status TEXT NOT NULL, queue_id TEXT, data TEXT NOT NULL); + INSERT INTO downloads VALUES ('orphaned', 'completed', 'custom-queue', '{"id":"orphaned","fileName":"orphaned.bin","status":"completed","queueId":"custom-queue"}'); + INSERT INTO downloads VALUES ('not-owned', 'completed', 'custom-queue', '{"id":"not-owned","fileName":"not-owned.bin","status":"completed","queueId":"custom-queue"}'); + CREATE TABLE download_ownership (id TEXT PRIMARY KEY, primary_path TEXT NOT NULL); + INSERT INTO download_ownership VALUES ('orphaned', '/downloads/orphaned.bin'); + INSERT INTO download_ownership VALUES ('not-owned', '/downloads/not-owned.bin'); + CREATE TABLE download_owned_paths (id TEXT PRIMARY KEY, paths TEXT NOT NULL); + INSERT INTO download_owned_paths VALUES ('orphaned', '["/downloads/orphaned.bin"]'); + CREATE TABLE queues (id TEXT PRIMARY KEY, data TEXT NOT NULL); + INSERT INTO queues VALUES ('custom-queue', '{"id":"custom-queue","name":"Custom"}'); + "#, + ) + .unwrap(); + } + + // Simulate a live row surviving a partial renderer snapshot while its + // previously persisted ownership record has no corresponding row. + connection + .execute( + "INSERT INTO downloads (id, status, queue_id, data) VALUES ('new-download', 'failed', 'main', '{\"id\":\"new-download\",\"status\":\"failed\"}')", + [], + ) + .unwrap(); + connection + .execute( + "INSERT INTO download_ownership (id, primary_path) VALUES ('orphaned', '/downloads/orphaned.bin')", + [], + ) + .unwrap(); + connection + .execute( + "INSERT INTO download_owned_paths (id, paths) VALUES ('orphaned', '[\"/downloads/orphaned.bin\"]')", + [], + ) + .unwrap(); + connection + .execute( + "INSERT INTO metadata (key, value) VALUES ('migration-backup-recovered:schema-v3', 'skipped')", + [], + ) + .unwrap(); + + recover_downloads_from_migration_backup(&mut connection, root.path(), false).unwrap(); + + let loaded = load_downloads(&connection).unwrap(); + assert_eq!(loaded.len(), 2); + assert!(loaded.iter().any(|data| data.contains("new-download"))); + assert!(loaded.iter().any(|data| data.contains("orphaned"))); + assert!(!loaded.iter().any(|data| data.contains("not-owned"))); + assert!(load_queues(&connection) + .unwrap() + .iter() + .any(|data| data.contains("custom-queue"))); + assert_eq!( + connection + .query_row::( + "SELECT value FROM metadata WHERE key = 'migration-backup-recovered:schema-v3'", + [], + |row| row.get(0), + ) + .unwrap(), + "complete" + ); + + // A later startup must not duplicate a repaired row. + recover_downloads_from_migration_backup(&mut connection, root.path(), false).unwrap(); + assert_eq!(load_downloads(&connection).unwrap().len(), 2); + + // The repair remains available even after the marker becomes + // complete, because a later stale snapshot can create the same + // orphan shape again. + connection + .execute("DELETE FROM downloads WHERE id = 'orphaned'", []) + .unwrap(); + recover_downloads_from_migration_backup(&mut connection, root.path(), false).unwrap(); + assert_eq!(load_downloads(&connection).unwrap().len(), 2); + } + + #[test] + fn recover_downloads_from_migration_backup_narrows_empty_table_to_owned_rows() { + let root = TempDir::new().unwrap(); + let db = init_at_path(root.path()).unwrap(); + let mut connection = db.lock().unwrap(); + + let backup_path = root + .path() + .join("firelink.sqlite.backup-schema-v3-20260907T120000Z-empty"); + { + let backup_conn = Connection::open(&backup_path).unwrap(); + backup_conn + .execute_batch( + r#" + CREATE TABLE downloads (id TEXT PRIMARY KEY, status TEXT NOT NULL, queue_id TEXT, data TEXT NOT NULL); + INSERT INTO downloads VALUES ('owned', 'completed', 'main', '{"id":"owned","fileName":"owned.bin","status":"completed"}'); + INSERT INTO downloads VALUES ('not-owned', 'completed', 'main', '{"id":"not-owned","fileName":"not-owned.bin","status":"completed"}'); + CREATE TABLE download_ownership (id TEXT PRIMARY KEY, primary_path TEXT NOT NULL); + INSERT INTO download_ownership VALUES ('owned', '/downloads/owned.bin'); + INSERT INTO download_ownership VALUES ('not-owned', '/downloads/not-owned.bin'); + "#, + ) + .unwrap(); + } + + // An empty downloads table is ambiguous once durable native ownership + // exists. Restore only the ownership-backed IDs; never repopulate the + // UI from unrelated backup rows. + connection + .execute( + "INSERT INTO download_ownership (id, primary_path) VALUES ('owned', '/downloads/owned.bin')", + [], + ) + .unwrap(); + + recover_downloads_from_migration_backup(&mut connection, root.path(), false).unwrap(); + + let loaded = load_downloads(&connection).unwrap(); + assert_eq!(loaded.len(), 1); + assert!(loaded[0].contains("owned")); + assert!(!loaded[0].contains("not-owned")); + assert_eq!( + connection + .query_row::( + "SELECT value FROM metadata WHERE key = 'migration-backup-recovered:schema-v3'", + [], + |row| row.get(0), + ) + .unwrap(), + "complete" + ); + } + + #[test] + fn recover_downloads_from_migration_backup_restores_nonterminal_removal_jobs() { + let root = TempDir::new().unwrap(); + let db = init_at_path(root.path()).unwrap(); + let mut connection = db.lock().unwrap(); + + let backup_path = root + .path() + .join("firelink.sqlite.backup-schema-v3-20260907T120000Z-removals"); + { + let backup_conn = Connection::open(&backup_path).unwrap(); + backup_conn + .execute_batch( + r#" + CREATE TABLE downloads (id TEXT PRIMARY KEY, status TEXT NOT NULL, queue_id TEXT, data TEXT NOT NULL); + INSERT INTO downloads VALUES ('pending-removal', 'paused', 'main', '{"id":"pending-removal","status":"paused"}'); + INSERT INTO downloads VALUES ('failed-removal', 'failed', 'main', '{"id":"failed-removal","status":"failed"}'); + INSERT INTO downloads VALUES ('completed-removal', 'completed', 'main', '{"id":"completed-removal","status":"completed"}'); + "#, + ) + .unwrap(); + } + + for (id, phase) in [ + ("pending-removal", "pending"), + ("failed-removal", "failed"), + ("completed-removal", "completed"), + ] { + connection + .execute( + "INSERT INTO download_removal_jobs (id, data) VALUES (?1, ?2)", + rusqlite::params![ + id, + format!( + "{{\"id\":\"{id}\",\"revision\":1,\"deleteAssets\":true,\"phase\":\"{phase}\",\"error\":null}}" + ) + ], + ) + .unwrap(); + } + + recover_downloads_from_migration_backup(&mut connection, root.path(), false).unwrap(); + + let loaded = load_downloads(&connection).unwrap(); + assert_eq!(loaded.len(), 2); + assert!(loaded.iter().any(|data| data.contains("pending-removal"))); + assert!(loaded.iter().any(|data| data.contains("failed-removal"))); + assert!(!loaded.iter().any(|data| data.contains("completed-removal"))); + } + + #[test] + fn recover_downloads_from_migration_backup_does_not_mark_unresolved_orphans_complete() { + let root = TempDir::new().unwrap(); + let db = init_at_path(root.path()).unwrap(); + let mut connection = db.lock().unwrap(); + + let backup_path = root + .path() + .join("firelink.sqlite.backup-schema-v3-20260907T120000Z-unrelated"); + { + let backup_conn = Connection::open(&backup_path).unwrap(); + backup_conn + .execute_batch( + r#" + CREATE TABLE downloads (id TEXT PRIMARY KEY, status TEXT NOT NULL, queue_id TEXT, data TEXT NOT NULL); + INSERT INTO downloads VALUES ('unrelated', 'completed', 'main', '{"id":"unrelated","status":"completed"}'); + "#, + ) + .unwrap(); + } + + connection + .execute( + "INSERT INTO downloads (id, status, queue_id, data) VALUES ('live', 'completed', 'main', '{\"id\":\"live\",\"status\":\"completed\"}')", + [], + ) + .unwrap(); + connection + .execute( + "INSERT INTO download_ownership (id, primary_path) VALUES ('missing-from-backup', '/downloads/missing.bin')", + [], + ) + .unwrap(); + + recover_downloads_from_migration_backup(&mut connection, root.path(), false).unwrap(); + + assert_eq!(load_downloads(&connection).unwrap().len(), 1); + let recovery_marker: Option = connection + .query_row( + "SELECT value FROM metadata WHERE key = 'migration-backup-recovered:schema-v3'", + [], + |row| row.get(0), + ) + .optional() + .unwrap(); + assert!(recovery_marker.is_none()); + } + + #[test] + fn recover_downloads_from_migration_backup_repairs_malformed_rows_by_id() { + let root = TempDir::new().unwrap(); + let db = init_at_path(root.path()).unwrap(); + let mut connection = db.lock().unwrap(); + + let backup_path = root + .path() + .join("firelink.sqlite.backup-schema-v3-20260907T120000Z-malformed-row"); + { + let backup_conn = Connection::open(&backup_path).unwrap(); + backup_conn + .execute_batch( + r#" + CREATE TABLE downloads (id TEXT PRIMARY KEY, status TEXT NOT NULL, queue_id TEXT, data TEXT NOT NULL); + INSERT INTO downloads VALUES ('corrupt', 'completed', 'main', '{"id":"corrupt","status":"completed"}'); + INSERT INTO downloads VALUES ('unrelated', 'completed', 'main', '{"id":"unrelated","status":"completed"}'); + "#, + ) + .unwrap(); + } + + connection + .execute( + "INSERT INTO downloads (id, status, queue_id, data) VALUES ('corrupt', 'completed', 'main', 'not-json')", + [], + ) + .unwrap(); + + recover_downloads_from_migration_backup(&mut connection, root.path(), false).unwrap(); + + let loaded = load_downloads(&connection).unwrap(); + assert_eq!(loaded.len(), 1); + assert!(loaded[0].contains("corrupt")); + assert!(!loaded[0].contains("unrelated")); + } + #[test] fn recover_downloads_from_migration_backup_skips_corrupt_candidate_and_continues_to_valid() { let root = TempDir::new().unwrap(); let corrupt_path = root.path().join("firelink.sqlite.backup-schema-v3-20260907T130000Z-corrupt"); fs::write(&corrupt_path, b"not a valid sqlite file").unwrap(); + let unrelated_path = root + .path() + .join("firelink.sqlite.backup-schema-v30-20260907T140000Z-unrelated"); + { + let unrelated_conn = Connection::open(&unrelated_path).unwrap(); + unrelated_conn + .execute_batch( + "CREATE TABLE downloads (id TEXT PRIMARY KEY, status TEXT NOT NULL, queue_id TEXT, data TEXT NOT NULL);\n INSERT INTO downloads VALUES ('unrelated', 'completed', 'main', '{\"id\":\"unrelated\"}');", + ) + .unwrap(); + } + + let malformed_path = root + .path() + .join("firelink.sqlite.backup-schema-v3-20260907T120000Z-malformed"); + { + let malformed_conn = Connection::open(&malformed_path).unwrap(); + malformed_conn + .execute_batch( + "CREATE TABLE downloads (id TEXT PRIMARY KEY, status TEXT NOT NULL, queue_id TEXT, data TEXT NOT NULL);\n INSERT INTO downloads VALUES ('malformed', 'completed', 'main', 'not-json');", + ) + .unwrap(); + } + let valid_path = root.path().join("firelink.sqlite.backup-schema-v3-20260907T110000Z-valid"); { let valid_conn = Connection::open(&valid_path).unwrap(); diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index ed5af23..88013d3 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -3516,6 +3516,13 @@ impl Aria2DaemonGuard { .store(available, Ordering::Release); } + fn startup_error_message(&self) -> Option { + self.startup_error + .lock() + .unwrap_or_else(|error| error.into_inner()) + .clone() + } + fn exit_allowed(&self) -> bool { self.shutdown_state.load(Ordering::SeqCst) == 2 } @@ -12494,7 +12501,22 @@ fn apply_aria2_torrent_peer_identity_options( } fn aria2_rpc_port_is_occupied(port: u16) -> bool { - std::net::TcpListener::bind(("127.0.0.1", port)).is_err() + let addresses = [ + std::net::SocketAddr::from(([127, 0, 0, 1], port)), + std::net::SocketAddr::from(([0, 0, 0, 0, 0, 0, 0, 1], port)), + ]; + addresses.into_iter().any(|address| { + std::net::TcpListener::bind(address) + .map(|_| false) + .unwrap_or_else(|error| error.kind() == std::io::ErrorKind::AddrInUse) + }) +} + +fn aria2_rpc_startup_reports_port_conflict(stderr: &str, port: u16) -> bool { + let lower = stderr.to_ascii_lowercase(); + lower.contains("rpc") + && lower.contains("failed to bind") + && lower.contains(&format!("port {port}")) } fn apply_aria2_torrent_global_options( @@ -14280,6 +14302,7 @@ mod tests { apply_aria2_torrent_dht_options, apply_aria2_server_stat_options, aria2_rpc_port_is_occupied, + aria2_rpc_startup_reports_port_conflict, parse_firelink_deep_link, parse_ffmpeg_version, parse_media_progress_line, collect_opened_torrent_paths, normalize_opened_torrent_argument, @@ -15190,6 +15213,36 @@ mod tests { assert!(aria2_rpc_port_is_occupied(port)); drop(listener); assert!(!aria2_rpc_port_is_occupied(port)); + + // Aria2's loopback RPC listener can be claimed through either address + // family. If IPv6 loopback is available, the launcher must not reuse + // a port that is already occupied there either. + if let Ok(ipv6_listener) = std::net::TcpListener::bind(("::1", 0)) { + let ipv6_port = ipv6_listener.local_addr().unwrap().port(); + assert!(aria2_rpc_port_is_occupied(ipv6_port)); + drop(ipv6_listener); + assert!(!aria2_rpc_port_is_occupied(ipv6_port)); + } + } + + #[test] + fn aria2_rpc_startup_port_conflicts_are_distinguished_from_other_failures() { + assert!(aria2_rpc_startup_reports_port_conflict( + "IPv4 RPC: failed to bind TCP port 6800", + 6800 + )); + assert!(aria2_rpc_startup_reports_port_conflict( + "IPv6 RPC: failed to bind TCP port 6800: Address already in use", + 6800 + )); + assert!(!aria2_rpc_startup_reports_port_conflict( + "Exception caught: unrecognized option --bad-option", + 6800 + )); + assert!(!aria2_rpc_startup_reports_port_conflict( + "IPv4 RPC: failed to bind TCP port 6800", + 6801 + )); } #[test] @@ -19296,7 +19349,12 @@ pub fn run() { let _ = pipe.take(4096).read_to_string(&mut stderr); } let stderr = stderr.trim().to_string(); - if aria2_rpc_port_is_occupied(attempt_port) { + if aria2_rpc_port_is_occupied(attempt_port) + || aria2_rpc_startup_reports_port_conflict( + &stderr, + attempt_port, + ) + { // The RPC port is occupied by another process; retry // with the next candidate. A configured Torrent port // cannot cause this branch because overlapping RPC diff --git a/src-tauri/src/queue.rs b/src-tauri/src/queue.rs index 6735a6d..1d6a322 100644 --- a/src-tauri/src/queue.rs +++ b/src-tauri/src/queue.rs @@ -6650,6 +6650,18 @@ fn is_aria2_rpc_unavailable(error: &str) -> bool { || lower.contains("connection reset") } +fn aria2_startup_failure_for_port(port: u16, startup_error: Option<&str>) -> Option { + if port != 0 { + return None; + } + startup_error.map(|error| { + format!( + "aria2 daemon unavailable: {}", + crate::redact_sensitive_text(error) + ) + }) +} + fn is_aria2_range_mode_error(error: &str) -> bool { let lower = error.to_ascii_lowercase(); lower.contains("invalid range header") @@ -8340,14 +8352,27 @@ impl ProductionSpawner { ) -> Result { let deadline = std::time::Instant::now() + std::time::Duration::from_secs(15); loop { - match crate::rpc_call( - state.aria2_port.load(std::sync::atomic::Ordering::Relaxed), - &state.aria2_secret, - method, - params.clone(), - ) - .await - { + let port = state + .aria2_port + .load(std::sync::atomic::Ordering::Acquire); + if port == 0 { + let startup_error = self + .app_handle + .state::() + .startup_error_message(); + if let Some(error) = + aria2_startup_failure_for_port(port, startup_error.as_deref()) + { + return Err(error); + } + if std::time::Instant::now() >= deadline { + return Err("aria2 daemon did not become ready within 15 seconds".to_string()); + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + continue; + } + + match crate::rpc_call(port, &state.aria2_secret, method, params.clone()).await { Ok(result) => return Ok(result), Err(error) => { if !is_aria2_rpc_unavailable(&error) || std::time::Instant::now() >= deadline { @@ -8406,6 +8431,18 @@ impl ProductionSpawner { impl SidecarSpawner for ProductionSpawner { async fn add_uri(&self, id: &str, payload: &SpawnPayload) -> Result { let state = self.app_handle.state::(); + let startup_error = self + .app_handle + .state::() + .startup_error_message(); + if let Some(error) = aria2_startup_failure_for_port( + state + .aria2_port + .load(std::sync::atomic::Ordering::Acquire), + startup_error.as_deref(), + ) { + return Err(error); + } let attempt_epoch = state.queue_manager.current_aria2_control_epoch(id).await; let admission_started = Instant::now(); let mut options = serde_json::Map::new(); @@ -12308,6 +12345,22 @@ mod tests { )); } + #[test] + fn known_aria2_startup_failure_is_returned_without_rpc_retry() { + assert_eq!( + aria2_startup_failure_for_port( + 0, + Some("aria2 exited before startup with token=secret") + ), + Some( + "aria2 daemon unavailable: aria2 exited before startup with token=[redacted]" + .to_string() + ) + ); + assert!(aria2_startup_failure_for_port(6800, Some("daemon unavailable")).is_none()); + assert!(aria2_startup_failure_for_port(0, None).is_none()); + } + #[test] fn aria2_internal_attempts_do_not_multiply_firelink_retry_budget() { assert_eq!(aria2_attempt_limit(Some(0)), 1);