diff --git a/src/orphans/orphan_checkup.rs b/src/orphans/orphan_checkup.rs index 0cb8eb3..d6becda 100644 --- a/src/orphans/orphan_checkup.rs +++ b/src/orphans/orphan_checkup.rs @@ -129,6 +129,7 @@ async fn candidate_attaches_before_rollback( in_memory_pieces: Arc::new(Mutex::new(crate::HashMap::new())), use_in_memory_pieces: false, torrent: torrent.clone(), + torrent_bytes: None, staged_path: String::new(), block_number: height, allow_during_reorg: true, diff --git a/src/rpc/client/syncing.rs b/src/rpc/client/syncing.rs index 0333db4..7667470 100644 --- a/src/rpc/client/syncing.rs +++ b/src/rpc/client/syncing.rs @@ -12,7 +12,6 @@ use crate::sled::Db; use crate::timeout; use crate::torrent::structs::Torrent; use crate::torrent::torrenting_system::save_block::verify_and_save_downloaded_block; -use crate::torrent::torrenting_system::save_torrent::save_staged_torrent; use crate::torrent::torrenting_system::setup_block_download::download_sync_candidate; use crate::torrent::torrenting_system::torrent_requests::send_request_torrent_message; use crate::verifications::verification_service::global_verification_service; @@ -24,6 +23,18 @@ use crate::TcpStream; use std::collections::BTreeMap; const SYNC_PREFETCH_WINDOW: usize = 100; +const SYNC_TRANSIENT_RETRY_LIMIT: u8 = 3; + +fn is_transient_sync_error(err: &str) -> bool { + err.contains("No available peer could provide remaining pieces") + || err.contains("Timed out waiting for piece reply") + || err.contains("Piece reply channel closed") + || err.contains("Peer could not provide piece") + || err.contains("Requested candidate not found") + || err.contains("piece not found") + || err.contains("Candidate download already active") + || err.contains("Timed out waiting for active candidate download") +} fn abort_pending_downloads( pending: &mut BTreeMap< @@ -76,9 +87,6 @@ fn spawn_sync_download( let torrent = Torrent::from_bytes(&torrent_bytes) .await .map_err(|err| err.to_string())?; - let staged_path = save_staged_torrent(height, &torrent_bytes) - .await - .map_err(|err| format!("Failed to save staged torrent: {err}"))?; set_torrent_status(height, &torrent.info.info_hash, TorrentStatus::Pending).await; let verification_service = global_verification_service() @@ -86,7 +94,8 @@ fn spawn_sync_download( download_sync_candidate( height, torrent, - staged_path, + String::new(), + torrent_bytes, allow_startup_peers, db, Arc::new(verification_service), @@ -160,6 +169,7 @@ pub async fn node_syncing( let mut next_to_schedule = local_height; let mut next_to_save = local_height; let mut pending = BTreeMap::new(); + let mut retry_counts = BTreeMap::::new(); // Download a small window of candidates ahead, but validate/save only // the exact next height so parent, difficulty, and balance rules still @@ -198,6 +208,18 @@ pub async fn node_syncing( } break; } + if is_transient_sync_error(&err) { + let retries = retry_counts.entry(next_to_save).or_insert(0); + if *retries < SYNC_TRANSIENT_RETRY_LIMIT { + *retries += 1; + warn!( + "[sync] transient download error at height={next_to_save}; retry {}/{} before orphan recovery: {err}", + *retries, SYNC_TRANSIENT_RETRY_LIMIT + ); + next_to_schedule = next_to_save; + continue; + } + } handle_sync_error( &err, next_to_save, @@ -228,6 +250,18 @@ pub async fn node_syncing( if let Err(err) = verify_and_save_downloaded_block(download, wallet.clone()).await { abort_pending_downloads(&mut pending); + if is_transient_sync_error(&err) { + let retries = retry_counts.entry(next_to_save).or_insert(0); + if *retries < SYNC_TRANSIENT_RETRY_LIMIT { + *retries += 1; + warn!( + "[sync] transient save/download error at height={next_to_save}; retry {}/{} before orphan recovery: {err}", + *retries, SYNC_TRANSIENT_RETRY_LIMIT + ); + next_to_schedule = next_to_save; + continue; + } + } handle_sync_error( &err, next_to_save, @@ -249,6 +283,7 @@ pub async fn node_syncing( } next_to_save += 1; + retry_counts.remove(&(next_to_save - 1)); } abort_pending_downloads(&mut pending); info!("[sync] node syncing complete, awaiting post-sync checks before mining resumes"); diff --git a/src/rpc/commands/token_lookup.rs b/src/rpc/commands/token_lookup.rs index a96f37b..bb37201 100644 --- a/src/rpc/commands/token_lookup.rs +++ b/src/rpc/commands/token_lookup.rs @@ -309,10 +309,13 @@ pub async fn token_catalog(db: &Db) -> RpcResponse { const BURN_COUNT_BYTES: usize = 4; let mut rows = Vec::new(); - let RpcResponse::Binary(token_list_bytes) = crate::rpc::commands::token_list::get_tokens(db).await; + let RpcResponse::Binary(token_list_bytes) = + crate::rpc::commands::token_list::get_tokens(db).await; if token_list_bytes.len() % TOKEN_LIST_ROW_BYTES != 0 { - return RpcResponse::Binary(b"error: Token list response had an unexpected byte length".to_vec()); + return RpcResponse::Binary( + b"error: Token list response had an unexpected byte length".to_vec(), + ); } for token_row in token_list_bytes.chunks_exact(TOKEN_LIST_ROW_BYTES) { diff --git a/src/torrent/structs.rs b/src/torrent/structs.rs index afae2de..886dfb4 100644 --- a/src/torrent/structs.rs +++ b/src/torrent/structs.rs @@ -52,6 +52,7 @@ pub struct DownloadSave { pub in_memory_pieces: Arc>>>, pub use_in_memory_pieces: bool, pub torrent: Torrent, + pub torrent_bytes: Option>, pub staged_path: String, pub block_number: u32, pub allow_during_reorg: bool, diff --git a/src/torrent/torrenting_system/save_block.rs b/src/torrent/torrenting_system/save_block.rs index 1e2f89a..532050b 100644 --- a/src/torrent/torrenting_system/save_block.rs +++ b/src/torrent/torrenting_system/save_block.rs @@ -8,7 +8,9 @@ use crate::records::record_chain::structs::{SaveBlockParams, SaveType}; use crate::records::unpack_block::load_by_binary_data::load_block_from_binary; use crate::torrent::structs::DownloadSave; use crate::torrent::torrenting_system::create_file::combine_pieces; -use crate::torrent::torrenting_system::save_torrent::promote_staged_torrent; +use crate::torrent::torrenting_system::save_torrent::{ + promote_staged_torrent, save_canonical_torrent_bytes, +}; use crate::torrent::torrenting_system::temp_database_storage::remove_block_pieces_from_db; use crate::wallets::structures::Wallet; use crate::Arc; @@ -134,8 +136,15 @@ async fn verify_and_save_block_payload( return Err(err); } - // Once the block is saved successfully, move the staged torrent into its canonical path. - match promote_staged_torrent(Path::new(¶ms.staged_path), params.block_number) { + // Once the block is saved successfully, publish its torrent metadata. + // Startup sync keeps torrent bytes in memory to avoid exposing partial staged files. + let promote_result = if let Some(torrent_bytes) = params.torrent_bytes.as_deref() { + save_canonical_torrent_bytes(params.block_number, torrent_bytes).await + } else { + promote_staged_torrent(Path::new(¶ms.staged_path), params.block_number) + }; + + match promote_result { Ok(_) => { // The saved block and canonical torrent can now serve future // piece requests, so this node no longer needs its temporary diff --git a/src/torrent/torrenting_system/save_torrent.rs b/src/torrent/torrenting_system/save_torrent.rs index c6fa492..b7c34ad 100644 --- a/src/torrent/torrenting_system/save_torrent.rs +++ b/src/torrent/torrenting_system/save_torrent.rs @@ -2,70 +2,71 @@ use crate::common::network_paths_and_settings::block_extension_and_paths; use crate::torrent::structs::Torrent; use crate::{create_dir_all, fs, read, read_dir, remove_file, AsyncWriteExt, Path}; use std::io::ErrorKind; - -fn staged_torrent_dir() -> Result { - // Keep staged torrents under a dedicated subdirectory so incoming - // network data is separated from canonical saved torrent files. - let ( - _network_name, - _padded_base_coin, - _block_ext, - out_path, - _wallet_path, - _block_path, - _db_path, - _balance_path, - _log_path, - ) = block_extension_and_paths(); - let staging_dir = Path::new(&out_path).join("staging"); - Ok(staging_dir) -} - +use std::time::{SystemTime, UNIX_EPOCH}; + +fn staged_torrent_dir() -> Result { + // Keep staged torrents under a dedicated subdirectory so incoming + // network data is separated from canonical saved torrent files. + let ( + _network_name, + _padded_base_coin, + _block_ext, + out_path, + _wallet_path, + _block_path, + _db_path, + _balance_path, + _log_path, + ) = block_extension_and_paths(); + let staging_dir = Path::new(&out_path).join("staging"); + Ok(staging_dir) +} + fn staged_torrent_path(height: u32, info_hash: &str) -> std::path::PathBuf { // Each candidate gets its own deterministic staging path, so competing // torrents at the same height cannot race over a shared suffix. let ( _network_name, _padded_base_coin, - _block_ext, - out_path, - _wallet_path, - _block_path, - _db_path, - _balance_path, - _log_path, + _block_ext, + out_path, + _wallet_path, + _block_path, + _db_path, + _balance_path, + _log_path, ) = block_extension_and_paths(); Path::new(&out_path) .join("staging") .join(format!("{height}.{info_hash}.torrent")) } - -fn canonical_torrent_path(height: u32) -> std::path::PathBuf { - let ( - _network_name, - _padded_base_coin, - _block_ext, - out_path, - _wallet_path, - _block_path, - _db_path, - _balance_path, - _log_path, - ) = block_extension_and_paths(); - Path::new(&out_path).join(format!("{height}.torrent")) -} - -async fn torrent_bytes_match(path: &Path, torrent_bytes: &[u8]) -> Result { - if !path.exists() { - return Ok(false); - } - - let existing = read(path) - .await - .map_err(|e| format!("Failed to read torrent {}: {}", path.display(), e))?; - Ok(existing == torrent_bytes) -} - + +fn canonical_torrent_path(height: u32) -> std::path::PathBuf { + let ( + _network_name, + _padded_base_coin, + _block_ext, + out_path, + _wallet_path, + _block_path, + _db_path, + _balance_path, + _log_path, + ) = block_extension_and_paths(); + Path::new(&out_path).join(format!("{height}.torrent")) +} + +async fn torrent_bytes_match(path: &Path, torrent_bytes: &[u8]) -> Result { + if !path.exists() { + return Ok(false); + } + + let existing = read(path) + .await + .map_err(|e| format!("Failed to read torrent {}: {}", path.display(), e))?; + Ok(existing == torrent_bytes) +} + fn parse_staged_torrent_file_name(file_name: &str) -> Option<(u32, String)> { // New staged names encode both height and candidate hash: // `..torrent`. @@ -84,58 +85,58 @@ fn parse_staged_torrent_file_name(file_name: &str) -> Option<(u32, String)> { let suffix = suffix_str.parse::().ok()?; Some((height, format!("legacy-{suffix:010}"))) } - -pub async fn list_staged_torrents() -> Result, String> { + +pub async fn list_staged_torrents() -> Result, String> { // Enumerate staged torrents in height/candidate order so replay and // cleanup logic handle them in a stable sequence. - let staging_dir = staged_torrent_dir()?; - - if !staging_dir.exists() { - return Ok(Vec::new()); - } - - let mut entries = read_dir(&staging_dir) - .await - .map_err(|e| format!("Failed to read staging directory: {e}"))?; - let mut staged = Vec::new(); - - while let Some(entry) = entries - .next_entry() - .await - .map_err(|e| format!("Failed to iterate staging directory: {e}"))? - { - let path = entry.path(); - let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else { - continue; - }; + let staging_dir = staged_torrent_dir()?; + + if !staging_dir.exists() { + return Ok(Vec::new()); + } + + let mut entries = read_dir(&staging_dir) + .await + .map_err(|e| format!("Failed to read staging directory: {e}"))?; + let mut staged = Vec::new(); + + while let Some(entry) = entries + .next_entry() + .await + .map_err(|e| format!("Failed to iterate staging directory: {e}"))? + { + let path = entry.path(); + let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else { + continue; + }; if let Some((height, candidate_key)) = parse_staged_torrent_file_name(file_name) { staged.push((height, candidate_key, path)); } } - - staged.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1))); - Ok(staged - .into_iter() - .map(|(height, _suffix, path)| (height, path)) - .collect()) -} - -pub async fn list_staged_torrents_for_height( - height: u32, -) -> Result, String> { - let staged = list_staged_torrents().await?; - Ok(staged - .into_iter() - .filter_map(|(staged_height, path)| { - if staged_height == height { - Some(path) - } else { - None - } - }) - .collect()) -} - + + staged.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1))); + Ok(staged + .into_iter() + .map(|(height, _suffix, path)| (height, path)) + .collect()) +} + +pub async fn list_staged_torrents_for_height( + height: u32, +) -> Result, String> { + let staged = list_staged_torrents().await?; + Ok(staged + .into_iter() + .filter_map(|(staged_height, path)| { + if staged_height == height { + Some(path) + } else { + None + } + }) + .collect()) +} + pub async fn save_staged_torrent(height: u32, torrent_bytes: &[u8]) -> Result { // Store this candidate by its advertised block info hash. That makes // staging idempotent per candidate and prevents same-height races from @@ -183,9 +184,9 @@ pub async fn save_staged_torrent(height: u32, torrent_bytes: &[u8]) -> Result return Err(format!("Failed to create staged torrent file: {err}")), }; - - // Keep torrent staging separate from the canonical torrent file so - // validation and orphan logic can decide when promotion is safe. + + // Keep torrent staging separate from the canonical torrent file so + // validation and orphan logic can decide when promotion is safe. torrent_file .write_all(torrent_bytes) .await @@ -194,55 +195,55 @@ pub async fn save_staged_torrent(height: u32, torrent_bytes: &[u8]) -> Result Result, String> { - read(path) - .await - .map_err(|e| format!("Failed to read staged torrent {}: {}", path.display(), e)) -} - -pub async fn remove_staged_torrent(path: &Path) -> Result<(), String> { - if path.exists() { - remove_file(path) - .await - .map_err(|e| format!("Failed to remove staged torrent {}: {}", path.display(), e))?; - } - Ok(()) -} - -pub async fn remove_staged_torrents_for_height(height: u32) -> Result<(), String> { - // Remove every staged copy for the requested height so stale - // torrents do not interfere with later replay or promotion. - let staged = list_staged_torrents().await?; - for (staged_height, path) in staged { - if staged_height == height { - remove_staged_torrent(&path).await?; - } - } - Ok(()) -} - -pub async fn prune_staged_torrents(current_height: u32) -> Result<(), String> { - // Snapshot-based pruning keeps recent orphan evidence available - // until the trusted rollback floor advances past it. - let cutoff = current_height; - let staged = list_staged_torrents().await?; - for (staged_height, path) in staged { - if staged_height <= cutoff { - remove_staged_torrent(&path).await?; - } - } - Ok(()) -} - -pub fn promote_staged_torrent(staged_path: &Path, height: u32) -> Result { - // Promotion moves the selected staged torrent into the canonical - // filename once the corresponding block has been saved. - let canonical_path = canonical_torrent_path(height); - + + Ok(torrent_file_path.to_string_lossy().to_string()) +} + +pub async fn read_staged_torrent(path: &Path) -> Result, String> { + read(path) + .await + .map_err(|e| format!("Failed to read staged torrent {}: {}", path.display(), e)) +} + +pub async fn remove_staged_torrent(path: &Path) -> Result<(), String> { + if path.exists() { + remove_file(path) + .await + .map_err(|e| format!("Failed to remove staged torrent {}: {}", path.display(), e))?; + } + Ok(()) +} + +pub async fn remove_staged_torrents_for_height(height: u32) -> Result<(), String> { + // Remove every staged copy for the requested height so stale + // torrents do not interfere with later replay or promotion. + let staged = list_staged_torrents().await?; + for (staged_height, path) in staged { + if staged_height == height { + remove_staged_torrent(&path).await?; + } + } + Ok(()) +} + +pub async fn prune_staged_torrents(current_height: u32) -> Result<(), String> { + // Snapshot-based pruning keeps recent orphan evidence available + // until the trusted rollback floor advances past it. + let cutoff = current_height; + let staged = list_staged_torrents().await?; + for (staged_height, path) in staged { + if staged_height <= cutoff { + remove_staged_torrent(&path).await?; + } + } + Ok(()) +} + +pub fn promote_staged_torrent(staged_path: &Path, height: u32) -> Result { + // Promotion moves the selected staged torrent into the canonical + // filename once the corresponding block has been saved. + let canonical_path = canonical_torrent_path(height); + if !staged_path.exists() { return Err(format!( "Staged torrent does not exist for block {} at {}", @@ -257,6 +258,64 @@ pub fn promote_staged_torrent(staged_path: &Path, height: u32) -> Result Result { + let canonical_path = canonical_torrent_path(height); + + if torrent_bytes_match(&canonical_path, torrent_bytes).await? { + return Ok(canonical_path.to_string_lossy().to_string()); + } + if canonical_path.exists() { + return Err(format!( + "Canonical torrent collision for height {height} at {}", + canonical_path.display() + )); + } + + let parent = canonical_path.parent().ok_or_else(|| { + format!( + "Canonical torrent path has no parent: {}", + canonical_path.display() + ) + })?; + create_dir_all(parent) + .await + .map_err(|e| format!("Failed to create canonical torrent directory: {e}"))?; + + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|e| format!("System clock error while saving torrent: {e}"))? + .as_nanos(); + let temp_path = parent.join(format!("{height}.torrent.sync-{unique}.tmp")); + + let mut temp_file = crate::tokio::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&temp_path) + .await + .map_err(|e| format!("Failed to create temporary canonical torrent file: {e}"))?; + temp_file + .write_all(torrent_bytes) + .await + .map_err(|e| format!("Failed to write temporary canonical torrent file: {e}"))?; + temp_file + .flush() + .await + .map_err(|e| format!("Failed to flush temporary canonical torrent file: {e}"))?; + drop(temp_file); + + if let Err(err) = crate::tokio::fs::rename(&temp_path, &canonical_path).await { + let _ = remove_file(&temp_path).await; + return Err(format!( + "Failed to promote temporary canonical torrent: {err}" + )); + } + + Ok(canonical_path.to_string_lossy().to_string()) +} diff --git a/src/torrent/torrenting_system/setup_block_download.rs b/src/torrent/torrenting_system/setup_block_download.rs index aa821e0..59773e1 100644 --- a/src/torrent/torrenting_system/setup_block_download.rs +++ b/src/torrent/torrenting_system/setup_block_download.rs @@ -38,6 +38,7 @@ pub async fn setup_download( in_memory_pieces: Arc::new(Mutex::new(HashMap::new())), use_in_memory_pieces, torrent, + torrent_bytes: None, staged_path, block_number, allow_during_reorg, @@ -57,6 +58,7 @@ pub async fn download_sync_candidate( block_number: u32, torrent: Torrent, staged_path: String, + torrent_bytes: Vec, allow_startup_peers: bool, db: Db, verification_service: Arc, @@ -71,6 +73,7 @@ pub async fn download_sync_candidate( in_memory_pieces: Arc::new(Mutex::new(HashMap::new())), use_in_memory_pieces: true, torrent, + torrent_bytes: Some(torrent_bytes), staged_path, block_number, allow_during_reorg: false,