syncing fixes

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,

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::<u32, u8>::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");

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) {

src/torrent/structs.rs

@@ -52,6 +52,7 @@ pub struct DownloadSave {
     pub in_memory_pieces: Arc<Mutex<HashMap<u8, Vec<u8>>>>,
     pub use_in_memory_pieces: bool,
     pub torrent: Torrent,
+    pub torrent_bytes: Option<Vec<u8>>,
     pub staged_path: String,
     pub block_number: u32,
     pub allow_during_reorg: bool,

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.
+    // Once the block is saved successfully, publish its torrent metadata.
-    match promote_staged_torrent(Path::new(&params.staged_path), params.block_number) {
+    // 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(&params.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

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;
-
+use std::time::{SystemTime, UNIX_EPOCH};
-fn staged_torrent_dir() -> Result<std::path::PathBuf, String> {
+
-    // Keep staged torrents under a dedicated subdirectory so incoming
+fn staged_torrent_dir() -> Result<std::path::PathBuf, String> {
-    // network data is separated from canonical saved torrent files.
+    // Keep staged torrents under a dedicated subdirectory so incoming
-    let (
+    // network data is separated from canonical saved torrent files.
-        _network_name,
+    let (
-        _padded_base_coin,
+        _network_name,
-        _block_ext,
+        _padded_base_coin,
-        out_path,
+        _block_ext,
-        _wallet_path,
+        out_path,
-        _block_path,
+        _wallet_path,
-        _db_path,
+        _block_path,
-        _balance_path,
+        _db_path,
-        _log_path,
+        _balance_path,
-    ) = block_extension_and_paths();
+        _log_path,
-    let staging_dir = Path::new(&out_path).join("staging");
+    ) = block_extension_and_paths();
-    Ok(staging_dir)
+    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,
+        _block_ext,
-        out_path,
+        out_path,
-        _wallet_path,
+        _wallet_path,
-        _block_path,
+        _block_path,
-        _db_path,
+        _db_path,
-        _balance_path,
+        _balance_path,
-        _log_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 {
+fn canonical_torrent_path(height: u32) -> std::path::PathBuf {
-    let (
+    let (
-        _network_name,
+        _network_name,
-        _padded_base_coin,
+        _padded_base_coin,
-        _block_ext,
+        _block_ext,
-        out_path,
+        out_path,
-        _wallet_path,
+        _wallet_path,
-        _block_path,
+        _block_path,
-        _db_path,
+        _db_path,
-        _balance_path,
+        _balance_path,
-        _log_path,
+        _log_path,
-    ) = block_extension_and_paths();
+    ) = block_extension_and_paths();
-    Path::new(&out_path).join(format!("{height}.torrent"))
+    Path::new(&out_path).join(format!("{height}.torrent"))
-}
+}
-
+
-async fn torrent_bytes_match(path: &Path, torrent_bytes: &[u8]) -> Result<bool, String> {
+async fn torrent_bytes_match(path: &Path, torrent_bytes: &[u8]) -> Result<bool, String> {
-    if !path.exists() {
+    if !path.exists() {
-        return Ok(false);
+        return Ok(false);
-    }
+    }
-
+
-    let existing = read(path)
+    let existing = read(path)
-        .await
+        .await
-        .map_err(|e| format!("Failed to read torrent {}: {}", path.display(), e))?;
+        .map_err(|e| format!("Failed to read torrent {}: {}", path.display(), e))?;
-    Ok(existing == torrent_bytes)
+    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:
     // `<height>.<info_hash>.torrent`.
@@ -84,58 +85,58 @@ fn parse_staged_torrent_file_name(file_name: &str) -> Option<(u32, String)> {
     let suffix = suffix_str.parse::<u32>().ok()?;
     Some((height, format!("legacy-{suffix:010}")))
 }
-
+
-pub async fn list_staged_torrents() -> Result<Vec<(u32, std::path::PathBuf)>, String> {
+pub async fn list_staged_torrents() -> Result<Vec<(u32, std::path::PathBuf)>, 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()?;
+    let staging_dir = staged_torrent_dir()?;
-
+
-    if !staging_dir.exists() {
+    if !staging_dir.exists() {
-        return Ok(Vec::new());
+        return Ok(Vec::new());
-    }
+    }
-
+
-    let mut entries = read_dir(&staging_dir)
+    let mut entries = read_dir(&staging_dir)
-        .await
+        .await
-        .map_err(|e| format!("Failed to read staging directory: {e}"))?;
+        .map_err(|e| format!("Failed to read staging directory: {e}"))?;
-    let mut staged = Vec::new();
+    let mut staged = Vec::new();
-
+
-    while let Some(entry) = entries
+    while let Some(entry) = entries
-        .next_entry()
+        .next_entry()
-        .await
+        .await
-        .map_err(|e| format!("Failed to iterate staging directory: {e}"))?
+        .map_err(|e| format!("Failed to iterate staging directory: {e}"))?
-    {
+    {
-        let path = entry.path();
+        let path = entry.path();
-        let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
+        let Some(file_name) = path.file_name().and_then(|name| name.to_str()) else {
-            continue;
+            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)));
+    staged.sort_by(|a, b| a.0.cmp(&b.0).then(a.1.cmp(&b.1)));
-    Ok(staged
+    Ok(staged
-        .into_iter()
+        .into_iter()
-        .map(|(height, _suffix, path)| (height, path))
+        .map(|(height, _suffix, path)| (height, path))
-        .collect())
+        .collect())
-}
+}
-
+
-pub async fn list_staged_torrents_for_height(
+pub async fn list_staged_torrents_for_height(
-    height: u32,
+    height: u32,
-) -> Result<Vec<std::path::PathBuf>, String> {
+) -> Result<Vec<std::path::PathBuf>, String> {
-    let staged = list_staged_torrents().await?;
+    let staged = list_staged_torrents().await?;
-    Ok(staged
+    Ok(staged
-        .into_iter()
+        .into_iter()
-        .filter_map(|(staged_height, path)| {
+        .filter_map(|(staged_height, path)| {
-            if staged_height == height {
+            if staged_height == height {
-                Some(path)
+                Some(path)
-            } else {
+            } else {
-                None
+                None
-            }
+            }
-        })
+        })
-        .collect())
+        .collect())
-}
+}
-
+
 pub async fn save_staged_torrent(height: u32, torrent_bytes: &[u8]) -> Result<String, String> {
     // 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<St
         }
         Err(err) => return Err(format!("Failed to create staged torrent file: {err}")),
     };
-
+
-    // Keep torrent staging separate from the canonical torrent file so
+    // Keep torrent staging separate from the canonical torrent file so
-    // validation and orphan logic can decide when promotion is safe.
+    // 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<St
         .flush()
         .await
         .map_err(|e| format!("Failed to flush staged torrent file: {e}"))?;
-
+
-    Ok(torrent_file_path.to_string_lossy().to_string())
+    Ok(torrent_file_path.to_string_lossy().to_string())
-}
+}
-
+
-pub async fn read_staged_torrent(path: &Path) -> Result<Vec<u8>, String> {
+pub async fn read_staged_torrent(path: &Path) -> Result<Vec<u8>, String> {
-    read(path)
+    read(path)
-        .await
+        .await
-        .map_err(|e| format!("Failed to read staged torrent {}: {}", path.display(), e))
+        .map_err(|e| format!("Failed to read staged torrent {}: {}", path.display(), e))
-}
+}
-
+
-pub async fn remove_staged_torrent(path: &Path) -> Result<(), String> {
+pub async fn remove_staged_torrent(path: &Path) -> Result<(), String> {
-    if path.exists() {
+    if path.exists() {
-        remove_file(path)
+        remove_file(path)
-            .await
+            .await
-            .map_err(|e| format!("Failed to remove staged torrent {}: {}", path.display(), e))?;
+            .map_err(|e| format!("Failed to remove staged torrent {}: {}", path.display(), e))?;
-    }
+    }
-    Ok(())
+    Ok(())
-}
+}
-
+
-pub async fn remove_staged_torrents_for_height(height: u32) -> Result<(), String> {
+pub async fn remove_staged_torrents_for_height(height: u32) -> Result<(), String> {
-    // Remove every staged copy for the requested height so stale
+    // Remove every staged copy for the requested height so stale
-    // torrents do not interfere with later replay or promotion.
+    // torrents do not interfere with later replay or promotion.
-    let staged = list_staged_torrents().await?;
+    let staged = list_staged_torrents().await?;
-    for (staged_height, path) in staged {
+    for (staged_height, path) in staged {
-        if staged_height == height {
+        if staged_height == height {
-            remove_staged_torrent(&path).await?;
+            remove_staged_torrent(&path).await?;
-        }
+        }
-    }
+    }
-    Ok(())
+    Ok(())
-}
+}
-
+
-pub async fn prune_staged_torrents(current_height: u32) -> Result<(), String> {
+pub async fn prune_staged_torrents(current_height: u32) -> Result<(), String> {
-    // Snapshot-based pruning keeps recent orphan evidence available
+    // Snapshot-based pruning keeps recent orphan evidence available
-    // until the trusted rollback floor advances past it.
+    // until the trusted rollback floor advances past it.
-    let cutoff = current_height;
+    let cutoff = current_height;
-    let staged = list_staged_torrents().await?;
+    let staged = list_staged_torrents().await?;
-    for (staged_height, path) in staged {
+    for (staged_height, path) in staged {
-        if staged_height <= cutoff {
+        if staged_height <= cutoff {
-            remove_staged_torrent(&path).await?;
+            remove_staged_torrent(&path).await?;
-        }
+        }
-    }
+    }
-    Ok(())
+    Ok(())
-}
+}
-
+
-pub fn promote_staged_torrent(staged_path: &Path, height: u32) -> Result<String, String> {
+pub fn promote_staged_torrent(staged_path: &Path, height: u32) -> Result<String, String> {
-    // Promotion moves the selected staged torrent into the canonical
+    // Promotion moves the selected staged torrent into the canonical
-    // filename once the corresponding block has been saved.
+    // filename once the corresponding block has been saved.
-    let canonical_path = canonical_torrent_path(height);
+    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<String,
 
     fs::rename(staged_path, &canonical_path)
         .map_err(|e| format!("Failed to promote staged torrent: {e}"))?;
-
+
-    Ok(canonical_path.to_string_lossy().to_string())
+    Ok(canonical_path.to_string_lossy().to_string())
-}
+}
+
+pub async fn save_canonical_torrent_bytes(
+    height: u32,
+    torrent_bytes: &[u8],
+) -> Result<String, String> {
+    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())
+}

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<u8>,
     allow_startup_peers: bool,
     db: Db,
     verification_service: Arc<VerificationService>,
@@ -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,