From 3cfd9906e2b324e476f3ad8d70caf08e989e072d Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 27 May 2026 10:38:01 -0700 Subject: [PATCH 1/5] feat: support in-file piece storage --- .../libtortillas/src/pieces/piece_manager.rs | 185 +++++++++++++++--- crates/libtortillas/src/torrent/piece_flow.rs | 98 +++++++--- crates/libtortillas/src/torrent/util.rs | 13 ++ 3 files changed, 241 insertions(+), 55 deletions(-) diff --git a/crates/libtortillas/src/pieces/piece_manager.rs b/crates/libtortillas/src/pieces/piece_manager.rs index be7a1012..1c3c76e5 100644 --- a/crates/libtortillas/src/pieces/piece_manager.rs +++ b/crates/libtortillas/src/pieces/piece_manager.rs @@ -5,7 +5,7 @@ use async_trait::async_trait; use bytes::Bytes; use tokio::{ fs::{OpenOptions, create_dir_all}, - io::{AsyncSeekExt, AsyncWriteExt}, + io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt}, }; use tracing::trace; @@ -140,38 +140,40 @@ impl FilePieceManager { pub fn path(&self) -> Option<&PathBuf> { self.0.as_ref() } -} -#[async_trait] -impl PieceManager for FilePieceManager { - fn info(&self) -> Option<&Info> { - self.1.as_ref() + pub(crate) async fn write_block( + &self, index: usize, offset: usize, block: Bytes, + ) -> anyhow::Result<()> { + self.write_piece_range(index, offset, &block).await } - async fn pre_start(&mut self, info_dict: Info) -> anyhow::Result<()> { - let base_path = self - .0 - .as_ref() - .ok_or_else(|| anyhow::anyhow!("path must be set before pre_start"))?; - - let info_hash = info_dict.hash()?; - - trace!(torrent_id = %info_hash, path = %base_path.display(), "Pre-starting piece manager"); + pub(crate) async fn read_piece(&self, index: usize) -> anyhow::Result { + let piece_bounds = self.piece_to_paths(index)?; + let piece_len = piece_bounds.iter().map(|(_, _, len)| len).sum(); + self.read_piece_range(index, 0, piece_len).await + } - self.1 = Some(info_dict); - Ok(()) + pub(crate) async fn read_piece_block( + &self, index: usize, offset: usize, length: usize, + ) -> anyhow::Result { + self.read_piece_range(index, offset, length).await } - #[allow(unused)] - async fn recv(&self, index: usize, data: Bytes) -> anyhow::Result<()> { + async fn write_piece_range( + &self, index: usize, offset: usize, data: &[u8], + ) -> anyhow::Result<()> { let base_path = self.path().ok_or_else(|| anyhow::anyhow!("path not set"))?; - let _info = self.info().ok_or_else(|| anyhow::anyhow!("info not set"))?; - - let piece_bounds = self.piece_to_paths(index)?; - let mut data_offset = 0; // Track how much of `data` we've consumed + let mut remaining = data.len(); + let mut skipped = offset; + let mut data_offset = 0; + + for (path, file_offset, len) in self.piece_to_paths(index)? { + if skipped >= len { + skipped -= len; + continue; + } - for (path, file_offset, len) in piece_bounds { - // Ensure parent directories exist + let write_len = remaining.min(len - skipped); let full_path = base_path.join(&path); if let Some(parent) = full_path.parent() { create_dir_all(parent).await?; @@ -183,28 +185,121 @@ impl PieceManager for FilePieceManager { .truncate(false) .open(&full_path) .await?; + file + .seek(SeekFrom::Start((file_offset + skipped) as u64)) + .await?; + file + .write_all(&data[data_offset..data_offset + write_len]) + .await?; - // Position file correctly - file.seek(SeekFrom::Start(file_offset as u64)).await?; + remaining -= write_len; + data_offset += write_len; + skipped = 0; - // Write correct slice from the piece buffer + if remaining == 0 { + return Ok(()); + } + } + + anyhow::bail!("piece range exceeds file lengths") + } + + async fn read_piece_range( + &self, index: usize, offset: usize, length: usize, + ) -> anyhow::Result { + let base_path = self.path().ok_or_else(|| anyhow::anyhow!("path not set"))?; + let mut remaining = length; + let mut skipped = offset; + let mut data = Vec::with_capacity(length); + + for (path, file_offset, len) in self.piece_to_paths(index)? { + if skipped >= len { + skipped -= len; + continue; + } + + let read_len = remaining.min(len - skipped); + let full_path = base_path.join(&path); + let mut file = OpenOptions::new().read(true).open(&full_path).await?; file - .write_all(&data[data_offset..data_offset + len]) + .seek(SeekFrom::Start((file_offset + skipped) as u64)) .await?; - data_offset += len; + let start = data.len(); + data.resize(start + read_len, 0); + file.read_exact(&mut data[start..]).await?; - trace!(index, offset = file_offset, len, path = %path.display(), "Wrote piece block to file"); + remaining -= read_len; + skipped = 0; + + if remaining == 0 { + return Ok(data.into()); + } } + anyhow::bail!("piece range exceeds file lengths") + } +} + +#[async_trait] +impl PieceManager for FilePieceManager { + fn info(&self) -> Option<&Info> { + self.1.as_ref() + } + + async fn pre_start(&mut self, info_dict: Info) -> anyhow::Result<()> { + let base_path = self + .0 + .as_ref() + .ok_or_else(|| anyhow::anyhow!("path must be set before pre_start"))?; + + let info_hash = info_dict.hash()?; + + trace!(torrent_id = %info_hash, path = %base_path.display(), "Pre-starting piece manager"); + + self.1 = Some(info_dict); Ok(()) } + + #[allow(unused)] + async fn recv(&self, index: usize, data: Bytes) -> anyhow::Result<()> { + let _info = self.info().ok_or_else(|| anyhow::anyhow!("info not set"))?; + self.write_piece_range(index, 0, &data).await + } } #[cfg(test)] mod tests { use super::*; - use crate::{prelude::MetaInfo, testing}; + use crate::{ + hashes::{Hash, HashVec}, + prelude::MetaInfo, + testing, + }; + + fn piece_hash(data: &[u8]) -> Hash<20> { + use sha1::{Digest, Sha1}; + + let mut hasher = Sha1::new(); + hasher.update(data); + Hash::from_bytes(hasher.finalize().into()) + } + + fn single_file_info() -> Info { + Info { + name: "data.bin".to_string(), + piece_length: 4, + pieces: HashVec::from(vec![piece_hash(b"abcd"), piece_hash(b"ef")]), + file: InfoKeys::Single { + length: 6, + md5sum: None, + }, + is_private: None, + publisher: None, + publisher_url: None, + source: None, + } + } #[tokio::test] async fn file_piece_manager_when_mapping_first_piece_then_returns_expected_file() { @@ -225,4 +320,30 @@ mod tests { // The first file should always be "Big Buck Bunny.en.srt" assert_eq!(file_path.extension().unwrap(), "srt"); } + + #[tokio::test] + async fn file_piece_manager_when_writing_blocks_then_reads_piece_back() { + let base_path = testing::torrent_temp_path(); + let manager = FilePieceManager(Some(base_path.clone()), Some(single_file_info())); + + manager + .write_block(0, 0, Bytes::from_static(b"ab")) + .await + .unwrap(); + manager + .write_block(0, 2, Bytes::from_static(b"cd")) + .await + .unwrap(); + + assert_eq!( + manager.read_piece(0).await.unwrap(), + Bytes::from_static(b"abcd") + ); + assert_eq!( + manager.read_piece_block(0, 1, 2).await.unwrap(), + Bytes::from_static(b"bc") + ); + + tokio::fs::remove_dir_all(base_path).await.unwrap(); + } } diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index cba8bd81..709a0444 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -7,7 +7,7 @@ use tokio::{ }; use tracing::{debug, info, trace, warn}; -use super::TorrentActor; +use super::{TorrentActor, util}; use crate::{ errors::TorrentError, peer::commands::{CancelPiece, Have, NeedPiece}, @@ -182,13 +182,29 @@ impl TorrentActor { } } } - PieceStorageStrategy::InFile => { - warn!( - piece_index = index, - offset, "In-file block writes are not implemented yet" - ); - false - } + PieceStorageStrategy::InFile => match &self.piece_manager { + crate::torrent::actor::PieceManagerProxy::Default(manager) => { + match manager.write_block(index, offset, block).await { + Ok(()) => true, + Err(err) => { + warn!( + ?err, + piece_index = index, + offset, + "Failed to write in-file block" + ); + false + } + } + } + crate::torrent::actor::PieceManagerProxy::Custom(_) => { + warn!( + piece_index = index, + offset, "In-file storage requires the default piece manager" + ); + false + } + }, } } @@ -286,11 +302,40 @@ impl TorrentActor { } } PieceStorageStrategy::InFile => { - warn!( - index, - "In-file piece validation is not implemented yet; re-requesting" - ); - return false; + let crate::torrent::actor::PieceManagerProxy::Default(manager) = &self.piece_manager + else { + warn!( + index, + "In-file storage requires the default piece manager; re-requesting" + ); + return false; + }; + + let data = match manager.read_piece(index).await { + Ok(data) => data, + Err(err) => { + warn!(?err, index, "Failed to read in-file piece; re-requesting"); + if let Some(blocks) = previous_blocks.as_ref() { + self + .piece_scheduler + .restore_piece_blocks(index, blocks.clone()); + } + self.request_blocks_from_peer(peer_id, 1).await; + return false; + } + }; + + if let Err(err) = util::validate_piece_bytes(&data, info_dict.pieces[index]) { + warn!( + ?err, + index, "Failed to validate in-file piece; re-requesting" + ); + if let Some(blocks) = previous_blocks { + self.piece_scheduler.restore_piece_blocks(index, blocks); + } + self.request_blocks_from_peer(peer_id, 1).await; + return false; + } } } true @@ -322,15 +367,22 @@ impl TorrentActor { pub(super) async fn read_piece_block( &self, index: usize, offset: usize, length: usize, ) -> anyhow::Result { - anyhow::ensure!( - matches!(self.piece_storage, PieceStorageStrategy::Disk(_)), - "in-file piece reads are not implemented yet" - ); - - let mut file = File::open(self.get_piece_path(index)?).await?; - let mut buffer = vec![0; length]; - file.seek(SeekFrom::Start(offset as u64)).await?; - file.read_exact(&mut buffer).await?; - Ok(buffer.into()) + match &self.piece_storage { + PieceStorageStrategy::Disk(_) => { + let mut file = File::open(self.get_piece_path(index)?).await?; + let mut buffer = vec![0; length]; + file.seek(SeekFrom::Start(offset as u64)).await?; + file.read_exact(&mut buffer).await?; + Ok(buffer.into()) + } + PieceStorageStrategy::InFile => match &self.piece_manager { + crate::torrent::actor::PieceManagerProxy::Default(manager) => { + manager.read_piece_block(index, offset, length).await + } + crate::torrent::actor::PieceManagerProxy::Custom(_) => { + anyhow::bail!("in-file reads require the default piece manager") + } + }, + } } } diff --git a/crates/libtortillas/src/torrent/util.rs b/crates/libtortillas/src/torrent/util.rs index 7fb1675d..78be9d52 100644 --- a/crates/libtortillas/src/torrent/util.rs +++ b/crates/libtortillas/src/torrent/util.rs @@ -120,6 +120,19 @@ pub async fn validate_piece_file( Ok(()) } +pub fn validate_piece_bytes(data: &[u8], hash: Hash<20>) -> anyhow::Result<()> { + let mut hasher = Sha1::new(); + hasher.update(data); + let piece_hash = Hash::from_bytes(hasher.finalize().into()); + + ensure!( + piece_hash == hash, + "Hashed bytes were not equal to hash from info dict" + ); + + Ok(()) +} + /// Creates a dir if it doesn't exist. This function can be called even if the /// directory exists -- if it does, nothing will happen. pub async fn create_dir(path: &PathBuf) -> Result<(), Error> { From 88512258db4054f7f5da388dcd5be8cde7507ad4 Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 27 May 2026 10:40:34 -0700 Subject: [PATCH 2/5] test: cover in-file piece flow --- crates/libtortillas/src/torrent/piece_flow.rs | 136 ++++++++++++++++++ 1 file changed, 136 insertions(+) diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index 709a0444..f76168e0 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -386,3 +386,139 @@ impl TorrentActor { } } } + +#[cfg(test)] +mod tests { + use std::{collections::HashMap, sync::atomic::AtomicU8}; + + use bitvec::vec::BitVec; + use bytes::Bytes; + use kameo::actor::Spawn; + use kameo_actors::scheduler::Scheduler; + use librqbit_utp::UtpSocket; + use sha1::{Digest, Sha1}; + + use super::*; + use crate::{ + hashes::{Hash, HashVec}, + metainfo::{Info, InfoKeys, MetaInfo, TorrentFile}, + peer::PeerId, + pieces::{FilePieceManager, PieceScheduler, PieceStoreActor}, + testing, + torrent::actor::{PieceManagerProxy, TorrentActorArgs}, + tracker::{Tracker, udp::UdpServer}, + }; + + fn piece_hash(data: &[u8]) -> Hash<20> { + let mut hasher = Sha1::new(); + hasher.update(data); + Hash::from_bytes(hasher.finalize().into()) + } + + fn test_info() -> Info { + Info { + name: "data.bin".to_string(), + piece_length: 4, + pieces: HashVec::from(vec![piece_hash(b"abcd")]), + file: InfoKeys::Single { + length: 4, + md5sum: None, + }, + is_private: None, + publisher: None, + publisher_url: None, + source: None, + } + } + + fn test_metainfo(info: Info) -> MetaInfo { + MetaInfo::Torrent(TorrentFile { + announce: Tracker::Http("http://127.0.0.1/announce".to_string()), + announce_list: None, + comment: None, + created_by: None, + creation_date: None, + encoding: None, + info, + url_list: None, + }) + } + + async fn test_actor(base_path: std::path::PathBuf) -> TorrentActor { + let info = test_info(); + let metainfo = test_metainfo(info.clone()); + let peer_id = testing::peer_id(); + let tracker_server = UdpServer::new(None).await; + let utp_server = UtpSocket::new_udp(testing::ephemeral_socket_addr()) + .await + .unwrap(); + + let actor_ref = TorrentActor::spawn(TorrentActorArgs { + peer_id, + metainfo: metainfo.clone(), + utp_server: utp_server.clone(), + tracker_server: tracker_server.clone(), + primary_addr: None, + piece_storage: PieceStorageStrategy::InFile, + autostart: Some(false), + sufficient_peers: Some(usize::MAX), + base_path: Some(base_path.clone()), + }); + + TorrentActor { + peers: HashMap::new(), + trackers: HashMap::new(), + bitfield: BitVec::::repeat(false, info.piece_count()), + id: peer_id, + info: Some(info.clone()), + metainfo, + tracker_server, + scheduler: Scheduler::spawn(Scheduler::new()), + utp_server, + actor_ref, + piece_storage: PieceStorageStrategy::InFile, + piece_store: PieceStoreActor::spawn(()), + piece_manager: PieceManagerProxy::Default(FilePieceManager(Some(base_path), Some(info))), + state: TorrentState::Downloading, + piece_scheduler: PieceScheduler::new(1), + start_time: None, + sufficient_peers: 6, + autostart: false, + pending_start: false, + ready_hook: Vec::new(), + } + } + + #[tokio::test] + async fn torrent_actor_when_using_in_file_storage_then_downloads_and_reads_back_piece() { + let base_path = testing::torrent_temp_path(); + let mut actor = test_actor(base_path.clone()).await; + + assert!( + actor + .write_block_to_storage(0, 0, Bytes::from_static(b"ab")) + .await + ); + assert!( + actor + .write_block_to_storage(0, 2, Bytes::from_static(b"cd")) + .await + ); + assert!( + actor + .validate_and_send_piece(PeerId::default(), 0, None) + .await + ); + + actor.bitfield.set_aliased(0, true); + assert_eq!( + actor.read_piece_block(0, 1, 2).await.unwrap(), + Bytes::from_static(b"bc") + ); + + actor.actor_ref.kill(); + actor.piece_store.kill(); + actor.scheduler.kill(); + tokio::fs::remove_dir_all(base_path).await.unwrap(); + } +} From bf4c56dcc3c85d875338b30fdeaf789b97317feb Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 27 May 2026 10:43:15 -0700 Subject: [PATCH 3/5] refactor: deduplicate in-file piece ranges --- crates/libtortillas/src/lib.rs | 9 ++ .../libtortillas/src/pieces/piece_manager.rs | 96 ++++++++----------- crates/libtortillas/src/torrent/piece_flow.rs | 11 +-- 3 files changed, 53 insertions(+), 63 deletions(-) diff --git a/crates/libtortillas/src/lib.rs b/crates/libtortillas/src/lib.rs index b869f661..66e9ebff 100644 --- a/crates/libtortillas/src/lib.rs +++ b/crates/libtortillas/src/lib.rs @@ -16,6 +16,7 @@ pub(crate) mod testing { use tokio::fs::read_to_string; use crate::{ + hashes::Hash, metainfo::{MagnetUri, MetaInfo, TorrentFile}, peer::PeerId, tracker::udp::UdpServer, @@ -82,6 +83,14 @@ pub(crate) mod testing { PeerId::default() } + pub(crate) fn piece_hash(data: &[u8]) -> Hash<20> { + use sha1::{Digest, Sha1}; + + let mut hasher = Sha1::new(); + hasher.update(data); + Hash::from_bytes(hasher.finalize().into()) + } + pub(crate) async fn udp_server() -> UdpServer { UdpServer::new(None).await } diff --git a/crates/libtortillas/src/pieces/piece_manager.rs b/crates/libtortillas/src/pieces/piece_manager.rs index 1c3c76e5..f1131ea4 100644 --- a/crates/libtortillas/src/pieces/piece_manager.rs +++ b/crates/libtortillas/src/pieces/piece_manager.rs @@ -159,13 +159,16 @@ impl FilePieceManager { self.read_piece_range(index, offset, length).await } - async fn write_piece_range( - &self, index: usize, offset: usize, data: &[u8], - ) -> anyhow::Result<()> { - let base_path = self.path().ok_or_else(|| anyhow::anyhow!("path not set"))?; - let mut remaining = data.len(); + fn piece_range_segments( + &self, index: usize, offset: usize, length: usize, + ) -> anyhow::Result> { + let mut remaining = length; let mut skipped = offset; - let mut data_offset = 0; + let mut segments = Vec::new(); + + if remaining == 0 { + return Ok(segments); + } for (path, file_offset, len) in self.piece_to_paths(index)? { if skipped >= len { @@ -173,7 +176,26 @@ impl FilePieceManager { continue; } - let write_len = remaining.min(len - skipped); + let segment_len = remaining.min(len - skipped); + segments.push((path, file_offset + skipped, segment_len)); + remaining -= segment_len; + skipped = 0; + + if remaining == 0 { + return Ok(segments); + } + } + + anyhow::bail!("piece range exceeds file lengths") + } + + async fn write_piece_range( + &self, index: usize, offset: usize, data: &[u8], + ) -> anyhow::Result<()> { + let base_path = self.path().ok_or_else(|| anyhow::anyhow!("path not set"))?; + let mut data_offset = 0; + + for (path, file_offset, len) in self.piece_range_segments(index, offset, data.len())? { let full_path = base_path.join(&path); if let Some(parent) = full_path.parent() { create_dir_all(parent).await?; @@ -185,59 +207,34 @@ impl FilePieceManager { .truncate(false) .open(&full_path) .await?; + file.seek(SeekFrom::Start(file_offset as u64)).await?; file - .seek(SeekFrom::Start((file_offset + skipped) as u64)) + .write_all(&data[data_offset..data_offset + len]) .await?; - file - .write_all(&data[data_offset..data_offset + write_len]) - .await?; - - remaining -= write_len; - data_offset += write_len; - skipped = 0; - if remaining == 0 { - return Ok(()); - } + data_offset += len; } - anyhow::bail!("piece range exceeds file lengths") + Ok(()) } async fn read_piece_range( &self, index: usize, offset: usize, length: usize, ) -> anyhow::Result { let base_path = self.path().ok_or_else(|| anyhow::anyhow!("path not set"))?; - let mut remaining = length; - let mut skipped = offset; let mut data = Vec::with_capacity(length); - for (path, file_offset, len) in self.piece_to_paths(index)? { - if skipped >= len { - skipped -= len; - continue; - } - - let read_len = remaining.min(len - skipped); + for (path, file_offset, len) in self.piece_range_segments(index, offset, length)? { let full_path = base_path.join(&path); let mut file = OpenOptions::new().read(true).open(&full_path).await?; - file - .seek(SeekFrom::Start((file_offset + skipped) as u64)) - .await?; + file.seek(SeekFrom::Start(file_offset as u64)).await?; let start = data.len(); - data.resize(start + read_len, 0); + data.resize(start + len, 0); file.read_exact(&mut data[start..]).await?; - - remaining -= read_len; - skipped = 0; - - if remaining == 0 { - return Ok(data.into()); - } } - anyhow::bail!("piece range exceeds file lengths") + Ok(data.into()) } } @@ -271,25 +268,16 @@ impl PieceManager for FilePieceManager { #[cfg(test)] mod tests { use super::*; - use crate::{ - hashes::{Hash, HashVec}, - prelude::MetaInfo, - testing, - }; - - fn piece_hash(data: &[u8]) -> Hash<20> { - use sha1::{Digest, Sha1}; - - let mut hasher = Sha1::new(); - hasher.update(data); - Hash::from_bytes(hasher.finalize().into()) - } + use crate::{hashes::HashVec, prelude::MetaInfo, testing}; fn single_file_info() -> Info { Info { name: "data.bin".to_string(), piece_length: 4, - pieces: HashVec::from(vec![piece_hash(b"abcd"), piece_hash(b"ef")]), + pieces: HashVec::from(vec![ + testing::piece_hash(b"abcd"), + testing::piece_hash(b"ef"), + ]), file: InfoKeys::Single { length: 6, md5sum: None, diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index f76168e0..c36b5ced 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -396,11 +396,10 @@ mod tests { use kameo::actor::Spawn; use kameo_actors::scheduler::Scheduler; use librqbit_utp::UtpSocket; - use sha1::{Digest, Sha1}; use super::*; use crate::{ - hashes::{Hash, HashVec}, + hashes::HashVec, metainfo::{Info, InfoKeys, MetaInfo, TorrentFile}, peer::PeerId, pieces::{FilePieceManager, PieceScheduler, PieceStoreActor}, @@ -409,17 +408,11 @@ mod tests { tracker::{Tracker, udp::UdpServer}, }; - fn piece_hash(data: &[u8]) -> Hash<20> { - let mut hasher = Sha1::new(); - hasher.update(data); - Hash::from_bytes(hasher.finalize().into()) - } - fn test_info() -> Info { Info { name: "data.bin".to_string(), piece_length: 4, - pieces: HashVec::from(vec![piece_hash(b"abcd")]), + pieces: HashVec::from(vec![testing::piece_hash(b"abcd")]), file: InfoKeys::Single { length: 4, md5sum: None, From a039af3f78d916e70c9ce2684bceb9e23e03448f Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 27 May 2026 10:46:31 -0700 Subject: [PATCH 4/5] test: cover piece storage strategies --- crates/libtortillas/src/torrent/piece_flow.rs | 53 +++++++++++++++++-- 1 file changed, 49 insertions(+), 4 deletions(-) diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index c36b5ced..62c85007 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -437,7 +437,9 @@ mod tests { }) } - async fn test_actor(base_path: std::path::PathBuf) -> TorrentActor { + async fn test_actor( + piece_storage: PieceStorageStrategy, base_path: std::path::PathBuf, + ) -> TorrentActor { let info = test_info(); let metainfo = test_metainfo(info.clone()); let peer_id = testing::peer_id(); @@ -452,7 +454,7 @@ mod tests { utp_server: utp_server.clone(), tracker_server: tracker_server.clone(), primary_addr: None, - piece_storage: PieceStorageStrategy::InFile, + piece_storage: piece_storage.clone(), autostart: Some(false), sufficient_peers: Some(usize::MAX), base_path: Some(base_path.clone()), @@ -469,7 +471,7 @@ mod tests { scheduler: Scheduler::spawn(Scheduler::new()), utp_server, actor_ref, - piece_storage: PieceStorageStrategy::InFile, + piece_storage, piece_store: PieceStoreActor::spawn(()), piece_manager: PieceManagerProxy::Default(FilePieceManager(Some(base_path), Some(info))), state: TorrentState::Downloading, @@ -485,7 +487,7 @@ mod tests { #[tokio::test] async fn torrent_actor_when_using_in_file_storage_then_downloads_and_reads_back_piece() { let base_path = testing::torrent_temp_path(); - let mut actor = test_actor(base_path.clone()).await; + let mut actor = test_actor(PieceStorageStrategy::InFile, base_path.clone()).await; assert!( actor @@ -514,4 +516,47 @@ mod tests { actor.scheduler.kill(); tokio::fs::remove_dir_all(base_path).await.unwrap(); } + + #[tokio::test] + async fn torrent_actor_when_using_disk_storage_then_keeps_piece_cache_and_writes_output_file() { + let piece_path = testing::torrent_temp_path(); + let base_path = testing::torrent_temp_path(); + let mut actor = test_actor( + PieceStorageStrategy::Disk(piece_path.clone()), + base_path.clone(), + ) + .await; + + assert!( + actor + .write_block_to_storage(0, 0, Bytes::from_static(b"ab")) + .await + ); + assert!( + actor + .write_block_to_storage(0, 2, Bytes::from_static(b"cd")) + .await + ); + assert!( + actor + .validate_and_send_piece(PeerId::default(), 0, None) + .await + ); + + actor.bitfield.set_aliased(0, true); + assert_eq!( + actor.read_piece_block(0, 1, 2).await.unwrap(), + Bytes::from_static(b"bc") + ); + assert_eq!( + tokio::fs::read(base_path.join("data.bin")).await.unwrap(), + b"abcd" + ); + + actor.actor_ref.kill(); + actor.piece_store.kill(); + actor.scheduler.kill(); + tokio::fs::remove_dir_all(piece_path).await.unwrap(); + tokio::fs::remove_dir_all(base_path).await.unwrap(); + } } From 12bd0f3b1894374e7ecd526ef76305e0bd0ae86b Mon Sep 17 00:00:00 2001 From: artrixdotdev Date: Wed, 27 May 2026 11:15:34 -0700 Subject: [PATCH 5/5] docs: clarify piece storage flow --- .../libtortillas/src/pieces/piece_manager.rs | 10 +++++++++ crates/libtortillas/src/torrent/piece_flow.rs | 21 ++++++++++++------- 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/crates/libtortillas/src/pieces/piece_manager.rs b/crates/libtortillas/src/pieces/piece_manager.rs index f1131ea4..4ff290a7 100644 --- a/crates/libtortillas/src/pieces/piece_manager.rs +++ b/crates/libtortillas/src/pieces/piece_manager.rs @@ -147,6 +147,11 @@ impl FilePieceManager { self.write_piece_range(index, offset, &block).await } + /// Reads a complete piece back from the final torrent files. + /// + /// This is used by `PieceStorageStrategy::InFile` after all blocks for a + /// piece have arrived, so the torrent actor can hash the exact bytes that + /// were written to disk. pub(crate) async fn read_piece(&self, index: usize) -> anyhow::Result { let piece_bounds = self.piece_to_paths(index)?; let piece_len = piece_bounds.iter().map(|(_, _, len)| len).sum(); @@ -159,6 +164,11 @@ impl FilePieceManager { self.read_piece_range(index, offset, length).await } + /// Converts a piece-relative byte range into file-relative segments. + /// + /// A single BitTorrent piece can cross file boundaries in multi-file + /// torrents. This helper centralizes that mapping so in-file reads and + /// writes cannot drift apart. fn piece_range_segments( &self, index: usize, offset: usize, length: usize, ) -> anyhow::Result> { diff --git a/crates/libtortillas/src/torrent/piece_flow.rs b/crates/libtortillas/src/torrent/piece_flow.rs index 62c85007..92ed1d5b 100644 --- a/crates/libtortillas/src/torrent/piece_flow.rs +++ b/crates/libtortillas/src/torrent/piece_flow.rs @@ -12,7 +12,7 @@ use crate::{ errors::TorrentError, peer::commands::{CancelPiece, Have, NeedPiece}, pieces::{PieceManager, ValidateAndRead, WriteBlock}, - torrent::{BLOCK_SIZE, PieceStorageStrategy, TorrentState}, + torrent::{BLOCK_SIZE, PieceStorageStrategy, TorrentState, actor::PieceManagerProxy}, tracker::{Announce, Event, TrackerUpdate}, }; @@ -157,6 +157,8 @@ impl TorrentActor { async fn write_block_to_storage(&self, index: usize, offset: usize, block: Bytes) -> bool { match &self.piece_storage { PieceStorageStrategy::Disk(_) => { + // Disk storage is the piece-cache strategy: blocks are assembled in + // standalone `.piece` files before being flushed to output files. let Ok(path) = self.get_piece_path(index) else { warn!(piece_index = index, "Failed to get piece path"); return false; @@ -183,7 +185,9 @@ impl TorrentActor { } } PieceStorageStrategy::InFile => match &self.piece_manager { - crate::torrent::actor::PieceManagerProxy::Default(manager) => { + // In-file storage skips the `.piece` cache and writes the incoming + // block straight to its final file offset. + PieceManagerProxy::Default(manager) => { match manager.write_block(index, offset, block).await { Ok(()) => true, Err(err) => { @@ -197,7 +201,7 @@ impl TorrentActor { } } } - crate::torrent::actor::PieceManagerProxy::Custom(_) => { + PieceManagerProxy::Custom(_) => { warn!( piece_index = index, offset, "In-file storage requires the default piece manager" @@ -261,6 +265,8 @@ impl TorrentActor { match &self.piece_storage { PieceStorageStrategy::Disk(_) => { + // Cached pieces are validated as standalone files, then handed to + // the piece manager to populate the final output files. let Ok(path) = self.get_piece_path(index) else { warn!(index, "Failed to get piece path; re-requesting"); if let Some(blocks) = previous_blocks.as_ref() { @@ -302,8 +308,9 @@ impl TorrentActor { } } PieceStorageStrategy::InFile => { - let crate::torrent::actor::PieceManagerProxy::Default(manager) = &self.piece_manager - else { + // Blocks were already written into output files, so validation must + // hash the piece bytes read back from those files. + let PieceManagerProxy::Default(manager) = &self.piece_manager else { warn!( index, "In-file storage requires the default piece manager; re-requesting" @@ -376,10 +383,10 @@ impl TorrentActor { Ok(buffer.into()) } PieceStorageStrategy::InFile => match &self.piece_manager { - crate::torrent::actor::PieceManagerProxy::Default(manager) => { + PieceManagerProxy::Default(manager) => { manager.read_piece_block(index, offset, length).await } - crate::torrent::actor::PieceManagerProxy::Custom(_) => { + PieceManagerProxy::Custom(_) => { anyhow::bail!("in-file reads require the default piece manager") } },