diff --git a/Cargo.lock b/Cargo.lock index 59714b5..a8288af 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -369,6 +369,29 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" +[[package]] +name = "futures-io" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" + +[[package]] +name = "futures-macro" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", +] + +[[package]] +name = "futures-sink" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" + [[package]] name = "futures-task" version = "0.3.31" @@ -382,7 +405,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" dependencies = [ "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "pin-utils", "slab", @@ -782,8 +809,10 @@ dependencies = [ "anyhow", "axum", "bendy", + "bytes", "cid", "crc32fast", + "futures-util", "hex", "md-5", "multihash", @@ -1076,6 +1105,7 @@ dependencies = [ "base64", "bytes", "futures-core", + "futures-util", "http", "http-body", "http-body-util", @@ -1095,12 +1125,14 @@ dependencies = [ "sync_wrapper", "tokio", "tokio-rustls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", "webpki-roots", ] @@ -1513,6 +1545,19 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "tower" version = "0.5.3" @@ -1771,6 +1816,19 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-streams" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "web-sys" version = "0.3.85" diff --git a/Cargo.toml b/Cargo.toml index cf02528..1efe79a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,7 +31,11 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" # HTTP client -reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls", "stream"] } + +# Async streaming +futures-util = "0.3" +bytes = "1.5" # Utilities tracing = "0.1" diff --git a/src/api/process.rs b/src/api/process.rs index 7d85fd6..ab49cea 100644 --- a/src/api/process.rs +++ b/src/api/process.rs @@ -8,8 +8,8 @@ use std::time::Instant; use crate::cache::CacheManager; use crate::client::MetaCoreClient; use crate::config::{HashAlgorithm, SharedConfig}; -use crate::hasher::{hash_bytes, hash_file}; -use crate::webdav::WebDavClient; +use crate::hasher::{hash_file, hash_stream}; +use crate::webdav::{WebDavClient, WebDavReader}; #[derive(Deserialize)] #[serde(rename_all = "camelCase")] @@ -115,7 +115,7 @@ enum ProcessResult { Skipped(String), } -/// File source - either local filesystem or WebDAV +/// File source - either local filesystem or WebDAV stream enum FileSource { Local { path: std::path::PathBuf, @@ -123,8 +123,8 @@ enum FileSource { mtime: i64, }, WebDav { - data: Vec, - size: u64, + reader: WebDavReader, + size: Option, }, } @@ -160,19 +160,18 @@ async fn do_process_file( mtime, } } else { - // Try WebDAV + // Try WebDAV - stream instead of loading into memory let webdav_url = std::env::var("WEBDAV_URL") .map_err(|_| anyhow::anyhow!("File not found locally and WEBDAV_URL not set: {}", req.file_path))?; - tracing::info!("Fetching via WebDAV: {}", req.file_path); + tracing::info!("Streaming via WebDAV: {}", req.file_path); let client = WebDavClient::new(&webdav_url); let reader = client.get_file(&req.file_path).await?; - let size = reader.size().unwrap_or(0); - let data = reader.bytes().await?; + let size = reader.size(); - tracing::info!("WebDAV fetch complete: {} bytes", data.len()); + tracing::info!("WebDAV stream ready, size: {:?}", size); FileSource::WebDav { - data, + reader, size, } }; @@ -180,7 +179,7 @@ async fn do_process_file( // Get file size and mtime for cache lookup let (size, mtime) = match &source { FileSource::Local { size, mtime, .. } => (*size, *mtime), - FileSource::WebDav { size, .. } => (*size, 0), // No mtime for WebDAV + FileSource::WebDav { size, .. } => (size.unwrap_or(0), 0), // No mtime for WebDAV }; // Get config @@ -243,11 +242,11 @@ async fn do_process_file( tokio::task::spawn_blocking(move || hash_file(&path, &enabled_set, buffer_size)) .await?? } - FileSource::WebDav { data, .. } => { - // Use bytes-based hashing for WebDAV data + FileSource::WebDav { reader, size: file_size } => { + // Use streaming hashing for WebDAV - process chunks as they arrive + let stream = reader.bytes_stream(); let filename_clone = filename.clone(); - tokio::task::spawn_blocking(move || hash_bytes(&data, &filename_clone, &enabled_set)) - .await?? + hash_stream(stream, &filename_clone, file_size, &enabled_set).await? } }; diff --git a/src/hasher/mod.rs b/src/hasher/mod.rs index 7a43265..81ac21e 100644 --- a/src/hasher/mod.rs +++ b/src/hasher/mod.rs @@ -2,8 +2,10 @@ mod algorithms; pub mod btih_v2; mod cid; pub mod midhash256; +mod stream; pub use algorithms::{hash_bytes, hash_file}; +pub use stream::hash_stream; pub use btih_v2::{ compute_btv2_hashes, compute_info_hash, compute_merkle_root, format_magnet_link, format_magnet_urn, BtV2Hashes, BT_BLOCK_SIZE, diff --git a/src/hasher/stream.rs b/src/hasher/stream.rs new file mode 100644 index 0000000..d27fa98 --- /dev/null +++ b/src/hasher/stream.rs @@ -0,0 +1,391 @@ +use crate::config::HashAlgorithm; +use anyhow::Result; +use bytes::Bytes; +use futures_util::{Stream, StreamExt}; +use md5::Md5; +use sha1::Sha1; +use sha2::{Digest, Sha256}; +use sha3::{Sha3_256, Sha3_384}; +use std::collections::{HashMap, HashSet}; + +use super::btih_v2::{compute_info_hash, compute_merkle_root, BT_BLOCK_SIZE}; +use super::cid::to_cid; + +/// Hash data from an async byte stream +/// +/// Processes chunks as they arrive from the network, maintaining O(1) memory +/// regardless of file size. All hashers are updated incrementally. +pub async fn hash_stream( + mut stream: S, + filename: &str, + file_size: Option, + enabled: &HashSet, +) -> Result> +where + S: Stream> + Unpin, + E: std::error::Error + Send + Sync + 'static, +{ + let mut results = HashMap::new(); + + let needs_bt = enabled.contains(&HashAlgorithm::BtPiecesRoot) + || enabled.contains(&HashAlgorithm::BtInfoHash); + let needs_midhash = enabled.contains(&HashAlgorithm::Midhash256); + + // Initialize standard hashers + let mut sha256 = enabled.contains(&HashAlgorithm::Sha256).then(Sha256::new); + let mut sha1 = enabled.contains(&HashAlgorithm::Sha1).then(Sha1::new); + let mut md5 = enabled.contains(&HashAlgorithm::Md5).then(Md5::new); + let mut crc32 = enabled + .contains(&HashAlgorithm::Crc32) + .then(crc32fast::Hasher::new); + let mut sha3_256 = enabled + .contains(&HashAlgorithm::Sha3_256) + .then(Sha3_256::new); + let mut sha3_384 = enabled + .contains(&HashAlgorithm::Sha3_384) + .then(Sha3_384::new); + + // BT v2 state + let mut bt_leaf_hashes: Vec<[u8; 32]> = Vec::new(); + let mut bt_block_buffer = vec![0u8; BT_BLOCK_SIZE]; + let mut bt_block_offset = 0usize; + + // Midhash256 state - we need to buffer the middle section + // This is the one algorithm that can't be purely streaming + const MIDHASH_SAMPLE_SIZE: usize = 1024 * 1024; // 1MB + let mut midhash_collector: Option = if needs_midhash { + Some(MidhashCollector::new(file_size, MIDHASH_SAMPLE_SIZE)) + } else { + None + }; + + let mut bytes_read: u64 = 0; + let known_size = file_size.unwrap_or(0); + + // Process stream chunks as they arrive + while let Some(chunk_result) = stream.next().await { + let chunk = chunk_result.map_err(|e| anyhow::anyhow!("Stream error: {}", e))?; + let chunk = chunk.as_ref(); + let chunk_start = bytes_read; + bytes_read += chunk.len() as u64; + + // Update all standard hashers + if let Some(ref mut h) = sha256 { + h.update(chunk); + } + if let Some(ref mut h) = sha1 { + h.update(chunk); + } + if let Some(ref mut h) = md5 { + h.update(chunk); + } + if let Some(ref mut h) = crc32 { + h.update(chunk); + } + if let Some(ref mut h) = sha3_256 { + h.update(chunk); + } + if let Some(ref mut h) = sha3_384 { + h.update(chunk); + } + + // BT v2: Process chunk into 16KB blocks + if needs_bt { + let mut chunk_offset = 0; + while chunk_offset < chunk.len() { + let space_in_block = BT_BLOCK_SIZE - bt_block_offset; + let bytes_to_copy = space_in_block.min(chunk.len() - chunk_offset); + + bt_block_buffer[bt_block_offset..bt_block_offset + bytes_to_copy] + .copy_from_slice(&chunk[chunk_offset..chunk_offset + bytes_to_copy]); + + bt_block_offset += bytes_to_copy; + chunk_offset += bytes_to_copy; + + if bt_block_offset == BT_BLOCK_SIZE { + let mut hasher = Sha256::new(); + hasher.update(&bt_block_buffer); + bt_leaf_hashes.push(hasher.finalize().into()); + bt_block_offset = 0; + } + } + } + + // Midhash: collect the middle section + if let Some(ref mut collector) = midhash_collector { + collector.process_chunk(chunk, chunk_start); + } + + // Progress logging + if known_size > 100 * 1024 * 1024 && bytes_read % (100 * 1024 * 1024) < chunk.len() as u64 { + let pct = (bytes_read as f64 / known_size as f64 * 100.0) as u32; + tracing::debug!( + "Stream hashing progress: {}% ({}/{})", + pct, + bytes_read, + known_size + ); + } + } + + let actual_size = bytes_read; + + // Finalize BT v2 + if needs_bt { + if actual_size == 0 { + bt_leaf_hashes.push([0u8; 32]); + } else if bt_block_offset > 0 { + bt_block_buffer[bt_block_offset..].fill(0); + let mut hasher = Sha256::new(); + hasher.update(&bt_block_buffer); + bt_leaf_hashes.push(hasher.finalize().into()); + } + + let pieces_root = compute_merkle_root(&bt_leaf_hashes); + + if enabled.contains(&HashAlgorithm::BtPiecesRoot) { + results.insert( + HashAlgorithm::BtPiecesRoot.key().to_string(), + to_cid(&pieces_root, HashAlgorithm::BtPiecesRoot), + ); + } + + if enabled.contains(&HashAlgorithm::BtInfoHash) { + if let Ok(info_hash) = compute_info_hash(filename, actual_size, &pieces_root) { + results.insert( + HashAlgorithm::BtInfoHash.key().to_string(), + to_cid(&info_hash, HashAlgorithm::BtInfoHash), + ); + } + } + } + + // Finalize midhash256 + if let Some(collector) = midhash_collector { + let hash = collector.finalize(actual_size); + results.insert( + HashAlgorithm::Midhash256.key().to_string(), + to_cid(&hash, HashAlgorithm::Midhash256), + ); + } + + // Finalize standard hashes + if let Some(h) = sha256 { + results.insert( + HashAlgorithm::Sha256.key().to_string(), + to_cid(&h.finalize(), HashAlgorithm::Sha256), + ); + } + if let Some(h) = sha1 { + results.insert( + HashAlgorithm::Sha1.key().to_string(), + to_cid(&h.finalize(), HashAlgorithm::Sha1), + ); + } + if let Some(h) = md5 { + results.insert( + HashAlgorithm::Md5.key().to_string(), + to_cid(&h.finalize(), HashAlgorithm::Md5), + ); + } + if let Some(h) = crc32 { + results.insert( + HashAlgorithm::Crc32.key().to_string(), + to_cid(&h.finalize().to_be_bytes(), HashAlgorithm::Crc32), + ); + } + if let Some(h) = sha3_256 { + results.insert( + HashAlgorithm::Sha3_256.key().to_string(), + to_cid(&h.finalize(), HashAlgorithm::Sha3_256), + ); + } + if let Some(h) = sha3_384 { + results.insert( + HashAlgorithm::Sha3_384.key().to_string(), + to_cid(&h.finalize(), HashAlgorithm::Sha3_384), + ); + } + + Ok(results) +} + +/// Collects the middle section of a file for midhash256 +/// +/// If file size is known upfront, only buffers the exact middle 1MB. +/// If unknown, buffers more conservatively and extracts middle at the end. +struct MidhashCollector { + /// Sample size (1MB) + sample_size: usize, + /// Buffered data for middle section + buffer: Vec, + /// Start offset of middle section (calculated from known size) + middle_start: Option, + /// Have we finished collecting? + done: bool, +} + +impl MidhashCollector { + fn new(known_size: Option, sample_size: usize) -> Self { + let middle_start = known_size.and_then(|size| { + if (size as usize) <= sample_size { + None // Small file, collect everything + } else { + Some((size - sample_size as u64) / 2) + } + }); + + // Pre-allocate based on whether we know the size + let capacity = if known_size.is_some() { + sample_size + } else { + // Unknown size - we'll need to buffer everything + // Start with a reasonable default + sample_size * 2 + }; + + Self { + sample_size, + buffer: Vec::with_capacity(capacity), + middle_start, + done: false, + } + } + + fn process_chunk(&mut self, chunk: &[u8], chunk_start: u64) { + if self.done { + return; + } + + match self.middle_start { + Some(middle_start) => { + // We know the file size - only collect the middle section + let middle_end = middle_start + self.sample_size as u64; + let chunk_end = chunk_start + chunk.len() as u64; + + // Check if this chunk overlaps with our target range + if chunk_end <= middle_start || chunk_start >= middle_end { + return; // No overlap + } + + // Calculate overlap within the chunk + let copy_start = if chunk_start < middle_start { + (middle_start - chunk_start) as usize + } else { + 0 + }; + let copy_end = if chunk_end > middle_end { + chunk.len() - (chunk_end - middle_end) as usize + } else { + chunk.len() + }; + + if copy_start < chunk.len() && copy_end > copy_start { + self.buffer.extend_from_slice(&chunk[copy_start..copy_end]); + } + + if self.buffer.len() >= self.sample_size { + self.done = true; + } + } + None => { + // Unknown size or small file - collect everything + // We'll extract the middle at the end + self.buffer.extend_from_slice(chunk); + } + } + } + + fn finalize(self, actual_size: u64) -> [u8; 32] { + let mut hasher = Sha256::new(); + hasher.update(actual_size.to_le_bytes()); + + if (actual_size as usize) <= self.sample_size { + // Small file: hash entire content + hasher.update(&self.buffer); + } else if self.middle_start.is_some() { + // We collected exactly the middle + hasher.update(&self.buffer); + } else { + // We collected everything, extract middle now + let middle_start = (self.buffer.len() - self.sample_size) / 2; + hasher.update(&self.buffer[middle_start..middle_start + self.sample_size]); + } + + hasher.finalize().into() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use futures_util::stream; + + #[tokio::test] + async fn test_hash_stream_basic() { + let data = b"hello world"; + let chunks: Vec> = + vec![Ok(Bytes::from_static(data))]; + let stream = stream::iter(chunks); + + let enabled: HashSet<_> = vec![HashAlgorithm::Sha256, HashAlgorithm::Md5] + .into_iter() + .collect(); + + let results = hash_stream(stream, "test.txt", Some(data.len() as u64), &enabled) + .await + .unwrap(); + + assert!(results.contains_key("cid_sha2-256")); + assert!(results.contains_key("cid_md5")); + } + + #[tokio::test] + async fn test_hash_stream_chunked() { + // Split "hello world" into multiple chunks + let chunks: Vec> = vec![ + Ok(Bytes::from_static(b"hello")), + Ok(Bytes::from_static(b" ")), + Ok(Bytes::from_static(b"world")), + ]; + let stream = stream::iter(chunks); + + let enabled: HashSet<_> = vec![HashAlgorithm::Sha256].into_iter().collect(); + + let results = hash_stream(stream, "test.txt", Some(11), &enabled) + .await + .unwrap(); + + // Single chunk version + let single_chunks: Vec> = + vec![Ok(Bytes::from_static(b"hello world"))]; + let single_stream = stream::iter(single_chunks); + + let single_results = hash_stream(single_stream, "test.txt", Some(11), &enabled) + .await + .unwrap(); + + // Results should match regardless of chunking + assert_eq!( + results.get("cid_sha2-256"), + single_results.get("cid_sha2-256") + ); + } + + #[tokio::test] + async fn test_hash_stream_empty() { + let chunks: Vec> = vec![]; + let stream = stream::iter(chunks); + + let enabled: HashSet<_> = vec![HashAlgorithm::Sha256, HashAlgorithm::BtPiecesRoot] + .into_iter() + .collect(); + + let results = hash_stream(stream, "empty.txt", Some(0), &enabled) + .await + .unwrap(); + + assert!(results.contains_key("cid_sha2-256")); + assert!(results.contains_key("cid_bt_pieces_root")); + } +} diff --git a/src/webdav.rs b/src/webdav.rs index 75796cb..daaee89 100644 --- a/src/webdav.rs +++ b/src/webdav.rs @@ -1,4 +1,6 @@ use anyhow::{Context, Result}; +use bytes::Bytes; +use futures_util::Stream; use reqwest::Client; /// WebDAV client for streaming file access @@ -87,6 +89,12 @@ impl WebDavReader { .context("Failed to read response bytes") } + /// Get a stream of bytes for streaming processing + /// This is the efficient path - data is processed as it arrives + pub fn bytes_stream(self) -> impl Stream> { + self.inner.bytes_stream() + } + /// Get the underlying response for streaming pub fn into_response(self) -> reqwest::Response { self.inner