diff --git a/Cargo.toml b/Cargo.toml index 6fd8eee..c737374 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ categories = ["finance", "command-line-utilities", "encoding"] [workspace] resolver = "2" -members = ["crates/*", "bin"] +members = ["crates/*", "bin", "benches"] default-members = ["bin"] [workspace.lints.rust] @@ -92,3 +92,8 @@ derive_more = { version = "1.0", default-features = false, features = ["display" # Testing approx = "0.5" + +# Benchmarking +criterion = { version = "0.5", features = ["async_tokio", "html_reports"] } +tempfile = "3.14" +which = "7.0" diff --git a/Justfile b/Justfile index abe17c3..14c6534 100644 --- a/Justfile +++ b/Justfile @@ -138,3 +138,21 @@ info instrument: fix: cargo +nightly fmt --all cargo clippy --workspace --all-targets --fix --allow-dirty --allow-staged + +# ============================================================ +# Benchmarks +# ============================================================ + +# Run criterion benchmarks +bench: + cargo bench --package paracas-bench + +# Run benchmark and output markdown table for README +bench-table: + cargo build --release + cargo run --package paracas-bench --bin benchmark_table --release + +# Quick benchmark (1-day only, fewer iterations) +bench-quick: + cargo build --release + cargo run --package paracas-bench --bin benchmark_table --release -- --quick diff --git a/README.md b/README.md index f9b921b..143ebae 100644 --- a/README.md +++ b/README.md @@ -76,6 +76,33 @@ paracas info eurusd | 4 hours | `-t h4` | 4-hour OHLCV bars | | 1 day | `-t d1` | Daily OHLCV bars | +## Performance + +Benchmark comparing paracas against [dukascopy-node](https://www.dukascopy-node.app/) for downloading EUR/USD tick data: + +| Data Range | paracas | dukascopy-node | Speedup | +|------------|---------|----------------|---------| +| 1 day | 5.24s | 8.26s | **1.6x** | +| 3 days | 18.97s | 24.27s | **1.3x** | + +*Benchmarks run on macOS (Apple Silicon). Results may vary based on network conditions.* + +### Running Benchmarks + +```bash +# Run benchmark and output markdown table +just bench-table + +# Run criterion benchmarks +just bench +``` + +To compare against dukascopy-node, install it first: + +```bash +npm install -g dukascopy-node +``` + ## License MIT License - see [LICENSE](LICENSE) for details. diff --git a/benches/Cargo.toml b/benches/Cargo.toml new file mode 100644 index 0000000..69e9124 --- /dev/null +++ b/benches/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "paracas-bench" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +publish = false + +[[bin]] +name = "benchmark_table" +path = "src/bin/benchmark_table.rs" + +[[bench]] +name = "download_benchmark" +harness = false + +[dependencies] +chrono = { workspace = true } +tempfile = { workspace = true } +which = { workspace = true } + +[dev-dependencies] +criterion = { workspace = true } diff --git a/benches/benches/download_benchmark.rs b/benches/benches/download_benchmark.rs new file mode 100644 index 0000000..7e48a66 --- /dev/null +++ b/benches/benches/download_benchmark.rs @@ -0,0 +1,115 @@ +//! Download benchmarks comparing paracas vs dukascopy-node. +//! +//! Run with: `cargo bench --package paracas-bench` +//! +//! Prerequisites: +//! - Build paracas: `cargo build --release` +//! - Install dukascopy-node: `npm install -g dukascopy-node` (optional) + +use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main}; +use paracas_bench::{ + BenchmarkConfig, check_dukascopy_node, find_paracas_binary, run_dukascopy_node, run_paracas, +}; +use std::time::Duration; +use tempfile::TempDir; + +/// Benchmark configurations for different data sizes. +fn benchmark_configs() -> Vec<(&'static str, BenchmarkConfig)> { + vec![ + ( + "1-day", + BenchmarkConfig { + instrument: "eurusd".to_string(), + start_date: "2024-01-02".to_string(), + end_date: "2024-01-02".to_string(), + format: "csv".to_string(), + }, + ), + ( + "3-days", + BenchmarkConfig { + instrument: "eurusd".to_string(), + start_date: "2024-01-02".to_string(), + end_date: "2024-01-04".to_string(), + format: "csv".to_string(), + }, + ), + ] +} + +fn download_benchmark(c: &mut Criterion) { + let paracas_bin = find_paracas_binary() + .expect("paracas binary not found. Run `cargo build --release` first."); + let has_dukascopy = check_dukascopy_node(); + + if !has_dukascopy { + eprintln!( + "Warning: dukascopy-node not found. Install with `npm install -g dukascopy-node`" + ); + eprintln!("Benchmarking paracas only."); + } + + let mut group = c.benchmark_group("download"); + group.sample_size(10); + group.measurement_time(Duration::from_secs(60)); + + for (name, config) in benchmark_configs() { + // Estimate throughput based on typical EURUSD tick count (~100k ticks/day) + let days = estimate_days(&config); + let estimated_ticks = days * 100_000; + group.throughput(Throughput::Elements(estimated_ticks)); + + // Benchmark paracas + group.bench_with_input(BenchmarkId::new("paracas", name), &config, |b, config| { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + for _ in 0..iters { + let temp_dir = TempDir::new().unwrap(); + let output = temp_dir.path().join("output.csv"); + let result = run_paracas(config, output.to_str().unwrap(), ¶cas_bin); + if result.success { + total += result.duration; + } + } + total + }); + }); + + // Benchmark dukascopy-node (if available) + if has_dukascopy { + group.bench_with_input( + BenchmarkId::new("dukascopy-node", name), + &config, + |b, config| { + b.iter_custom(|iters| { + let mut total = Duration::ZERO; + for _ in 0..iters { + let temp_dir = TempDir::new().unwrap(); + let result = run_dukascopy_node( + config, + temp_dir.path().to_str().unwrap(), + "npx", + ); + if result.success { + total += result.duration; + } + } + total + }); + }, + ); + } + } + + group.finish(); +} + +fn estimate_days(config: &BenchmarkConfig) -> u64 { + use chrono::NaiveDate; + let start = NaiveDate::parse_from_str(&config.start_date, "%Y-%m-%d").unwrap(); + let end = NaiveDate::parse_from_str(&config.end_date, "%Y-%m-%d").unwrap(); + (end - start).num_days().max(1) as u64 +} + +criterion_group!(benches, download_benchmark); +criterion_main!(benches); diff --git a/benches/src/bin/benchmark_table.rs b/benches/src/bin/benchmark_table.rs new file mode 100644 index 0000000..34f9c3f --- /dev/null +++ b/benches/src/bin/benchmark_table.rs @@ -0,0 +1,264 @@ +//! Benchmark runner that outputs a markdown table for the README. +//! +//! Run with: `cargo run --package paracas-bench --bin benchmark_table --release` +//! +//! Prerequisites: +//! - Build paracas: `cargo build --release` +//! - Install dukascopy-node: `npm install -g dukascopy-node` (optional) + +use paracas_bench::{ + BenchmarkConfig, BenchmarkResult, check_dukascopy_node, find_paracas_binary, format_bytes, + format_duration, run_dukascopy_node, run_paracas, +}; +use std::io::Write; + +/// Number of iterations per benchmark for statistical significance. +const ITERATIONS: usize = 3; + +fn main() { + println!("Paracas vs dukascopy-node Benchmark"); + println!("====================================\n"); + + let paracas_bin = match find_paracas_binary() { + Some(bin) => { + println!("Found paracas binary: {}", bin); + bin + } + None => { + eprintln!("Error: paracas binary not found."); + eprintln!("Run `cargo build --release` first."); + std::process::exit(1); + } + }; + + let has_dukascopy = check_dukascopy_node(); + if has_dukascopy { + println!("Found dukascopy-node"); + } else { + println!("Warning: dukascopy-node not found"); + println!("Install with: npm install -g dukascopy-node"); + println!("Benchmarking paracas only.\n"); + } + + println!("\nRunning benchmarks ({} iterations each)...\n", ITERATIONS); + + let configs = vec![ + ( + "1 day", + BenchmarkConfig { + instrument: "eurusd".to_string(), + start_date: "2024-01-02".to_string(), + end_date: "2024-01-02".to_string(), + format: "csv".to_string(), + }, + ), + ( + "3 days", + BenchmarkConfig { + instrument: "eurusd".to_string(), + start_date: "2024-01-02".to_string(), + end_date: "2024-01-04".to_string(), + format: "csv".to_string(), + }, + ), + ]; + + let mut results: Vec<(String, Vec, Vec)> = Vec::new(); + + for (name, config) in &configs { + print!("Benchmarking {} data... ", name); + std::io::stdout().flush().unwrap(); + + // Run paracas multiple times + let mut paracas_results = Vec::new(); + for i in 0..ITERATIONS { + let temp_dir = tempfile::TempDir::new().unwrap(); + let output = temp_dir.path().join("output.csv"); + let result = run_paracas(config, output.to_str().unwrap(), ¶cas_bin); + paracas_results.push(result); + print!("P{} ", i + 1); + std::io::stdout().flush().unwrap(); + } + + // Run dukascopy-node multiple times (if available) + let mut dukascopy_results = Vec::new(); + if has_dukascopy { + for i in 0..ITERATIONS { + let temp_dir = tempfile::TempDir::new().unwrap(); + let result = run_dukascopy_node(config, temp_dir.path().to_str().unwrap(), "npx"); + dukascopy_results.push(result); + print!("D{} ", i + 1); + std::io::stdout().flush().unwrap(); + } + } + + results.push((name.to_string(), paracas_results, dukascopy_results)); + println!("done"); + } + + println!("\n## Results\n"); + + // Print markdown table + if has_dukascopy { + println!("| Data Range | paracas | dukascopy-node | Speedup |"); + println!("|------------|---------|----------------|---------|"); + } else { + println!("| Data Range | paracas | Data Points | Throughput |"); + println!("|------------|---------|-------------|------------|"); + } + + for (name, paracas_results, dukascopy_results) in &results { + let paracas_avg = average_results(paracas_results); + + if has_dukascopy && !dukascopy_results.is_empty() { + let dukascopy_avg = average_results(dukascopy_results); + let speedup = if paracas_avg.duration.as_secs_f64() > 0.0 { + dukascopy_avg.duration.as_secs_f64() / paracas_avg.duration.as_secs_f64() + } else { + 0.0 + }; + + println!( + "| {} | {} | {} | **{:.1}x** |", + name, + format_duration(paracas_avg.duration), + format_duration(dukascopy_avg.duration), + speedup + ); + } else { + let data_points = paracas_avg + .data_points + .map(|dp| format!("{:.0}k", dp as f64 / 1000.0)) + .unwrap_or_else(|| "N/A".to_string()); + let throughput = format!("{:.1} MB/s", paracas_avg.throughput_mbps()); + + println!( + "| {} | {} | {} | {} |", + name, + format_duration(paracas_avg.duration), + data_points, + throughput + ); + } + } + + println!("\n### Details\n"); + + for (name, paracas_results, dukascopy_results) in &results { + println!("**{}:**", name); + + // Paracas details + if let Some(result) = paracas_results.first() { + if result.success { + println!( + "- paracas: {} ({} ticks, {})", + format_duration(average_results(paracas_results).duration), + result.data_points.unwrap_or(0), + format_bytes(result.output_size) + ); + } else { + println!("- paracas: FAILED - {:?}", result.error); + } + } + + // dukascopy-node details + if has_dukascopy && let Some(result) = dukascopy_results.first() { + if result.success { + println!( + "- dukascopy-node: {} ({} ticks, {})", + format_duration(average_results(dukascopy_results).duration), + result.data_points.unwrap_or(0), + format_bytes(result.output_size) + ); + } else { + println!("- dukascopy-node: FAILED - {:?}", result.error); + } + } + println!(); + } + + // Print environment info + println!("### Environment\n"); + println!("- OS: {}", std::env::consts::OS); + println!("- Arch: {}", std::env::consts::ARCH); + println!( + "- paracas version: {}", + get_paracas_version(¶cas_bin).unwrap_or_else(|| "unknown".to_string()) + ); + if has_dukascopy { + println!( + "- dukascopy-node version: {}", + get_dukascopy_version().unwrap_or_else(|| "unknown".to_string()) + ); + } +} + +fn average_results(results: &[BenchmarkResult]) -> BenchmarkResult { + let successful: Vec<_> = results.iter().filter(|r| r.success).collect(); + + if successful.is_empty() { + return results.first().cloned().unwrap_or(BenchmarkResult { + tool: "unknown".to_string(), + duration: std::time::Duration::ZERO, + output_size: 0, + data_points: None, + success: false, + error: Some("No successful runs".to_string()), + }); + } + + let avg_duration = std::time::Duration::from_secs_f64( + successful + .iter() + .map(|r| r.duration.as_secs_f64()) + .sum::() + / successful.len() as f64, + ); + + let avg_size = successful.iter().map(|r| r.output_size).sum::() / successful.len() as u64; + + let avg_points = if successful.iter().all(|r| r.data_points.is_some()) { + Some(successful.iter().filter_map(|r| r.data_points).sum::() / successful.len() as u64) + } else { + None + }; + + BenchmarkResult { + tool: successful + .first() + .map(|r| r.tool.clone()) + .unwrap_or_default(), + duration: avg_duration, + output_size: avg_size, + data_points: avg_points, + success: true, + error: None, + } +} + +fn get_paracas_version(bin: &str) -> Option { + std::process::Command::new(bin) + .arg("--version") + .output() + .ok() + .and_then(|o| { + String::from_utf8(o.stdout) + .ok() + .map(|s| s.trim().to_string()) + }) +} + +fn get_dukascopy_version() -> Option { + // dukascopy-node doesn't have a simple --version flag, check npm package version + std::process::Command::new("npm") + .args(["list", "-g", "dukascopy-node", "--depth=0"]) + .output() + .ok() + .and_then(|o| { + String::from_utf8(o.stdout).ok().and_then(|s| { + s.lines() + .find(|line| line.contains("dukascopy-node")) + .map(|line| line.trim().to_string()) + }) + }) +} diff --git a/benches/src/lib.rs b/benches/src/lib.rs new file mode 100644 index 0000000..c12bdec --- /dev/null +++ b/benches/src/lib.rs @@ -0,0 +1,302 @@ +//! Benchmark utilities for paracas. + +use std::process::Command; +use std::time::{Duration, Instant}; + +/// Result of a single benchmark run. +#[derive(Debug, Clone)] +pub struct BenchmarkResult { + /// Name of the tool being benchmarked. + pub tool: String, + /// Duration of the download. + pub duration: Duration, + /// Size of the output file in bytes. + pub output_size: u64, + /// Number of data points (ticks/rows). + pub data_points: Option, + /// Whether the run was successful. + pub success: bool, + /// Error message if failed. + pub error: Option, +} + +impl BenchmarkResult { + /// Calculate throughput in MB/s. + pub fn throughput_mbps(&self) -> f64 { + let bytes = self.output_size as f64; + let secs = self.duration.as_secs_f64(); + if secs > 0.0 { + (bytes / 1_000_000.0) / secs + } else { + 0.0 + } + } + + /// Calculate data points per second. + pub fn data_points_per_sec(&self) -> Option { + self.data_points.map(|dp| { + let secs = self.duration.as_secs_f64(); + if secs > 0.0 { dp as f64 / secs } else { 0.0 } + }) + } +} + +/// Configuration for a benchmark run. +#[derive(Debug, Clone)] +pub struct BenchmarkConfig { + /// Instrument to download (e.g., "eurusd"). + pub instrument: String, + /// Start date in YYYY-MM-DD format. + pub start_date: String, + /// End date in YYYY-MM-DD format. + pub end_date: String, + /// Output format (csv, json). + pub format: String, +} + +impl Default for BenchmarkConfig { + fn default() -> Self { + Self { + instrument: "eurusd".to_string(), + // Default to 1 day of data for quick benchmarks + start_date: "2024-01-02".to_string(), + end_date: "2024-01-02".to_string(), + format: "csv".to_string(), + } + } +} + +/// Run paracas benchmark. +pub fn run_paracas( + config: &BenchmarkConfig, + output_path: &str, + paracas_bin: &str, +) -> BenchmarkResult { + let start = Instant::now(); + + let result = Command::new(paracas_bin) + .args([ + "download", + &config.instrument, + "-s", + &config.start_date, + "-e", + &config.end_date, + "-o", + output_path, + "-f", + &config.format, + "-q", // quiet mode + ]) + .output(); + + let duration = start.elapsed(); + + match result { + Ok(output) => { + if output.status.success() { + let output_size = std::fs::metadata(output_path).map(|m| m.len()).unwrap_or(0); + let data_points = count_csv_rows(output_path); + + BenchmarkResult { + tool: "paracas".to_string(), + duration, + output_size, + data_points, + success: true, + error: None, + } + } else { + BenchmarkResult { + tool: "paracas".to_string(), + duration, + output_size: 0, + data_points: None, + success: false, + error: Some(String::from_utf8_lossy(&output.stderr).to_string()), + } + } + } + Err(e) => BenchmarkResult { + tool: "paracas".to_string(), + duration, + output_size: 0, + data_points: None, + success: false, + error: Some(e.to_string()), + }, + } +} + +/// Run dukascopy-node benchmark. +pub fn run_dukascopy_node( + config: &BenchmarkConfig, + output_path: &str, + npx_bin: &str, +) -> BenchmarkResult { + let start = Instant::now(); + + // dukascopy-node uses lowercase instrument names + let instrument = config.instrument.to_lowercase(); + + // dukascopy-node's -to date is EXCLUSIVE, so we need to add 1 day + // to get the same data range as paracas + let end_date_exclusive = increment_date(&config.end_date); + + // dukascopy-node CLI: npx dukascopy-node -i eurusd -from 2024-01-02 -to 2024-01-03 -t tick -f csv + let result = Command::new(npx_bin) + .args([ + "dukascopy-node", + "-i", + &instrument, + "-from", + &config.start_date, + "-to", + &end_date_exclusive, + "-t", + "tick", + "-f", + &config.format, + "-dir", + output_path, + "-s", // silent mode + ]) + .output(); + + let duration = start.elapsed(); + + match result { + Ok(output) => { + if output.status.success() { + // dukascopy-node outputs to a directory, find the file + let output_file = find_output_file(output_path, &config.format); + let (output_size, data_points) = match &output_file { + Some(path) => { + let size = std::fs::metadata(path).map(|m| m.len()).unwrap_or(0); + let points = count_csv_rows(path); + (size, points) + } + None => (0, None), + }; + + BenchmarkResult { + tool: "dukascopy-node".to_string(), + duration, + output_size, + data_points, + success: true, + error: None, + } + } else { + BenchmarkResult { + tool: "dukascopy-node".to_string(), + duration, + output_size: 0, + data_points: None, + success: false, + error: Some(String::from_utf8_lossy(&output.stderr).to_string()), + } + } + } + Err(e) => BenchmarkResult { + tool: "dukascopy-node".to_string(), + duration, + output_size: 0, + data_points: None, + success: false, + error: Some(e.to_string()), + }, + } +} + +/// Count rows in a CSV file (excluding header). +fn count_csv_rows(path: &str) -> Option { + std::fs::read_to_string(path) + .ok() + .map(|content| content.lines().count().saturating_sub(1) as u64) +} + +/// Find output file in directory. +fn find_output_file(dir: &str, format: &str) -> Option { + let ext = match format { + "csv" => "csv", + "json" => "json", + _ => "csv", + }; + + std::fs::read_dir(dir) + .ok()? + .filter_map(|e| e.ok()) + .find(|e| { + e.path() + .extension() + .and_then(|e| e.to_str()) + .map(|e| e == ext) + .unwrap_or(false) + }) + .map(|e| e.path().to_string_lossy().to_string()) +} + +/// Increment a date string by one day (for dukascopy-node exclusive end date). +fn increment_date(date_str: &str) -> String { + use chrono::NaiveDate; + let date = NaiveDate::parse_from_str(date_str, "%Y-%m-%d").expect("valid date"); + let next_day = date + chrono::Duration::days(1); + next_day.format("%Y-%m-%d").to_string() +} + +/// Format duration for display. +pub fn format_duration(d: Duration) -> String { + let secs = d.as_secs_f64(); + if secs < 1.0 { + format!("{:.0}ms", secs * 1000.0) + } else if secs < 60.0 { + format!("{:.2}s", secs) + } else { + let mins = secs / 60.0; + format!("{:.1}m", mins) + } +} + +/// Format bytes for display. +pub fn format_bytes(bytes: u64) -> String { + if bytes < 1024 { + format!("{} B", bytes) + } else if bytes < 1024 * 1024 { + format!("{:.1} KB", bytes as f64 / 1024.0) + } else if bytes < 1024 * 1024 * 1024 { + format!("{:.1} MB", bytes as f64 / (1024.0 * 1024.0)) + } else { + format!("{:.2} GB", bytes as f64 / (1024.0 * 1024.0 * 1024.0)) + } +} + +/// Check if dukascopy-node is available. +pub fn check_dukascopy_node() -> bool { + // dukascopy-node --version requires an instrument, so just check --help + Command::new("npx") + .args(["dukascopy-node", "--help"]) + .output() + .map(|o| o.status.success()) + .unwrap_or(false) +} + +/// Check if paracas binary exists. +pub fn find_paracas_binary() -> Option { + // Try release binary first + let release_path = "target/release/paracas"; + if std::path::Path::new(release_path).exists() { + return Some(release_path.to_string()); + } + + // Try debug binary + let debug_path = "target/debug/paracas"; + if std::path::Path::new(debug_path).exists() { + return Some(debug_path.to_string()); + } + + // Try system PATH + which::which("paracas") + .ok() + .map(|p| p.to_string_lossy().to_string()) +} diff --git a/crates/paracas-fetch/src/client.rs b/crates/paracas-fetch/src/client.rs index 1238682..4a8f9a3 100644 --- a/crates/paracas-fetch/src/client.rs +++ b/crates/paracas-fetch/src/client.rs @@ -69,8 +69,18 @@ impl DownloadClient { /// Returns an error if the HTTP client cannot be created. pub fn new(config: ClientConfig) -> Result { let client = Client::builder() + // Connection pooling - maintain up to concurrency idle connections per host .pool_max_idle_per_host(config.concurrency) + // Keep connections alive for reuse (Dukascopy supports persistent connections) + .pool_idle_timeout(Duration::from_secs(90)) + // Disable Nagle's algorithm for lower latency + .tcp_nodelay(true) + // Keep TCP connections alive + .tcp_keepalive(Duration::from_secs(60)) + // Request timeout .timeout(config.timeout) + // Connection timeout (separate from request timeout) + .connect_timeout(Duration::from_secs(10)) .user_agent(&config.user_agent) .gzip(true) .build()?; diff --git a/crates/paracas-fetch/src/stream.rs b/crates/paracas-fetch/src/stream.rs index 3199791..48263af 100644 --- a/crates/paracas-fetch/src/stream.rs +++ b/crates/paracas-fetch/src/stream.rs @@ -84,22 +84,31 @@ pub fn tick_stream<'a>( .map(move |hour| { let url = tick_url(&instrument_id, hour); let client = client.clone(); - async move { (hour, client.download(&url).await) } + async move { + let result = client.download(&url).await; + // Process immediately after download (decompression is offloaded to spawn_blocking) + process_download_result(hour, result, decimal_factor).await + } }) .buffer_unordered(concurrency) - .map(move |(hour, result)| process_download_result(hour, result, decimal_factor)) } /// Processes a download result into a tick batch. -fn process_download_result( +/// +/// Decompression is offloaded to a blocking thread pool to avoid blocking +/// the async executor. +async fn process_download_result( hour: DateTime, result: Result, crate::DownloadError>, decimal_factor: f64, ) -> Result { match result { Ok(Some(compressed)) => { - let decompressed = - decompress_bi5(&compressed).map_err(|e| ParacasError::Decompress(e.to_string()))?; + // Offload CPU-intensive LZMA decompression to blocking thread pool + let decompressed = tokio::task::spawn_blocking(move || decompress_bi5(&compressed)) + .await + .map_err(|e| ParacasError::Decompress(format!("spawn_blocking failed: {e}")))? + .map_err(|e| ParacasError::Decompress(e.to_string()))?; let ticks: Vec = parse_ticks(&decompressed) .map_err(|e| ParacasError::Parse(e.to_string()))? @@ -144,36 +153,42 @@ pub fn tick_stream_resilient<'a>( .map(move |hour| { let url = tick_url(&instrument_id, hour); let client = client.clone(); - async move { (hour, client.download(&url).await) } + async move { + let result = client.download(&url).await; + // Process immediately after download (decompression is offloaded to spawn_blocking) + process_download_result_resilient(hour, result, decimal_factor).await + } }) .buffer_unordered(concurrency) - .map(move |(hour, result)| process_download_result_resilient(hour, result, decimal_factor)) } /// Processes a download result into a tick batch, skipping errors. -#[allow(clippy::option_if_let_else)] // Nested matches are more readable here -fn process_download_result_resilient( +/// +/// Decompression is offloaded to a blocking thread pool to avoid blocking +/// the async executor. +async fn process_download_result_resilient( hour: DateTime, result: Result, crate::DownloadError>, decimal_factor: f64, ) -> TickBatch { match result { Ok(Some(compressed)) => { - match decompress_bi5(&compressed) { - Ok(decompressed) => match parse_ticks(&decompressed) { - Ok(raw_ticks) => { + // Offload CPU-intensive LZMA decompression to blocking thread pool + let decompress_result = + tokio::task::spawn_blocking(move || decompress_bi5(&compressed)).await; + + match decompress_result { + Ok(Ok(decompressed)) => parse_ticks(&decompressed).map_or_else( + |_| TickBatch::skipped_error(hour), + |raw_ticks| { let ticks: Vec = raw_ticks .map(|raw| raw.normalize(hour, decimal_factor)) .collect(); TickBatch::new(hour, ticks) - } - Err(_) => { - // Parse error - return empty batch with error flag - TickBatch::skipped_error(hour) - } - }, - Err(_) => { - // Decompression error - return empty batch with error flag + }, + ), + _ => { + // Decompression error or spawn_blocking failed - return empty batch with error flag TickBatch::skipped_error(hour) } }