79 lines
2.9 KiB
Rust
79 lines
2.9 KiB
Rust
|
use crate::records::memory::response_channels::{reserve_entry, Command};
|
||
|
|
use crate::rpc::command_maps::RPC_TORRENT_CANDIDATES;
|
||
|
|
use crate::rpc::responses::RpcResponse;
|
||
|
|
use crate::torrent::torrenting_system::save_torrent::save_staged_torrent;
|
||
|
|
use crate::{timeout, Arc, Duration, Mutex, TcpStream};
|
||
|
|
|
||
|
|
pub async fn hydrate_torrent_candidates(
|
||
|
|
stream: Arc<Mutex<TcpStream>>,
|
||
|
|
map: Arc<Mutex<Command>>,
|
||
|
|
connections_key: String,
|
||
|
|
) -> Result<usize, String> {
|
||
|
|
// Reserve a reply slot and send a small request packet asking the peer for
|
||
|
|
// its staged/local torrent candidates.
|
||
|
|
let (hashmap_key, _tx, rx) = reserve_entry(map.clone()).await;
|
||
|
|
let mut message = Vec::with_capacity(4);
|
||
|
|
message.push(RPC_TORRENT_CANDIDATES);
|
||
|
|
message.extend_from_slice(&hashmap_key);
|
||
|
|
|
||
|
|
RpcResponse::send_raw(&stream, Some(&connections_key), &message).await;
|
||
|
|
|
||
|
|
let response = {
|
||
|
|
let mut rx = rx.lock().await;
|
||
|
|
timeout(Duration::from_secs(30), rx.recv())
|
||
|
|
.await
|
||
|
|
.map_err(|_| "Timed out waiting for torrent candidates".to_string())?
|
||
|
|
.ok_or_else(|| "Torrent candidate response channel closed".to_string())?
|
||
|
|
};
|
||
|
|
|
||
|
|
if response.len() < 4 {
|
||
|
|
return Err("Torrent candidate response was too short".to_string());
|
||
|
|
}
|
||
|
|
|
||
|
|
// The response starts with the number of entries, then each entry is
|
||
|
|
// height + byte length + raw torrent bytes.
|
||
|
|
let mut offset = 0_usize;
|
||
|
|
let candidate_count = u32::from_le_bytes(
|
||
|
|
response[offset..offset + 4]
|
||
|
|
.try_into()
|
||
|
|
.map_err(|_| "Failed to read torrent candidate count".to_string())?,
|
||
|
|
);
|
||
|
|
offset += 4;
|
||
|
|
|
||
|
|
let mut imported = 0_usize;
|
||
|
|
for _ in 0..candidate_count {
|
||
|
|
// Refuse truncated entries rather than trying to save partial torrent
|
||
|
|
// bytes into staging.
|
||
|
|
if response.len().saturating_sub(offset) < 8 {
|
||
|
|
return Err("Torrent candidate entry was truncated".to_string());
|
||
|
|
}
|
||
|
|
|
||
|
|
let height = u32::from_le_bytes(
|
||
|
|
response[offset..offset + 4]
|
||
|
|
.try_into()
|
||
|
|
.map_err(|_| "Failed to read torrent candidate height".to_string())?,
|
||
|
|
);
|
||
|
|
offset += 4;
|
||
|
|
|
||
|
|
let torrent_len = u32::from_le_bytes(
|
||
|
|
response[offset..offset + 4]
|
||
|
|
.try_into()
|
||
|
|
.map_err(|_| "Failed to read torrent candidate length".to_string())?,
|
||
|
|
) as usize;
|
||
|
|
offset += 4;
|
||
|
|
|
||
|
|
if response.len().saturating_sub(offset) < torrent_len {
|
||
|
|
return Err("Torrent candidate bytes were truncated".to_string());
|
||
|
|
}
|
||
|
|
|
||
|
|
let torrent_bytes = &response[offset..offset + torrent_len];
|
||
|
|
// Imported candidates are staged locally; normal orphan replay decides
|
||
|
|
// later whether any of them should become canonical blocks.
|
||
|
|
save_staged_torrent(height, torrent_bytes).await?;
|
||
|
|
imported += 1;
|
||
|
|
offset += torrent_len;
|
||
|
|
}
|
||
|
|
|
||
|
|
Ok(imported)
|
||
|
|
}
|