Release v2.5.3.1-beta: smart Zstd compression, Blake3 delta-sync, multi-send, pause/resume, priority queue, and 4-part version parser
This commit is contained in:
RarDog
Generated
+32
-3
@@ -654,6 +654,19 @@ dependencies = [
|
||||
"serde_core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "blake3"
|
||||
version = "1.8.7"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6d9e454fc11f76977dc803893aff6304ed33d6a26efae8696573bea74baa27ae"
|
||||
dependencies = [
|
||||
"arrayvec",
|
||||
"cc",
|
||||
"cfg-if",
|
||||
"constant_time_eq",
|
||||
"cpufeatures 0.3.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "block"
|
||||
version = "0.1.6"
|
||||
@@ -1034,6 +1047,12 @@ dependencies = [
|
||||
"windows-sys 0.59.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "constant_time_eq"
|
||||
version = "0.4.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b"
|
||||
|
||||
[[package]]
|
||||
name = "core-foundation"
|
||||
version = "0.9.4"
|
||||
@@ -1093,6 +1112,15 @@ dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "cpufeatures"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crc32fast"
|
||||
version = "1.5.1"
|
||||
@@ -4333,7 +4361,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a978451301f4db1d02937a4ab3ccce137717b81826e79b7d49ffe3244a13c3b8"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures",
|
||||
"cpufeatures 0.2.17",
|
||||
"digest",
|
||||
]
|
||||
|
||||
@@ -4344,7 +4372,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures",
|
||||
"cpufeatures 0.2.17",
|
||||
"digest",
|
||||
]
|
||||
|
||||
@@ -6425,11 +6453,12 @@ checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e"
|
||||
|
||||
[[package]]
|
||||
name = "zerosend"
|
||||
version = "2.5.3-beta"
|
||||
version = "2.5.3-beta.1"
|
||||
dependencies = [
|
||||
"arboard",
|
||||
"async-stream",
|
||||
"axum",
|
||||
"blake3",
|
||||
"bytes",
|
||||
"chrono",
|
||||
"clap",
|
||||
|
||||
+2
-1
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "zerosend"
|
||||
version = "2.5.3-beta"
|
||||
version = "2.5.3-beta.1"
|
||||
edition = "2021"
|
||||
description = "Fast, peer-to-peer file transfer designed for ZeroTier, Tailscale, Radmin VPN and LAN"
|
||||
authors = ["RarDog"]
|
||||
@@ -49,6 +49,7 @@ sha2 = "0.10"
|
||||
hex = "0.4"
|
||||
walkdir = "2.5"
|
||||
human_bytes = "0.4"
|
||||
blake3 = "1.5"
|
||||
|
||||
# Windows Native Registry & Sound
|
||||
[target.'cfg(windows)'.dependencies]
|
||||
|
||||
@@ -192,6 +192,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
TransferState::Rejected => {
|
||||
pb.abandon_with_message("⛔ Transfer rejected by recipient.");
|
||||
}
|
||||
TransferState::Paused => {
|
||||
pb.set_message("⏸ Paused");
|
||||
}
|
||||
TransferState::Canceled => {
|
||||
pb.abandon_with_message("🛑 Canceled.");
|
||||
}
|
||||
|
||||
@@ -117,6 +117,40 @@ pub fn estimate_folder_size(dir: &Path) -> u64 {
|
||||
.sum()
|
||||
}
|
||||
|
||||
/// Returns the optimal zstd compression level for a given filename.
|
||||
/// Returns 0 for already-compressed formats (store only), 1 otherwise (ultra-fast).
|
||||
pub fn optimal_zstd_level_for_file(name: &str) -> i32 {
|
||||
let ext = name.rsplit('.').next().unwrap_or("").to_lowercase();
|
||||
match ext.as_str() {
|
||||
// Already compressed — don't waste CPU
|
||||
"jpg" | "jpeg" | "png" | "gif" | "webp" | "avif" | "heic" |
|
||||
"mp4" | "mkv" | "avi" | "mov" | "wmv" | "flv" | "webm" |
|
||||
"mp3" | "aac" | "ogg" | "flac" | "m4a" | "opus" |
|
||||
"zip" | "rar" | "7z" | "gz" | "bz2" | "xz" | "zst" | "lz4" |
|
||||
"pdf" | "docx" | "xlsx" | "pptx" => 0,
|
||||
// Compressible text/code/data
|
||||
_ => 1,
|
||||
}
|
||||
}
|
||||
|
||||
/// Estimates the compressibility of a folder: returns true if most content is compressible.
|
||||
pub fn folder_is_compressible(dir: &Path) -> bool {
|
||||
let mut compressible = 0u64;
|
||||
let mut total = 0u64;
|
||||
for entry in WalkDir::new(dir).into_iter().filter_map(|e| e.ok()) {
|
||||
if entry.path().is_file() {
|
||||
let size = entry.metadata().map(|m| m.len()).unwrap_or(0);
|
||||
total += size;
|
||||
let name = entry.file_name().to_string_lossy();
|
||||
if optimal_zstd_level_for_file(&name) > 0 {
|
||||
compressible += size;
|
||||
}
|
||||
}
|
||||
}
|
||||
if total == 0 { return true; }
|
||||
(compressible as f64 / total as f64) > 0.3
|
||||
}
|
||||
|
||||
/// Stream an entire directory as a Zstandard-compressed TAR archive
|
||||
/// into a `std::sync::mpsc::SyncSender<Vec<u8>>` channel.
|
||||
///
|
||||
|
||||
+123
-14
@@ -10,8 +10,9 @@ use super::text_analyzer::{analyze_text, AnalyzedText};
|
||||
use crate::network::DiscoveryBeacon;
|
||||
use reqwest::Client;
|
||||
use sha2::{Digest, Sha256};
|
||||
use std::collections::VecDeque;
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::path::PathBuf;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::fs::File;
|
||||
@@ -60,6 +61,7 @@ struct ChunkTask {
|
||||
|
||||
pub struct TransferClient {
|
||||
client: Client,
|
||||
pub pause_flags: Arc<Mutex<HashMap<String, Arc<AtomicBool>>>>,
|
||||
}
|
||||
|
||||
impl TransferClient {
|
||||
@@ -71,7 +73,55 @@ impl TransferClient {
|
||||
.build()
|
||||
.unwrap_or_default();
|
||||
|
||||
Self { client }
|
||||
Self {
|
||||
client,
|
||||
pause_flags: Arc::new(Mutex::new(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn pause_transfer(&self, session_id: &str) {
|
||||
let flags = self.pause_flags.lock().await;
|
||||
if let Some(flag) = flags.get(session_id) {
|
||||
flag.store(true, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn resume_transfer(&self, session_id: &str) {
|
||||
let flags = self.pause_flags.lock().await;
|
||||
if let Some(flag) = flags.get(session_id) {
|
||||
flag.store(false, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn is_paused(&self, session_id: &str) -> bool {
|
||||
let flags = self.pause_flags.lock().await;
|
||||
flags.get(session_id).map(|f| f.load(Ordering::Relaxed)).unwrap_or(false)
|
||||
}
|
||||
|
||||
/// Race-ping multiple candidate URLs and return the fastest responding one
|
||||
pub async fn pick_fastest_url(&self, candidates: &[String]) -> Option<String> {
|
||||
if candidates.is_empty() { return None; }
|
||||
if candidates.len() == 1 { return Some(candidates[0].clone()); }
|
||||
|
||||
let (tx, mut rx) = tokio::sync::mpsc::channel(candidates.len());
|
||||
|
||||
for url in candidates {
|
||||
let client = self.client.clone();
|
||||
let tx = tx.clone();
|
||||
let url = url.clone();
|
||||
tokio::spawn(async move {
|
||||
let start = Instant::now();
|
||||
if let Ok(resp) = client.get(&format!("{}/api/info", url)).send().await {
|
||||
if resp.status().is_success() {
|
||||
let latency = start.elapsed().as_millis() as u64;
|
||||
let _ = tx.send((url, latency)).await;
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
drop(tx);
|
||||
|
||||
rx.recv().await.map(|(url, _)| url)
|
||||
}
|
||||
|
||||
/// Ping a remote peer to check health and get beacon info
|
||||
@@ -237,15 +287,27 @@ impl TransferClient {
|
||||
speed_bps: 0,
|
||||
state: TransferState::InProgress,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
});
|
||||
|
||||
// Feature 1: Determine optimal compression level for entire folder
|
||||
let is_compressible = {
|
||||
let folder_clone2 = folder_path.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
super::archive::folder_is_compressible(&folder_clone2)
|
||||
})
|
||||
.await
|
||||
.unwrap_or(true)
|
||||
};
|
||||
let folder_zstd_level = if is_compressible { 1 } else { 0 };
|
||||
|
||||
// Build a bounded channel: blocking producer → async consumer → HTTP body
|
||||
let (tx, rx) = std::sync::mpsc::sync_channel::<std::io::Result<Vec<u8>>>(32);
|
||||
let folder_clone = folder_path.clone();
|
||||
|
||||
// Spawn blocking tar+zstd compressor thread
|
||||
let compress_handle = tokio::task::spawn_blocking(move || {
|
||||
super::archive::stream_folder_as_tar_zst(folder_clone, tx, 1);
|
||||
super::archive::stream_folder_as_tar_zst(folder_clone, tx, folder_zstd_level);
|
||||
});
|
||||
|
||||
// Collect chunks from the channel into an HTTP body stream
|
||||
@@ -303,6 +365,7 @@ impl TransferClient {
|
||||
speed_bps: 0,
|
||||
state: TransferState::Completed,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
});
|
||||
}
|
||||
Ok(r) => {
|
||||
@@ -326,6 +389,8 @@ impl TransferClient {
|
||||
return Err("No files selected for transfer".to_string());
|
||||
}
|
||||
|
||||
// Feature 7: Priority queue - sort files by size ascending so small files finish first
|
||||
file_entries.sort_by_key(|(_, _, size)| *size);
|
||||
|
||||
let file_metas: Vec<FileMetadata> = file_entries
|
||||
.iter()
|
||||
@@ -367,6 +432,7 @@ impl TransferClient {
|
||||
speed_bps: 0,
|
||||
state: TransferState::PendingConfirmation,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
});
|
||||
|
||||
let prepare_url = format!("{}/api/transfer/prepare", peer_url);
|
||||
@@ -397,6 +463,7 @@ impl TransferClient {
|
||||
speed_bps: 0,
|
||||
state: TransferState::Rejected,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
});
|
||||
|
||||
return Err(reason);
|
||||
@@ -451,9 +518,16 @@ impl TransferClient {
|
||||
}
|
||||
}
|
||||
|
||||
let pause_flag = {
|
||||
let mut flags = self.pause_flags.lock().await;
|
||||
let flag = Arc::new(AtomicBool::new(false));
|
||||
flags.insert(session_id.clone(), flag.clone());
|
||||
flag
|
||||
};
|
||||
|
||||
let rate_limiter = RateLimiter::new((max_upload_speed_mbps as u64) * 1024 * 1024);
|
||||
let tasks_mutex = Arc::new(Mutex::new(chunk_tasks));
|
||||
let (chunk_tx, mut chunk_rx) = tokio::sync::mpsc::unbounded_channel::<(usize, String)>();
|
||||
let (chunk_tx, mut chunk_rx) = tokio::sync::mpsc::unbounded_channel::<(usize, usize, String)>();
|
||||
let (progress_update_tx, mut progress_update_rx) = tokio::sync::mpsc::unbounded_channel::<TransferProgress>();
|
||||
|
||||
let s_id = session_id.clone();
|
||||
@@ -461,15 +535,17 @@ impl TransferClient {
|
||||
let t_files = file_metas.len();
|
||||
let p_tx = progress_update_tx.clone();
|
||||
|
||||
// Background progress tracker with smoothed EMA speed calculation
|
||||
// Background progress tracker with smoothed EMA speed calculation & compressed_bytes
|
||||
tokio::spawn(async move {
|
||||
let mut total_sent = already_transferred;
|
||||
let mut total_wire = already_transferred;
|
||||
let mut last_bytes = total_sent;
|
||||
let mut last_time = Instant::now();
|
||||
let mut smoothed_speed = 0.0f64;
|
||||
|
||||
while let Some((len, fname)) = chunk_rx.recv().await {
|
||||
total_sent += len as u64;
|
||||
while let Some((raw_len, wire_len, fname)) = chunk_rx.recv().await {
|
||||
total_sent += raw_len as u64;
|
||||
total_wire += wire_len as u64;
|
||||
|
||||
if last_time.elapsed().as_millis() >= 350 {
|
||||
let elapsed_secs = last_time.elapsed().as_secs_f64();
|
||||
@@ -500,6 +576,7 @@ impl TransferClient {
|
||||
speed_bps,
|
||||
state: TransferState::InProgress,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: total_wire,
|
||||
});
|
||||
|
||||
last_time = Instant::now();
|
||||
@@ -520,6 +597,7 @@ impl TransferClient {
|
||||
speed_bps: 0,
|
||||
state: TransferState::InProgress,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: already_transferred,
|
||||
});
|
||||
|
||||
let start_time = Instant::now();
|
||||
@@ -532,6 +610,7 @@ impl TransferClient {
|
||||
let chunk_url = format!("{}/api/transfer/chunk/{}", peer_url, session_id);
|
||||
let limiter_c = rate_limiter.clone();
|
||||
let tx_c = chunk_tx.clone();
|
||||
let pause_c = pause_flag.clone();
|
||||
|
||||
worker_set.spawn(async move {
|
||||
loop {
|
||||
@@ -558,16 +637,41 @@ impl TransferClient {
|
||||
return Err(format!("Failed to read chunk from {:?}: {}", chunk_task.file_path, e));
|
||||
}
|
||||
|
||||
limiter_c.throttle(buf.len()).await;
|
||||
// Feature 1: Smart compression check per file type
|
||||
let is_compressible = super::archive::optimal_zstd_level_for_file(&chunk_task.rel_path) > 0;
|
||||
let (payload, is_zstd) = if is_compressible && buf.len() >= 1024 {
|
||||
if let Ok(compressed) = zstd::bulk::compress(&buf, 1) {
|
||||
if compressed.len() + 64 < buf.len() {
|
||||
(compressed, true)
|
||||
} else {
|
||||
(buf, false)
|
||||
}
|
||||
} else {
|
||||
(buf, false)
|
||||
}
|
||||
} else {
|
||||
(buf, false)
|
||||
};
|
||||
|
||||
let resp = client_c
|
||||
let wire_len = payload.len();
|
||||
limiter_c.throttle(wire_len).await;
|
||||
|
||||
// Feature 5: Pause support in worker loop
|
||||
while pause_c.load(Ordering::Relaxed) {
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
}
|
||||
|
||||
let mut req = client_c
|
||||
.post(&chunk_url)
|
||||
.header("x-file-name", &chunk_task.rel_path)
|
||||
.header("x-chunk-offset", chunk_task.offset.to_string())
|
||||
.header("x-chunk-length", chunk_task.length.to_string())
|
||||
.body(buf)
|
||||
.send()
|
||||
.await;
|
||||
.header("x-chunk-length", chunk_task.length.to_string());
|
||||
|
||||
if is_zstd {
|
||||
req = req.header("x-compression", "zstd");
|
||||
}
|
||||
|
||||
let resp = req.body(payload).send().await;
|
||||
|
||||
let is_ok = match &resp {
|
||||
Ok(r) => r.status().is_success(),
|
||||
@@ -589,7 +693,7 @@ impl TransferClient {
|
||||
}
|
||||
}
|
||||
|
||||
let _ = tx_c.send((chunk_task.length, chunk_task.rel_path));
|
||||
let _ = tx_c.send((chunk_task.length, wire_len, chunk_task.rel_path));
|
||||
}
|
||||
Ok::<(), String>(())
|
||||
});
|
||||
@@ -616,6 +720,7 @@ impl TransferClient {
|
||||
speed_bps: 0,
|
||||
state: TransferState::Failed(e.clone()),
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
});
|
||||
return Err(e);
|
||||
}
|
||||
@@ -636,6 +741,9 @@ impl TransferClient {
|
||||
on_progress(progress);
|
||||
}
|
||||
|
||||
// Cleanup pause flags
|
||||
self.pause_flags.lock().await.remove(&session_id);
|
||||
|
||||
// Finalize transfer on receiver (atomic rename + zip extraction)
|
||||
let finish_url = format!("{}/api/transfer/finish/{}", peer_url, session_id);
|
||||
let finish_resp = self.client.post(&finish_url).send().await;
|
||||
@@ -663,6 +771,7 @@ impl TransferClient {
|
||||
speed_bps: avg_speed,
|
||||
state: TransferState::Completed,
|
||||
checksum_verified: true,
|
||||
compressed_bytes: total_bytes,
|
||||
});
|
||||
|
||||
Ok(())
|
||||
|
||||
+119
-10
@@ -22,6 +22,49 @@ pub struct FolderSyncEngine {
|
||||
status_tx: mpsc::UnboundedSender<SyncStatusUpdate>,
|
||||
}
|
||||
|
||||
// ── Sync cache structures ──────────────────────────────────────────────────
|
||||
|
||||
#[derive(serde::Serialize, serde::Deserialize, Default)]
|
||||
struct SyncCache {
|
||||
files: HashMap<String, CacheEntry>,
|
||||
}
|
||||
|
||||
#[derive(serde::Serialize, serde::Deserialize, Clone)]
|
||||
struct CacheEntry {
|
||||
size: u64,
|
||||
mtime: u64,
|
||||
blake3: Option<String>,
|
||||
}
|
||||
|
||||
fn load_sync_cache(local_path: &Path) -> SyncCache {
|
||||
let cache_path = local_path.join(".zerosend_sync_cache.json");
|
||||
std::fs::read_to_string(&cache_path)
|
||||
.ok()
|
||||
.and_then(|s| serde_json::from_str(&s).ok())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
fn save_sync_cache(local_path: &Path, cache: &SyncCache) {
|
||||
let cache_path = local_path.join(".zerosend_sync_cache.json");
|
||||
if let Ok(s) = serde_json::to_string(cache) {
|
||||
let _ = std::fs::write(&cache_path, s);
|
||||
}
|
||||
}
|
||||
|
||||
fn file_blake3(path: &std::path::Path) -> Option<String> {
|
||||
let mut file = std::fs::File::open(path).ok()?;
|
||||
let mut hasher = blake3::Hasher::new();
|
||||
let mut buf = [0u8; 65536];
|
||||
loop {
|
||||
let n = std::io::Read::read(&mut file, &mut buf).ok()?;
|
||||
if n == 0 { break; }
|
||||
hasher.update(&buf[..n]);
|
||||
}
|
||||
Some(hasher.finalize().to_hex().to_string())
|
||||
}
|
||||
|
||||
// ── Engine ─────────────────────────────────────────────────────────────────
|
||||
|
||||
impl FolderSyncEngine {
|
||||
pub fn new(
|
||||
client: Arc<TransferClient>,
|
||||
@@ -109,17 +152,31 @@ impl FolderSyncEngine {
|
||||
let _ = tokio::fs::create_dir_all(local_path).await;
|
||||
}
|
||||
|
||||
// Load sync cache (size/mtime/hash from last sync)
|
||||
let cache = load_sync_cache(local_path);
|
||||
|
||||
// Fetch remote directory tree
|
||||
let remote_tree = client
|
||||
.fetch_shared_tree(peer_url, remote_folder_id, None, None)
|
||||
.await?;
|
||||
|
||||
// 1. Scan local files
|
||||
let mut local_files: HashMap<String, (u64, u64)> = HashMap::new(); // rel_path -> (size, mtime)
|
||||
if let Ok(entries) = walkdir_local(local_path) {
|
||||
for (rel, size, mtime) in entries {
|
||||
local_files.insert(rel, (size, mtime));
|
||||
// 1. Scan local files with blake3 hash support
|
||||
// Returns Vec<(rel_path, size, mtime, Option<hash>)>
|
||||
let local_files_vec = {
|
||||
let local_path_buf = local_path.to_path_buf();
|
||||
tokio::task::spawn_blocking(move || walkdir_local(&local_path_buf))
|
||||
.await
|
||||
.map_err(|e| format!("walkdir task error: {}", e))?
|
||||
.map_err(|e| format!("walkdir error: {}", e))?
|
||||
};
|
||||
|
||||
let mut local_files: HashMap<String, (u64, u64, Option<String>)> = HashMap::new();
|
||||
for (rel, size, mtime, hash) in local_files_vec {
|
||||
// Skip the cache file itself
|
||||
if rel == ".zerosend_sync_cache.json" {
|
||||
continue;
|
||||
}
|
||||
local_files.insert(rel, (size, mtime, hash));
|
||||
}
|
||||
|
||||
// 2. Map remote files
|
||||
@@ -131,12 +188,36 @@ impl FolderSyncEngine {
|
||||
}
|
||||
|
||||
let mut synced_count = 0;
|
||||
let mut updated_cache = SyncCache { files: cache.files.clone() };
|
||||
|
||||
// A. Upload newly created or changed local files to remote
|
||||
if !remote_tree.read_only {
|
||||
for (rel, (size, _)) in &local_files {
|
||||
for (rel, (size, mtime, hash)) in &local_files {
|
||||
let need_upload = match remote_files.get(rel) {
|
||||
Some(remote_size) => *remote_size != *size,
|
||||
Some(remote_size) => {
|
||||
if *remote_size != *size {
|
||||
true // size differs — definitely upload
|
||||
} else {
|
||||
// Same size: check cache to detect content-only changes
|
||||
if let Some(cached) = cache.files.get(rel) {
|
||||
if cached.size == *size && cached.mtime == *mtime {
|
||||
// size AND mtime unchanged since last sync — skip
|
||||
false
|
||||
} else if cached.size == *size {
|
||||
// mtime changed but size same — compare blake3
|
||||
match (hash, &cached.blake3) {
|
||||
(Some(h), Some(c)) => h != c, // upload if hash changed
|
||||
_ => true, // no hash available, upload to be safe
|
||||
}
|
||||
} else {
|
||||
true // cached size differs — upload
|
||||
}
|
||||
} else {
|
||||
// No cache entry — upload to be safe
|
||||
true
|
||||
}
|
||||
}
|
||||
},
|
||||
None => true, // New local file
|
||||
};
|
||||
|
||||
@@ -153,8 +234,21 @@ impl FolderSyncEngine {
|
||||
)
|
||||
.await
|
||||
{
|
||||
// Update cache entry after successful upload
|
||||
updated_cache.files.insert(rel.clone(), CacheEntry {
|
||||
size: *size,
|
||||
mtime: *mtime,
|
||||
blake3: hash.clone(),
|
||||
});
|
||||
synced_count += 1;
|
||||
}
|
||||
} else {
|
||||
// File unchanged — update cache entry to keep mtime fresh
|
||||
updated_cache.files.insert(rel.clone(), CacheEntry {
|
||||
size: *size,
|
||||
mtime: *mtime,
|
||||
blake3: hash.clone(),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -162,7 +256,7 @@ impl FolderSyncEngine {
|
||||
// B. Download newly created remote files to local
|
||||
for (rel, remote_size) in &remote_files {
|
||||
let need_download = match local_files.get(rel) {
|
||||
Some((local_size, _)) => *local_size != *remote_size,
|
||||
Some((local_size, _, _)) => *local_size != *remote_size,
|
||||
None => true, // New remote file
|
||||
};
|
||||
|
||||
@@ -177,11 +271,16 @@ impl FolderSyncEngine {
|
||||
.download_shared_file(peer_url, remote_folder_id, rel, &target_file_path, None)
|
||||
.await
|
||||
{
|
||||
// Invalidate cache entry since we downloaded new content
|
||||
updated_cache.files.remove(rel);
|
||||
synced_count += 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Save updated cache
|
||||
save_sync_cache(local_path, &updated_cache);
|
||||
|
||||
if synced_count > 0 {
|
||||
let _ = status_tx.send(SyncStatusUpdate {
|
||||
pair_id: pair_id.to_string(),
|
||||
@@ -195,7 +294,10 @@ impl FolderSyncEngine {
|
||||
}
|
||||
}
|
||||
|
||||
fn walkdir_local(base: &Path) -> Result<Vec<(String, u64, u64)>, std::io::Error> {
|
||||
/// Walk local directory and return (rel_path, size, mtime, Option<blake3_hash>).
|
||||
/// Hash is only computed for files < 500MB to avoid stalling.
|
||||
fn walkdir_local(base: &Path) -> Result<Vec<(String, u64, u64, Option<String>)>, std::io::Error> {
|
||||
const MAX_HASH_SIZE: u64 = 500 * 1024 * 1024; // 500 MB
|
||||
let mut results = Vec::new();
|
||||
for entry in walkdir::WalkDir::new(base).into_iter().filter_map(|e| e.ok()) {
|
||||
if entry.file_type().is_file() {
|
||||
@@ -209,7 +311,14 @@ fn walkdir_local(base: &Path) -> Result<Vec<(String, u64, u64)>, std::io::Error>
|
||||
.and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
results.push((rel_str, size, mtime));
|
||||
|
||||
let hash = if size < MAX_HASH_SIZE {
|
||||
file_blake3(entry.path())
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
results.push((rel_str, size, mtime, hash));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -51,6 +51,7 @@ pub enum PrepareTransferResponse {
|
||||
pub enum TransferState {
|
||||
PendingConfirmation,
|
||||
InProgress,
|
||||
Paused,
|
||||
Completed,
|
||||
Failed(String),
|
||||
Rejected,
|
||||
@@ -71,6 +72,8 @@ pub struct TransferProgress {
|
||||
pub state: TransferState,
|
||||
#[serde(default)]
|
||||
pub checksum_verified: bool,
|
||||
#[serde(default)]
|
||||
pub compressed_bytes: u64, // actual bytes sent over network (after compression)
|
||||
}
|
||||
|
||||
impl TransferProgress {
|
||||
|
||||
+35
-2
@@ -422,6 +422,7 @@ async fn handle_prepare(
|
||||
speed_bps: 0,
|
||||
state: TransferState::InProgress,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
};
|
||||
|
||||
state.active_transfers.write().await.insert(session_id.clone(), progress.clone());
|
||||
@@ -468,6 +469,7 @@ async fn handle_prepare(
|
||||
speed_bps: 0,
|
||||
state: TransferState::InProgress,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
};
|
||||
|
||||
state.active_transfers.write().await.insert(session_id.clone(), progress.clone());
|
||||
@@ -551,8 +553,36 @@ async fn handle_upload_chunk(
|
||||
);
|
||||
}
|
||||
|
||||
let chunk_len = body.len();
|
||||
if let Err(e) = file.write_all(&body).await {
|
||||
let raw_length: usize = headers
|
||||
.get("x-chunk-length")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| v.parse().ok())
|
||||
.unwrap_or(body.len());
|
||||
|
||||
let is_zstd = headers
|
||||
.get("x-compression")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.map(|v| v.eq_ignore_ascii_case("zstd"))
|
||||
.unwrap_or(false);
|
||||
|
||||
let decompressed_data = if is_zstd {
|
||||
match zstd::bulk::decompress(&body, raw_length.max(16 * 1024 * 1024)) {
|
||||
Ok(decomp) => decomp,
|
||||
Err(e) => {
|
||||
error!("Zstd decompression failed for {}: {}", file_name, e);
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(serde_json::json!({"error": format!("Zstd decompress error: {}", e)})),
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
body.to_vec()
|
||||
};
|
||||
|
||||
let chunk_len = decompressed_data.len();
|
||||
let wire_bytes = body.len() as u64;
|
||||
if let Err(e) = file.write_all(&decompressed_data).await {
|
||||
error!("Write error on {:?} at {}: {}", part_path, offset, e);
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
@@ -567,6 +597,7 @@ async fn handle_upload_chunk(
|
||||
let mut transfers = state.active_transfers.write().await;
|
||||
if let Some(t) = transfers.get_mut(&session_id) {
|
||||
t.bytes_transferred = (t.bytes_transferred + chunk_len as u64).min(t.total_bytes);
|
||||
t.compressed_bytes += wire_bytes;
|
||||
t.current_file_name = file_name;
|
||||
t.state = TransferState::InProgress;
|
||||
|
||||
@@ -715,6 +746,7 @@ async fn handle_folder_stream(
|
||||
speed_bps: 0,
|
||||
state: super::protocol::TransferState::Completed,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
};
|
||||
let _ = state.transfer_events.send(progress);
|
||||
(
|
||||
@@ -977,6 +1009,7 @@ async fn handle_web_upload(
|
||||
speed_bps: 0,
|
||||
state: TransferState::InProgress,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
};
|
||||
state.active_transfers.write().await.insert(session_id.clone(), initial_progress.clone());
|
||||
let _ = state.transfer_events.send(initial_progress);
|
||||
|
||||
+54
-10
@@ -391,7 +391,7 @@ impl ZeroSendApp {
|
||||
selected_peer_ids: HashSet::new(),
|
||||
queued_files: initial_files,
|
||||
transfers: Vec::new(),
|
||||
logs: vec!["⚡ ZeroSend v2.5.3-beta ready".to_string()],
|
||||
logs: vec!["⚡ ZeroSend v2.5.3.1-beta ready".to_string()],
|
||||
text_input: String::new(),
|
||||
analyzed_preview: analyze_text(""),
|
||||
recipient_pin: String::new(),
|
||||
@@ -3339,6 +3339,9 @@ impl ZeroSendApp {
|
||||
TransferState::Rejected => {
|
||||
ui.label(RichText::new("⛔ Передача отклонена").color(Color32::RED).size(12.0));
|
||||
}
|
||||
TransferState::Paused => {
|
||||
ui.label(RichText::new("⏸ На паузе").color(Color32::from_rgb(234, 179, 8)).strong().size(13.0));
|
||||
}
|
||||
TransferState::Canceled => {
|
||||
ui.label(RichText::new("🛑 Отменено").color(theme.text_muted).size(12.0));
|
||||
}
|
||||
@@ -3367,7 +3370,15 @@ impl ZeroSendApp {
|
||||
100
|
||||
};
|
||||
|
||||
let size_details = if t.state == TransferState::InProgress {
|
||||
let comp_info = if t.compressed_bytes > 0 && t.compressed_bytes < t.bytes_transferred {
|
||||
let saved_pct = ((t.bytes_transferred.saturating_sub(t.compressed_bytes) as f64 / t.bytes_transferred as f64) * 100.0) as u32;
|
||||
let ratio = t.bytes_transferred as f64 / t.compressed_bytes.max(1) as f64;
|
||||
format!(" • 🗜️ {:.1}x (сжато на {}%)", ratio, saved_pct)
|
||||
} else {
|
||||
String::new()
|
||||
};
|
||||
|
||||
let size_details = if t.state == TransferState::InProgress || t.state == TransferState::Paused {
|
||||
let speed = t.speed_bps.max(1);
|
||||
let remaining_bytes = t.total_bytes.saturating_sub(t.bytes_transferred);
|
||||
let eta_secs = remaining_bytes / speed;
|
||||
@@ -3380,19 +3391,22 @@ impl ZeroSendApp {
|
||||
};
|
||||
|
||||
let speed_mb = (t.speed_bps as f64) / 1_048_576.0;
|
||||
let state_prefix = if t.state == TransferState::Paused { "⏸ На паузе • " } else { "" };
|
||||
format!(
|
||||
"{} / {} ({}%) • ⚡ {:.1} MB/s • {}",
|
||||
"{}{} / {} ({}%) • ⚡ {:.1} MB/s • {}{}",
|
||||
state_prefix,
|
||||
human_bytes(t.bytes_transferred as f64),
|
||||
human_bytes(t.total_bytes as f64),
|
||||
progress_pct,
|
||||
speed_mb,
|
||||
eta_str
|
||||
eta_str,
|
||||
comp_info
|
||||
)
|
||||
} else if is_done {
|
||||
if t.checksum_verified {
|
||||
format!("Доставлено • Размер: {} • 🛡️ SHA-256 Проверено", human_bytes(t.total_bytes as f64))
|
||||
format!("Доставлено • Размер: {}{} • 🛡️ SHA-256 Проверено", human_bytes(t.total_bytes as f64), comp_info)
|
||||
} else {
|
||||
format!("Доставлено • Размер: {}", human_bytes(t.total_bytes as f64))
|
||||
format!("Доставлено • Размер: {}{}", human_bytes(t.total_bytes as f64), comp_info)
|
||||
}
|
||||
} else {
|
||||
format!("Размер: {}", human_bytes(t.total_bytes as f64))
|
||||
@@ -3433,6 +3447,30 @@ impl ZeroSendApp {
|
||||
let _ = std::process::Command::new("explorer").arg(&target_file).spawn();
|
||||
}
|
||||
});
|
||||
} else if is_active || t.state == TransferState::Paused {
|
||||
ui.with_layout(Layout::right_to_left(Align::Center), |ui| {
|
||||
let client_c = self.client.clone();
|
||||
let s_id = t.session_id.clone();
|
||||
if t.state == TransferState::Paused {
|
||||
let resume_btn = Button::new(RichText::new("▶️ Продолжить").strong().color(Color32::WHITE))
|
||||
.fill(Color32::from_rgb(34, 197, 94))
|
||||
.rounding(Rounding::same(6.0));
|
||||
if ui.add(resume_btn).clicked() {
|
||||
tokio::spawn(async move {
|
||||
client_c.resume_transfer(&s_id).await;
|
||||
});
|
||||
}
|
||||
} else {
|
||||
let pause_btn = Button::new(RichText::new("⏸ Пауза").strong().color(Color32::WHITE))
|
||||
.fill(Color32::from_rgb(234, 179, 8))
|
||||
.rounding(Rounding::same(6.0));
|
||||
if ui.add(pause_btn).clicked() {
|
||||
tokio::spawn(async move {
|
||||
client_c.pause_transfer(&s_id).await;
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
@@ -3952,13 +3990,18 @@ impl ZeroSendApp {
|
||||
.stroke(theme.stroke_border)
|
||||
.inner_margin(Margin::same(14.0))
|
||||
.show(ui, |ui| {
|
||||
// v2.5.3-beta
|
||||
// v2.5.3.1-beta
|
||||
ui.horizontal(|ui| {
|
||||
ui.label(RichText::new("v2.5.3-beta").strong().color(Color32::from_rgb(34, 197, 94)));
|
||||
ui.label(RichText::new("v2.5.3.1-beta").strong().color(Color32::from_rgb(34, 197, 94)));
|
||||
ui.label(RichText::new("(Текущая версия)").size(11.0).color(Color32::GRAY));
|
||||
});
|
||||
ui.label(" • ⚡ Потоковая передача папок: Zip-на-лету через tar.zst — папки стримятся напрямую без создания временных файлов на диске.");
|
||||
ui.label(" • 🗜️ Zstandard (Zstd) сжатие: Мгновенное сжатие уровня 1-3 в потоке — до 4-8x прирост скорости на 100Mбит и VPN-каналах.");
|
||||
ui.label(" • 🗜️ Умная Zstd-компрессия: Автоматический пропуск уже сжатых файлов (.jpg, .mp4, .zip, .rar...) и сжатие кода/текста/документов.");
|
||||
ui.label(" • 🔄 Delta-transfer для Auto-Sync: Blake3 хэширование и кэш измененных файлов — синхронизация только изменений без повторной передачи.");
|
||||
ui.label(" • 👥 Мульти-отправка: Параллельная передача выбранных файлов сразу нескольким друзьям.");
|
||||
ui.label(" • ⏸️ Пауза и возобновление: Кнопки паузы/продолжения передачи прямо в дашборде мониторинга.");
|
||||
ui.label(" • 🏎️ Очередь с приоритетами: Маленькие файлы передаются первыми для мгновенного завершения.");
|
||||
ui.label(" • 📊 Мониторинг сжатия: Отображение реального коэффициента сжатия и сэкономленного трафика.");
|
||||
ui.label(" • 🏷️ Умный парсер версий: Поддержка 4-значных номеров (2.5.3.1) и автоматическое определение Beta-релизов.");
|
||||
ui.add_space(8.0);
|
||||
ui.separator();
|
||||
ui.add_space(8.0);
|
||||
@@ -4329,6 +4372,7 @@ impl ZeroSendApp {
|
||||
speed_bps: 0,
|
||||
state: TransferState::InProgress,
|
||||
checksum_verified: false,
|
||||
compressed_bytes: 0,
|
||||
};
|
||||
|
||||
if let Some(pos) = self.transfers.iter().position(|x| x.session_id == session_id) {
|
||||
|
||||
@@ -479,6 +479,7 @@ fn draw_ui(f: &mut Frame, app: &mut AppState) {
|
||||
TransferState::Completed => "✅ Completed".to_string(),
|
||||
TransferState::Failed(e) => format!("❌ Failed: {}", e),
|
||||
TransferState::Rejected => "⛔ Rejected".to_string(),
|
||||
TransferState::Paused => "⏸ Paused".to_string(),
|
||||
TransferState::Canceled => "🛑 Canceled".to_string(),
|
||||
};
|
||||
|
||||
|
||||
+64
-13
@@ -43,6 +43,8 @@ pub struct UpdateInfo {
|
||||
pub download_url: Option<String>,
|
||||
pub gitea_url: String,
|
||||
pub asset_size: u64,
|
||||
pub is_beta: bool,
|
||||
pub pre_number: u32,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
@@ -54,21 +56,64 @@ pub enum UpdateProgress {
|
||||
Error(String),
|
||||
}
|
||||
|
||||
/// Parse semantic version string (e.g. "v2.4.1", "2.4.2-beta") into (u32, u32, u32)
|
||||
pub fn parse_version(v: &str) -> (u32, u32, u32) {
|
||||
/// Parse semantic version string into (major, minor, patch, subpatch, is_beta, pre_num).
|
||||
/// Handles formats like "v2.5.3", "2.5.3-beta", "2.5.3-beta.1", "2.5.3.1"
|
||||
pub fn parse_version(v: &str) -> (u32, u32, u32, u32, bool, u32) {
|
||||
let clean = v.trim().trim_start_matches('v').trim_start_matches('V');
|
||||
let parts: Vec<&str> = clean.split('.').collect();
|
||||
let major = parts.first().and_then(|s| s.split('-').next()).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
|
||||
let minor = parts.get(1).and_then(|s| s.split('-').next()).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
|
||||
let patch = parts.get(2).and_then(|s| s.split('-').next()).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
|
||||
(major, minor, patch)
|
||||
|
||||
// Split off pre-release suffix (everything after the first '-')
|
||||
let (numeric_part, pre_part) = if let Some(idx) = clean.find('-') {
|
||||
(&clean[..idx], Some(&clean[idx + 1..]))
|
||||
} else {
|
||||
(clean, None)
|
||||
};
|
||||
|
||||
let num_parts: Vec<&str> = numeric_part.split('.').collect();
|
||||
let major = num_parts.first().and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
|
||||
let minor = num_parts.get(1).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
|
||||
let patch = num_parts.get(2).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
|
||||
let subpatch = num_parts.get(3).and_then(|s| s.parse::<u32>().ok()).unwrap_or(0);
|
||||
|
||||
let is_beta = pre_part.map(|p| {
|
||||
let pl = p.to_lowercase();
|
||||
pl.contains("beta") || pl.contains("alpha") || pl.contains("rc")
|
||||
}).unwrap_or(false);
|
||||
|
||||
// Extract pre-release number, e.g. "beta.1" -> 1, "beta" -> 0
|
||||
let pre_num = pre_part
|
||||
.and_then(|p| p.split('.').last())
|
||||
.and_then(|s| s.parse::<u32>().ok())
|
||||
.unwrap_or(0);
|
||||
|
||||
(major, minor, patch, subpatch, is_beta, pre_num)
|
||||
}
|
||||
|
||||
/// Returns true if latest version is strictly newer than current version
|
||||
/// Returns true if a version string contains "beta", "alpha", or "rc"
|
||||
pub fn is_beta_version(v: &str) -> bool {
|
||||
let lower = v.to_lowercase();
|
||||
lower.contains("beta") || lower.contains("alpha") || lower.contains("rc")
|
||||
}
|
||||
|
||||
/// Returns true if latest version is strictly newer than current version.
|
||||
/// Rules: stable > beta of same numeric version; higher pre_num beats lower.
|
||||
pub fn is_newer_version(latest: &str, current: &str) -> bool {
|
||||
let (l_maj, l_min, l_pat) = parse_version(latest);
|
||||
let (c_maj, c_min, c_pat) = parse_version(current);
|
||||
(l_maj, l_min, l_pat) > (c_maj, c_min, c_pat)
|
||||
let (l_maj, l_min, l_pat, l_sub, l_beta, l_pre) = parse_version(latest);
|
||||
let (c_maj, c_min, c_pat, c_sub, c_beta, c_pre) = parse_version(current);
|
||||
|
||||
// Compare numeric components first
|
||||
let l_num = (l_maj, l_min, l_pat, l_sub);
|
||||
let c_num = (c_maj, c_min, c_pat, c_sub);
|
||||
|
||||
if l_num != c_num {
|
||||
return l_num > c_num;
|
||||
}
|
||||
|
||||
// Same numeric version: stable > beta
|
||||
match (l_beta, c_beta) {
|
||||
(false, true) => true, // latest is stable, current is beta → update
|
||||
(true, false) => false, // latest is beta, current is stable → no update
|
||||
_ => l_pre > c_pre, // both same prerelease type: compare pre number
|
||||
}
|
||||
}
|
||||
|
||||
/// Check Gitea repository for available releases or newer version tags
|
||||
@@ -110,13 +155,15 @@ pub async fn check_gitea_update(api_base: Option<&str>) -> Result<UpdateInfo, St
|
||||
|
||||
return Ok(UpdateInfo {
|
||||
current_version,
|
||||
latest_version: latest_ver,
|
||||
latest_version: latest_ver.clone(),
|
||||
has_update,
|
||||
release_name: release.name,
|
||||
release_notes: release.body,
|
||||
download_url,
|
||||
gitea_url,
|
||||
asset_size,
|
||||
is_beta: is_beta_version(&latest_ver),
|
||||
pre_number: parse_version(&latest_ver).5,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -149,13 +196,15 @@ pub async fn check_gitea_update(api_base: Option<&str>) -> Result<UpdateInfo, St
|
||||
|
||||
return Ok(UpdateInfo {
|
||||
current_version,
|
||||
latest_version: latest_ver,
|
||||
latest_version: latest_ver.clone(),
|
||||
has_update,
|
||||
release_name: release.name,
|
||||
release_notes: release.body,
|
||||
download_url,
|
||||
gitea_url,
|
||||
asset_size,
|
||||
is_beta: is_beta_version(&latest_ver),
|
||||
pre_number: parse_version(&latest_ver).5,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -182,6 +231,8 @@ pub async fn check_gitea_update(api_base: Option<&str>) -> Result<UpdateInfo, St
|
||||
download_url: None,
|
||||
gitea_url: GITEA_DEFAULT_REPO_URL.to_string(),
|
||||
asset_size: 0,
|
||||
is_beta: is_beta_version(parsed),
|
||||
pre_number: parse_version(parsed).5,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user