//! Central SFTP Transfer Manager (port of `files/transferManager.js`). //! //! Every file is one job with its own SFTP channel, a bounded pipeline of positioned //! reads/writes and a confirmed offset that only advances over contiguous acknowledged chunks. //! Pause and cancel stop issuing requests; resume reopens the partial file, verifies the tail //! against the source and continues from that offset. Completed uploads are published by the //! shared remote helper (verified hash, original metadata, atomic rename, privileged when //! needed); completed downloads are hashed locally and renamed into place. Job state (never //! secrets) survives restarts in `transfers.json`, in the same format as 0.1.0, so transfers //! interrupted by the update can be resumed. use super::editor::{remote_checksum, Authorization}; use super::pipeline::{pump, verify_tail, Consumer, LocalFile, Producer, PumpStop, TAIL_CHECK}; use super::posix; use super::service::{file_error, normalize_remote_path, or_root, safe_entry_name, RemoteFileService}; use crate::core::error::{CoreError, CoreResult}; use crate::core::logging::Logger; use crate::core::ssh::session::SessionManager; use crate::core::ssh::sftp::{FileHandle, Sftp, SftpError, Stat}; use futures::future::BoxFuture; use futures::{FutureExt, StreamExt}; use parking_lot::Mutex; use rand::RngCore; use serde::{Deserialize, Deserializer, Serialize}; use serde_json::{json, Map, Value}; use sha2::{Digest, Sha256}; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::Duration; use tokio::sync::watch; use zeroize::Zeroizing; const STATES: [&str; 11] = ["waiting", "connecting", "transferring", "paused", "interrupted", "reconnecting", "verifying", "finalizing", "completed", "failed", "cancelled"]; const RUNNING: [&str; 4] = ["connecting", "transferring", "verifying", "finalizing"]; const FINAL: [&str; 3] = ["completed", "failed", "cancelled"]; const EMIT_MS: u64 = 250; const PERSIST_MS: u64 = 1000; const RETRY_DELAYS_MS: [u64; 5] = [2000, 4000, 8000, 16000, 32000]; const MAX_TOTAL_RETRIES: u32 = 50; const MAX_HISTORY: usize = 500; const MAX_JOBS: usize = 20000; const MAX_PLAN_ENTRIES: usize = 10000; const MAX_PLAN_BYTES: u64 = 4 * 1024 * 1024 * 1024 * 1024; const PLAN_TTL_MS: i64 = 10 * 60 * 1000; const ORPHAN_AGE_MS: i64 = 24 * 3600 * 1000; const HASH_RATE: u64 = 10 * 1024 * 1024; const CONNECTION_CODES: [&str; 9] = ["SESSION_NOT_CONNECTED", "SFTP_OPEN_FAILED", "TRANSFER_STALLED", "REMOTE_COMMAND_TIMEOUT", "REMOTE_COMMAND_FAILED", "FILE_EXEC_FAILED", "ECONNRESET", "ETIMEDOUT", "EPIPE"]; const NO_HELPER_CODES: [&str; 2] = ["FILE_DEPENDENCY_UNAVAILABLE", "FEATURE_PLATFORM_UNSUPPORTED"]; const AUTH_CODES: [&str; 4] = ["SUDO_PASSWORD_REQUIRED", "SUDO_PASSWORD_INCORRECT", "FILE_PERMISSION_DENIED", "FILE_METADATA_PERMISSION"]; fn now() -> i64 { chrono::Utc::now().timestamp_millis() } fn hex_id(bytes: usize) -> String { let mut buffer = vec![0u8; bytes]; rand::rngs::OsRng.fill_bytes(&mut buffer); hex::encode(buffer) } fn part_name(name: &str, id: &str) -> String { format!(".{}.chj-part-{id}", name.chars().take(120).collect::()) } fn hash_timeout(size: u64) -> u64 { 120_000 + (size as f64 / HASH_RATE as f64 * 1000.0).ceil() as u64 } // --- lenient JSON numbers (0.1.0 wrote JavaScript numbers) ------------------------------------------ fn number(value: &Value) -> Option { match value { Value::Number(n) => n.as_f64(), Value::String(s) => s.parse().ok(), _ => None, } } fn de_u64<'de, D: Deserializer<'de>>(d: D) -> Result { Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u64).unwrap_or(0)) } fn de_i64<'de, D: Deserializer<'de>>(d: D) -> Result { Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite()).map(|n| n as i64).unwrap_or(0)) } fn de_opt_i64<'de, D: Deserializer<'de>>(d: D) -> Result, D::Error> { Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite()).map(|n| n as i64)) } fn de_u32<'de, D: Deserializer<'de>>(d: D) -> Result { Ok(number(&Value::deserialize(d)?).filter(|n| n.is_finite() && *n >= 0.0).map(|n| n as u32).unwrap_or(0)) } /// Keeps JSON `null` as `Some(Null)` (0.1.0 distinguishes `undefined` from `null`). fn de_present<'de, D: Deserializer<'de>>(d: D) -> Result, D::Error> { Ok(Some(Value::deserialize(d)?)) } fn de_opt_string<'de, D: Deserializer<'de>>(d: D) -> Result, D::Error> { Ok(match Value::deserialize(d)? { Value::String(s) => Some(s), _ => None, }) } #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct SessionIdentity { #[serde(default)] pub profile_id: Value, #[serde(default)] pub host: Value, #[serde(default)] pub port: Value, #[serde(default)] pub username: Value, #[serde(default)] pub jump: Vec, } impl SessionIdentity { fn key(&self) -> String { let profile = self.profile_id.as_str().filter(|id| !id.starts_with("adhoc:")).unwrap_or(""); serde_json::to_string(&json!([profile, self.host, self.port, self.username, self.jump])).unwrap_or_default() } fn of_status(status: &Value) -> SessionIdentity { SessionIdentity { profile_id: status["profileId"].clone(), host: status["host"].clone(), port: status["port"].clone(), username: status["username"].clone(), jump: status["jump"].as_array().map(|hops| hops.iter().map(|hop| json!({ "host": hop["host"], "port": hop["port"], "username": hop["username"] })).collect()).unwrap_or_default(), } } } fn matching_session(sessions: &[Value], identity: &SessionIdentity, preferred: &str) -> Option { sessions.iter().find(|s| s["state"] == "connected" && s["sessionId"] == preferred && SessionIdentity::of_status(s) == *identity) .and_then(|s| s["sessionId"].as_str().map(str::to_string)) } #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] #[serde(rename_all = "camelCase")] pub struct LocalIdentity { #[serde(default, deserialize_with = "de_u64")] pub size: u64, #[serde(default, deserialize_with = "de_i64")] pub mtime_ms: i64, #[serde(default, deserialize_with = "de_u64")] pub ino: u64, #[serde(default, deserialize_with = "de_u64")] pub dev: u64, } impl LocalIdentity { pub fn of(meta: &std::fs::Metadata) -> LocalIdentity { let mtime_ms = meta .modified() .ok() .map(|t| match t.duration_since(std::time::UNIX_EPOCH) { Ok(d) => d.as_millis() as i64, Err(e) => -(e.duration().as_millis() as i64), }) .unwrap_or(0); #[cfg(unix)] let (ino, dev) = { use std::os::unix::fs::MetadataExt; (meta.ino(), meta.dev()) }; // Stable Rust exposes no file index on Windows; size + mtime identify the source. #[cfg(not(unix))] let (ino, dev) = (0, 0); LocalIdentity { size: meta.len(), mtime_ms, ino, dev } } fn same(&self, other: &LocalIdentity) -> bool { let loose = cfg!(not(unix)) || self.ino == 0 || other.ino == 0; self.size == other.size && self.mtime_ms == other.mtime_ms && (loose || (self.ino == other.ino && self.dev == other.dev)) } } #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] pub struct RemoteIdentity { #[serde(default, deserialize_with = "de_u64")] pub size: u64, #[serde(default, deserialize_with = "de_u64")] pub mtime: u64, } #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] pub struct JobError { #[serde(default)] pub code: String, #[serde(default)] pub message: String, } #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct Job { pub id: String, #[serde(default)] pub owner: String, #[serde(default)] pub batch_id: String, #[serde(default)] pub direction: String, #[serde(default)] pub identity: SessionIdentity, #[serde(default)] pub session_id: Value, #[serde(default, deserialize_with = "de_u64")] pub order: u64, #[serde(default)] pub name: String, #[serde(default)] pub local_path: String, #[serde(default)] pub remote_path: String, #[serde(default, deserialize_with = "de_u64")] pub size: u64, #[serde(default, deserialize_with = "de_u64")] pub transferred: u64, #[serde(default, deserialize_with = "de_u64")] pub moved_bytes: u64, #[serde(default, deserialize_with = "de_u64")] pub active_ms: u64, pub state: String, #[serde(default)] pub mode: String, #[serde(default)] pub overwrite: bool, #[serde(default)] pub local_identity: Option, #[serde(default)] pub remote_identity: Option, #[serde(default)] pub remote_mode: Option, #[serde(default)] pub finalize: Value, #[serde(default, deserialize_with = "de_u32")] pub attempts: u32, #[serde(default, deserialize_with = "de_u32")] pub total_retries: u32, #[serde(default)] pub error: Option, #[serde(default, deserialize_with = "de_i64")] pub created_at: i64, #[serde(default, deserialize_with = "de_i64")] pub updated_at: i64, #[serde(default, deserialize_with = "de_opt_i64", skip_serializing_if = "Option::is_none")] pub started_at: Option, #[serde(default, deserialize_with = "de_opt_i64", skip_serializing_if = "Option::is_none")] pub finished_at: Option, #[serde(default, deserialize_with = "de_opt_i64")] pub next_retry_at: Option, #[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")] pub resolved_path: Option, #[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")] pub part_path: Option, /// `None` = not prepared yet (0.1.0 `undefined`), `Some(Null)` = target did not exist. #[serde(default, deserialize_with = "de_present", skip_serializing_if = "Option::is_none")] pub target_baseline: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub helper: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub owner_uid: Option, #[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")] pub hash: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub verification: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub staged: Option, #[serde(default, deserialize_with = "de_opt_string", skip_serializing_if = "Option::is_none")] pub local_part: Option, #[serde(default)] pub auth_required: bool, #[serde(default)] pub resumable: bool, #[serde(default)] pub finalize_with_sudo: bool, #[serde(default)] pub cleanup_pending: bool, /// Unknown fields written by other versions are kept unchanged. #[serde(flatten)] pub extra: Map, } impl Job { fn finalize_state(&self) -> &str { self.finalize["state"].as_str().unwrap_or("pending") } fn session(&self) -> String { self.session_id.as_str().unwrap_or("").to_string() } } #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct Settings { pub max_concurrent: u32, pub auto_retry: bool, pub verify: bool, } impl Default for Settings { fn default() -> Self { Settings { max_concurrent: 2, auto_retry: true, verify: true } } } #[derive(Debug, Clone)] struct PlanFile { name: String, local_path: String, remote_path: String, size: u64, local_identity: Option, remote_identity: Option, mode: Option, conflict: bool, overwrite_confirmed: bool, } #[derive(Debug, Clone)] struct Plan { id: String, owner: String, direction: &'static str, session_id: String, identity: SessionIdentity, directory: String, local_directory: String, files: Vec, directories: Vec, skipped: Vec, bytes: u64, created_at: i64, mode: String, privilege: Option, /// Server-to-server: the source session and its identity. The plan's session is the destination. relay: Option<(String, SessionIdentity)>, /// Source folders of a move, removed when empty after the batch (a job exists only per file). source_directories: Vec, } struct Runtime { running: bool, stop: watch::Sender, generation: u64, speed: f64, samples: u32, last_bytes: u64, last_time: i64, active_since: Option, progress_since_resume: u64, retry_task: Option>, } impl Runtime { fn new(transferred: u64) -> Self { Runtime { running: false, stop: watch::channel(String::new()).0, generation: 0, speed: 0.0, samples: 0, last_bytes: transferred, last_time: now(), active_since: None, progress_since_resume: 0, retry_task: None } } } struct BatchMove { /// `None` = the folders are on this computer. session: Option, directories: Vec, } #[derive(Default)] struct Inner { batch_moves: HashMap, jobs: HashMap, runtime: HashMap, secrets: HashMap>, plans: HashMap, settings: Settings, order: u64, changed: HashSet, pending_owners: HashSet, full_owners: HashSet, emit_scheduled: bool, cleaned_identities: HashSet, closed: bool, } /// Stop requested (pause/cancel/shutdown/lost) or a real failure. enum RunError { Stop(String), Error(CoreError), } impl From for RunError { fn from(error: CoreError) -> Self { RunError::Error(error) } } type RunResult = Result; /// 0.1.0 `errorCode` for Core errors. fn error_code(error: &CoreError) -> String { if !error.code.is_empty() { return error.code.clone(); } let lower = error.message.to_lowercase(); if ["no response", "not connected", "connection lost", "connection closed", "channel", "econnreset", "socket"].iter().any(|p| lower.contains(p)) { return "SESSION_NOT_CONNECTED".into(); } "TRANSFER_FAILED".into() } fn sftp_core(error: SftpError) -> CoreError { CoreError::new(error.transfer_code(), error.message()) } fn io_core(error: std::io::Error) -> CoreError { let code = match error.kind() { std::io::ErrorKind::NotFound => "ENOENT", std::io::ErrorKind::PermissionDenied => "EACCES", std::io::ErrorKind::StorageFull => "ENOSPC", std::io::ErrorKind::AlreadyExists => "EEXIST", _ => "EIO", }; CoreError::new(code, error.to_string()) } fn percent(transferred: u64, size: u64) -> Value { if size == 0 { return Value::Null; } json!(((transferred.min(size) as f64 / size as f64) * 1000.0).floor().min(1000.0) / 10.0) } /// Aggregate view: percentages are byte based. pub fn summarize(jobs: &[Value]) -> Value { let counted: Vec<&Value> = jobs.iter().filter(|j| j["state"] != "cancelled").collect(); let sizes_known = counted.iter().all(|j| j["size"].is_number()); let total: f64 = counted.iter().map(|j| j["size"].as_f64().unwrap_or(0.0)).sum(); let transferred: f64 = counted.iter().map(|j| j["transferred"].as_f64().unwrap_or(0.0).min(j["size"].as_f64().unwrap_or(f64::INFINITY))).sum(); let completed = counted.iter().filter(|j| j["state"] == "completed").count(); let active: Vec<&Value> = jobs.iter().filter(|j| RUNNING.contains(&j["state"].as_str().unwrap_or(""))).collect(); let speed: f64 = active.iter().map(|j| j["speed"].as_f64().unwrap_or(0.0)).sum(); let remaining: f64 = counted.iter().filter(|j| !FINAL.contains(&j["state"].as_str().unwrap_or(""))).map(|j| (j["size"].as_f64().unwrap_or(0.0) - j["transferred"].as_f64().unwrap_or(0.0)).max(0.0)).sum(); let eta_reliable = !active.is_empty() && active.iter().all(|j| j["state"] != "transferring" || !j["eta"].is_null()); let mut overall: Option = None; if sizes_known && total > 0.0 { overall = Some((transferred / total * 1000.0).floor() / 10.0); } else if sizes_known && !counted.is_empty() && completed == counted.len() { overall = Some(100.0); } if overall.is_some_and(|o| o >= 100.0) && completed != counted.len() { overall = Some(99.9); } let count = |state: &str| jobs.iter().filter(|j| j["state"] == state).count(); json!({ "total": jobs.len(), "completed": completed, "active": active.len(), "waiting": count("waiting") + count("reconnecting"), "paused": count("paused"), "interrupted": count("interrupted"), "failed": count("failed"), "cancelled": jobs.len() - counted.len(), "totalBytes": total as u64, "transferredBytes": transferred as u64, "sizesKnown": sizes_known, "percent": overall, "speed": speed, "eta": if speed > 0.0 && eta_reliable && remaining > 0.0 { json!((remaining / speed).ceil() as u64) } else if remaining == 0.0 && completed == counted.len() { json!(0) } else { Value::Null } }) } /// Native dialogs for one planning call (shown by the command layer for the calling module). pub type PickPaths = Box BoxFuture<'static, Option>> + Send>; pub type PickSavePath = Box BoxFuture<'static, Option> + Send>; pub type PickDirectory = Box BoxFuture<'static, Option> + Send>; pub type UpdateEmitter = Arc; pub struct TransferManager { sessions: Arc, files: Arc, logger: Arc, file: Option, inner: Mutex, emitter: UpdateEmitter, persist_pending: AtomicBool, cleanup_scheduled: AtomicBool, scheduling: AtomicBool, } impl TransferManager { pub fn new(sessions: Arc, files: Arc, storage_root: Option<&Path>, logger: Arc, emitter: UpdateEmitter) -> Arc { let manager = Arc::new(TransferManager { sessions, files, logger, file: storage_root.map(|r| r.join("transfers.json")), inner: Mutex::new(Inner::default()), emitter, persist_pending: AtomicBool::new(false), cleanup_scheduled: AtomicBool::new(false), scheduling: AtomicBool::new(false), }); manager.load(); manager } /// Subscribes to session state changes (must run inside the async runtime). pub fn start_session_listener(self: &Arc) { let mut events = self.sessions.state_events.subscribe(); let weak = Arc::downgrade(self); tokio::spawn(async move { loop { match events.recv().await { Ok(status) => match weak.upgrade() { Some(manager) => manager.session_state(&status), None => break, }, Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, Err(_) => break, } } }); } // ---- public API (owner = module id) ---------------------------------------------------------- pub fn list(self: &Arc, owner: &str) -> Value { let jobs = { let inner = self.inner.lock(); Self::owned(&inner, owner).iter().map(|j| Self::public(&inner, j)).collect::>() }; self.cleanup_orphans_later(); let settings = self.inner.lock().settings.clone(); json!({ "jobs": jobs, "summary": summarize(&jobs), "settings": settings }) } pub fn update_settings(self: &Arc, patch: &Value) -> CoreResult { let settings = { let mut inner = self.inner.lock(); if let Some(value) = patch.get("maxConcurrent").filter(|v| !v.is_null()) { match number(value) { Some(n) if n.fract() == 0.0 && (1.0..=8.0).contains(&n) => inner.settings.max_concurrent = n as u32, _ => return Err(file_error("TRANSFER_INVALID_SETTINGS", "Concurrent transfers must be between 1 and 8.")), } } if let Some(value) = patch.get("autoRetry").filter(|v| !v.is_null()) { inner.settings.auto_retry = *value == true; } if let Some(value) = patch.get("verify").filter(|v| !v.is_null()) { inner.settings.verify = *value == true; } inner.settings.clone() }; self.persist(true, false); self.schedule(); Ok(serde_json::to_value(settings).unwrap_or_default()) } pub fn identity(&self, session_id: &str) -> CoreResult { let session = self .sessions .list() .into_iter() .find(|s| s["sessionId"] == session_id && s["state"] == "connected") .ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "The SSH session is not connected."))?; Ok(SessionIdentity::of_status(&session)) } pub async fn plan_upload(self: &Arc, owner: &str, session_id: &str, directory: &Value, pick: PickPaths) -> CoreResult { let directory = normalize_remote_path(&or_root(directory))?; let identity = self.identity(session_id)?; let mut local_paths: Vec = Vec::new(); for path in pick().await.unwrap_or_default() { if !local_paths.contains(&path) { local_paths.push(path); } } if local_paths.is_empty() { return Ok(json!({ "canceled": true })); } let mut plan = Plan { id: hex_id(16), owner: owner.into(), direction: "upload", session_id: session_id.into(), identity, directory: directory.clone(), local_directory: String::new(), files: Vec::new(), directories: Vec::new(), skipped: Vec::new(), bytes: 0, created_at: now(), mode: String::new(), privilege: None, relay: None, source_directories: Vec::new(), }; let walked = { let directory = directory.clone(); tokio::task::spawn_blocking(move || -> CoreResult<(Vec, Vec, Vec, Vec, u64)> { let mut walk = LocalWalk::default(); for local in &local_paths { let name = safe_entry_name(&local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default())?; walk.visit(local, &posix::join(&directory, &name), 0)?; } Ok((walk.files, walk.directories, walk.source_directories, walk.skipped, walk.bytes)) }) .await .map_err(|_| CoreError::plain("Local folder scan failed."))?? }; (plan.files, plan.directories, plan.source_directories, plan.skipped, plan.bytes) = walked; let sftp = self.files.sftp(session_id).await?; plan.mode = directory_mode(&sftp, &directory).await?.into(); remote_conflicts(&sftp, &mut plan).await; drop(sftp); if plan.mode == "privileged" { let privilege = self.files.privileges(session_id).await?; if privilege["mode"] == "none" { let code = if privilege["sudoInstalled"] == true { "SUDO_NOT_PERMITTED" } else { "SUDO_UNAVAILABLE" }; return Err(file_error(code, "The destination is not writable and the SSH account cannot use sudo.")); } plan.privilege = Some(privilege); } self.store_plan(plan) } pub async fn plan_download(self: &Arc, owner: &str, session_id: &str, paths: &Value, options: &Value, pick_file: PickSavePath, pick_directory: PickDirectory) -> CoreResult { let identity = self.identity(session_id)?; let values: Vec = match paths { Value::Array(list) => list.clone(), other => vec![other.clone()], }; let mut targets: Vec = Vec::new(); for value in values.into_iter().filter(|v| !v.is_null()) { let target = normalize_remote_path(&value)?; if !targets.contains(&target) { targets.push(target); } } if targets.is_empty() || targets.len() > 100 { return Err(file_error("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries.")); } if targets.iter().any(|t| t == "/") { return Err(file_error("TRANSFER_INVALID_SELECTION", "Downloading the entire remote root is not allowed in one batch.")); } let sudo = options["sudo"] == true; let mut plan = Plan { id: hex_id(16), owner: owner.into(), direction: "download", session_id: session_id.into(), identity, directory: String::new(), local_directory: String::new(), files: Vec::new(), directories: Vec::new(), skipped: Vec::new(), bytes: 0, created_at: now(), mode: if sudo { "privileged-allowed".into() } else { "direct".into() }, privilege: None, relay: None, source_directories: Vec::new(), }; let sftp = self.files.sftp(session_id).await?; let first = if targets.len() == 1 { Some(self.remote_stat(&sftp, &targets[0], options, session_id).await?) } else { None }; if let Some(info) = first.filter(|i| i.kind == "file") { let Some(chosen) = pick_file(posix::basename(&targets[0])).await else { return Ok(json!({ "canceled": true })) }; plan.local_directory = chosen.parent().map(|p| p.to_string_lossy().to_string()).unwrap_or_default(); add_download(&mut plan, &targets[0], &chosen.to_string_lossy(), &info); plan.files[0].overwrite_confirmed = options["noSaveDialog"] != true; // The native save dialog already confirmed replacing. } else { let Some(chosen) = pick_directory().await else { return Ok(json!({ "canceled": true })) }; plan.local_directory = chosen.to_string_lossy().to_string(); for target in &targets { let local = chosen.join(safe_entry_name(&posix::basename(target))?); self.walk_remote(&sftp, session_id, &mut plan, target, &local.to_string_lossy(), options, true, 0).await?; } } drop(sftp); for file in plan.files.iter_mut() { if !file.overwrite_confirmed && std::fs::symlink_metadata(&file.local_path).is_ok() { file.conflict = true; } } self.store_plan(plan) } /// Server-to-server plan. The data streams through this computer: nothing is stored on its disk. pub async fn plan_relay(self: &Arc, owner: &str, source_session: &str, paths: &Value, dest_session: &str, directory: &Value) -> CoreResult { let directory = normalize_remote_path(&or_root(directory))?; let identity = self.identity(dest_session)?; let source_identity = self.identity(source_session)?; let mut targets: Vec = Vec::new(); for value in paths.as_array().cloned().unwrap_or_default() { let target = normalize_remote_path(&value)?; if !targets.contains(&target) { targets.push(target); } } if targets.is_empty() || targets.len() > 100 || targets.iter().any(|t| t == "/") { return Err(file_error("TRANSFER_INVALID_SELECTION", "Select between 1 and 100 remote entries, not the remote root.")); } if source_identity == identity { for target in &targets { if posix::dirname(target) == directory { return Err(file_error("TRANSFER_SAME_PATH", "The source and the destination are the same folder.")); } if directory == *target || directory.starts_with(&format!("{target}/")) { return Err(file_error("TRANSFER_INTO_ITSELF", "A folder cannot be copied into itself.")); } } } let mut plan = Plan { id: hex_id(16), owner: owner.into(), direction: "relay", session_id: dest_session.into(), identity, directory: directory.clone(), local_directory: String::new(), files: Vec::new(), directories: Vec::new(), skipped: Vec::new(), bytes: 0, created_at: now(), mode: String::new(), privilege: None, relay: Some((source_session.to_string(), source_identity)), source_directories: Vec::new(), }; let source = self.files.sftp(source_session).await?; for target in &targets { let destination = posix::join(&directory, &safe_entry_name(&posix::basename(target))?); walk_relay(&source, &mut plan, target, &destination, 0).await?; } drop(source); let sftp = self.files.sftp(dest_session).await?; plan.mode = directory_mode(&sftp, &directory).await?.into(); remote_conflicts(&sftp, &mut plan).await; drop(sftp); if plan.mode == "privileged" { let privilege = self.files.privileges(dest_session).await?; if privilege["mode"] == "none" { let code = if privilege["sudoInstalled"] == true { "SUDO_NOT_PERMITTED" } else { "SUDO_UNAVAILABLE" }; return Err(file_error(code, "The destination is not writable and the SSH account cannot use sudo.")); } plan.privilege = Some(privilege); } self.store_plan(plan) } pub async fn start(self: &Arc, owner: &str, plan_id: &Value, options: &Value) -> CoreResult { let plan = { let inner = self.inner.lock(); inner.plans.get(plan_id.as_str().unwrap_or("")).filter(|p| p.owner == owner && now() - p.created_at <= PLAN_TTL_MS).cloned() } .ok_or_else(|| file_error("TRANSFER_PLAN_EXPIRED", "The transfer plan expired; select the files again."))?; if options["sessionId"].as_str().is_some_and(|id| id != plan.session_id) || self.identity(&plan.session_id)? != plan.identity { return Err(file_error("TRANSFER_PLAN_SERVER_CHANGED", "The transfer plan belongs to a different server or connection.")); } let mut job_extra = Map::new(); if let Some((source, source_identity)) = &plan.relay { if self.identity(source)? != *source_identity { return Err(file_error("TRANSFER_PLAN_SERVER_CHANGED", "The transfer plan belongs to a different server or connection.")); } job_extra.insert("relaySource".into(), json!({ "sessionId": source, "identity": source_identity })); } job_extra.insert("hostKey".into(), json!(self.sessions.trusted_host_key(&plan.session_id))); if let Some((source, _)) = &plan.relay { job_extra.insert("sourceHostKey".into(), json!(self.sessions.trusted_host_key(source))); } if options["move"] == true { job_extra.insert("moveSource".into(), json!(true)); job_extra.insert("moveRoot".into(), json!(options["sourceDirectory"].as_str().unwrap_or(""))); } let conflict = options["conflict"].as_str().filter(|c| ["overwrite", "skip"].contains(c)).unwrap_or("fail"); let conflicts = plan.files.iter().filter(|f| f.conflict).count(); if conflicts > 0 && conflict == "fail" { return Err(file_error("FILE_EXISTS", &format!("{conflicts} destination file(s) already exist."))); } if self.inner.lock().jobs.len() + plan.files.len() > MAX_JOBS { return Err(file_error("TRANSFER_QUEUE_FULL", "Remove finished transfers before starting more.")); } let password = options["sudoPassword"].as_str().unwrap_or("").to_string(); let privileged = plan.mode == "privileged" || (plan.direction == "download" && (plan.mode == "privileged-allowed" || options["sudo"] == true)); if plan.mode == "privileged" { // Verifies the sudo authorization once before any data moves (and caches it on request). let auth = Authorization { sudo: true, sudo_password: (!password.is_empty()).then(|| password.clone()), remember_sudo: options["rememberSudo"] == true, ..Default::default() }; self.files.helper(&plan.session_id, json!({ "operation": "stat", "path": plan.directory }), &auth).await?; } self.create_directories(&plan, privileged, &password).await?; let mut created = 0; { let mut inner = self.inner.lock(); for file in &plan.files { if file.conflict && conflict == "skip" { continue; } inner.order += 1; let order = inner.order; let job = Job { id: hex_id(16), owner: owner.into(), batch_id: plan.id.clone(), direction: plan.direction.into(), identity: plan.identity.clone(), session_id: json!(plan.session_id), order, name: file.name.clone(), local_path: file.local_path.clone(), remote_path: file.remote_path.clone(), size: file.size, transferred: 0, moved_bytes: 0, active_ms: 0, state: "waiting".into(), mode: if plan.direction != "download" { plan.mode.clone() } else if privileged { "privileged-allowed".into() } else { "direct".into() }, overwrite: conflict == "overwrite" || file.overwrite_confirmed, local_identity: file.local_identity.clone(), remote_identity: file.remote_identity.clone(), remote_mode: file.mode.map(Value::from), finalize: json!({ "state": "pending" }), attempts: 0, total_retries: 0, error: None, created_at: now(), updated_at: now(), started_at: None, finished_at: None, next_retry_at: None, resolved_path: None, part_path: None, target_baseline: None, helper: None, owner_uid: None, hash: None, verification: None, staged: None, local_part: None, auth_required: false, resumable: false, finalize_with_sudo: false, cleanup_pending: false, extra: job_extra.clone(), }; if !password.is_empty() && privileged { inner.secrets.insert(job.id.clone(), Zeroizing::new(password.clone())); } let id = job.id.clone(); inner.jobs.insert(id.clone(), job); created += 1; Self::mark(&mut inner, &id, owner); } inner.plans.remove(&plan.id); if options["move"] == true && !plan.source_directories.is_empty() { let session = plan.relay.as_ref().map(|(source, _)| source.clone()).or_else(|| (plan.direction == "download").then(|| plan.session_id.clone())); inner.batch_moves.insert(plan.id.clone(), BatchMove { session, directories: plan.source_directories.clone() }); } Self::trim_history(&mut inner); } self.finish_batch_move(&plan.id).await; self.emit_soon(); self.persist(true, false); self.logger.info("Transfer batch queued.", json!({ "direction": plan.direction, "files": created, "bytes": plan.bytes, "mode": plan.mode, "sessionId": plan.session_id })); self.schedule(); Ok(json!({ "batchId": plan.id, "jobs": created, "skipped": plan.files.len() - created })) } pub fn pause(self: &Arc, owner: &str, ids: &Value) -> Value { let mut count = 0; for id in self.select(owner, ids) { let state = self.job_state(&id); if ["waiting", "reconnecting", "interrupted"].contains(&state.as_str()) { self.clear_retry(&id); self.set(&id, "paused", None); count += 1; } else if RUNNING.contains(&state.as_str()) && state != "finalizing" { self.stop(&id, "pause"); count += 1; } } self.persist(true, false); json!({ "paused": count }) } pub fn resume(self: &Arc, owner: &str, ids: &Value, options: &Value) -> Value { let mut count = 0; for id in self.select(owner, ids) { let resumable = { let inner = self.inner.lock(); inner.jobs.get(&id).is_some_and(|j| ["paused", "interrupted", "reconnecting"].contains(&j.state.as_str()) || (j.state == "failed" && j.resumable)) }; if !resumable { continue; } if !self.bind_session(&id, options) { continue; } self.authorize(&id, options); { let mut inner = self.inner.lock(); if let Some(job) = inner.jobs.get_mut(&id) { if options["overwrite"] == true && job.error.as_ref().is_some_and(|e| e.code == "FILE_EXISTS" || e.code == "FILE_CONFLICT") { job.overwrite = true; job.target_baseline = None; job.finalize = json!({ "state": "pending" }); } job.attempts = 0; job.auth_required = false; } } self.clear_retry(&id); self.set(&id, "waiting", None); count += 1; } self.persist(true, false); self.schedule(); json!({ "resumed": count }) } /// Restart from offset zero (used when the partial data cannot be safely continued). pub fn retry(self: &Arc, owner: &str, ids: &Value, options: &Value) -> Value { let mut count = 0; for id in self.select(owner, ids) { let local_part = { let inner = self.inner.lock(); match inner.jobs.get(&id) { Some(j) if ["failed", "interrupted", "paused", "cancelled"].contains(&j.state.as_str()) => Some(j.local_part.clone()), _ => None, } }; let Some(local_part) = local_part else { continue }; if !self.bind_session(&id, options) { continue; } // The deterministic partial file (same job id) is truncated and rewritten from offset zero. if let Some(part) = local_part { let _ = std::fs::remove_file(part); } self.authorize(&id, options); { let mut inner = self.inner.lock(); if let Some(job) = inner.jobs.get_mut(&id) { job.transferred = 0; job.staged = None; job.target_baseline = None; job.hash = None; job.verification = None; job.finalize = json!({ "state": "pending" }); job.attempts = 0; job.auth_required = false; job.error = None; job.resumable = false; job.finalize_with_sudo = false; if job.direction == "upload" { if let Ok(meta) = std::fs::metadata(&job.local_path) { job.local_identity = Some(LocalIdentity::of(&meta)); job.size = meta.len(); } } else { job.remote_identity = None; } if options["overwrite"] == true { job.overwrite = true; } } } self.set(&id, "waiting", None); count += 1; } self.persist(true, false); self.schedule(); json!({ "retried": count }) } pub async fn cancel(self: &Arc, owner: &str, ids: &Value) -> Value { let mut count = 0; for id in self.select(owner, ids) { let state = self.job_state(&id); if FINAL.contains(&state.as_str()) { continue; } count += 1; if RUNNING.contains(&state.as_str()) { if state == "finalizing" { continue; // The remote replacement must finish or fail on its own. } self.stop(&id, "cancel"); continue; } self.clear_retry(&id); self.finish_cancel(&id).await; } self.persist(true, false); json!({ "cancelled": count }) } pub fn remove(self: &Arc, owner: &str, ids: &Value) -> Value { let mut count = 0; { let mut inner = self.inner.lock(); let selected = Self::select_ids(&inner, owner, ids); for id in selected { let removable = inner.jobs.get(&id).is_some_and(|j| FINAL.contains(&j.state.as_str()) && !j.cleanup_pending); if !removable { continue; } inner.jobs.remove(&id); inner.secrets.remove(&id); inner.runtime.remove(&id); inner.changed.insert(id); count += 1; } inner.pending_owners.insert(owner.to_string()); inner.full_owners.insert(owner.to_string()); } self.persist(true, false); self.emit_soon(); json!({ "removed": count }) } pub fn shutdown(self: &Arc) { let ids: Vec<(String, String)> = { let inner = self.inner.lock(); if inner.closed { return; } inner.jobs.values().map(|j| (j.id.clone(), j.state.clone())).collect() }; for (id, state) in ids { self.clear_retry(&id); if RUNNING.contains(&state.as_str()) { self.stop(&id, "shutdown"); } } // Running jobs are recorded as interrupted so the next start offers to resume them. { let mut inner = self.inner.lock(); for job in inner.jobs.values_mut() { if RUNNING.contains(&job.state.as_str()) || job.state == "waiting" || job.state == "reconnecting" { if job.state == "finalizing" || job.finalize_state() == "running" { job.finalize["state"] = json!("unknown"); } job.state = "interrupted".into(); job.error = Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() }); } } } self.persist(true, true); let mut inner = self.inner.lock(); inner.closed = true; // Late completions of stopped jobs must not overwrite the saved state. inner.secrets.clear(); } // ---- planning helpers ---------------------------------------------------------------------- fn store_plan(&self, plan: Plan) -> CoreResult { if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES { return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries."))); } if plan.bytes > MAX_PLAN_BYTES { return Err(file_error("TRANSFER_TOO_LARGE", "The selection exceeds the transfer size limit.")); } let conflicts: Vec<&PlanFile> = plan.files.iter().filter(|f| f.conflict).collect(); let result = json!({ "canceled": false, "planId": plan.id, "direction": plan.direction, "files": plan.files.len(), "directories": plan.directories.len(), "bytes": plan.bytes, "mode": plan.mode, "requiresSudo": plan.mode == "privileged" && plan.privilege.as_ref().is_some_and(|p| p["mode"] != "root"), "sudoPasswordRequired": plan.mode == "privileged" && plan.privilege.as_ref().is_some_and(|p| p["mode"] == "sudo-password" && p["sudoCached"] != true), "conflicts": conflicts.iter().take(50).map(|f| if plan.direction != "download" { f.remote_path.clone() } else { f.local_path.clone() }).collect::>(), "conflictCount": conflicts.len(), "skipped": plan.skipped.iter().take(50).cloned().collect::>(), "skippedCount": plan.skipped.len(), "destination": if plan.direction != "download" { plan.directory.clone() } else { plan.local_directory.clone() } }); let mut inner = self.inner.lock(); let cutoff = now() - PLAN_TTL_MS; inner.plans.retain(|_, p| p.created_at >= cutoff); inner.plans.insert(plan.id.clone(), plan); Ok(result) } async fn remote_stat(&self, sftp: &Sftp, target: &str, options: &Value, session_id: &str) -> CoreResult { match sftp.stat(target).await { Ok(stat) => Ok(RemoteInfo { kind: if stat.is_dir() { "directory" } else { "file" }, size: stat.size, mtime: stat.mtime as u64, mode: stat.mode }), Err(error) => { let code = error.transfer_code(); if code != "FILE_PERMISSION_DENIED" || options["sudo"] != true { return Err(file_error(code, &format!("{target} is not accessible."))); } let value = self.files.helper(session_id, json!({ "operation": "stat", "path": target }), &Authorization::from_options(options).sudo()).await?; if value["exists"] != true { return Err(file_error("FILE_NOT_FOUND", &format!("{target} does not exist."))); } let mtime_ns = number(&value["baseline"]["mtimeNs"]).unwrap_or(0.0); Ok(RemoteInfo { kind: if value["type"] == "directory" { "directory" } else { "file" }, size: number(&value["baseline"]["size"]).unwrap_or(0.0) as u64, mtime: (mtime_ns / 1e9).trunc() as u64, mode: number(&value["baseline"]["mode"]).unwrap_or(0.0) as u32, }) } } } #[allow(clippy::too_many_arguments)] fn walk_remote<'a>(&'a self, sftp: &'a Sftp, session_id: &'a str, plan: &'a mut Plan, remote: &'a str, local: &'a str, options: &'a Value, top: bool, depth: u32) -> BoxFuture<'a, CoreResult<()>> { async move { if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES { return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries."))); } let info = if top { self.remote_stat(sftp, remote, options, session_id).await? } else { let stat = sftp.lstat(remote).await.map_err(|e| file_error(e.transfer_code(), &format!("{remote} is not accessible.")))?; if stat.is_symlink() { plan.skipped.push(json!({ "path": remote, "reason": "symlink" })); return Ok(()); } RemoteInfo { kind: if stat.is_dir() { "directory" } else if !stat.is_file() { "other" } else { "file" }, size: stat.size, mtime: stat.mtime as u64, mode: stat.mode } }; if info.kind == "other" { plan.skipped.push(json!({ "path": remote, "reason": "type" })); return Ok(()); } if info.kind != "directory" { add_download(plan, remote, local, &info); return Ok(()); } if depth > 64 { return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep.")); } plan.directories.push(local.to_string()); plan.source_directories.push(remote.to_string()); match sftp.read_dir(remote).await { Ok(entries) => { for entry in entries { if entry.name == "." || entry.name == ".." { continue; } let Ok(name) = safe_entry_name(&entry.name) else { plan.skipped.push(json!({ "path": posix::join(remote, &entry.name), "reason": "name" })); continue; }; let child_local = Path::new(local).join(&name).to_string_lossy().to_string(); self.walk_remote(sftp, session_id, plan, &posix::join(remote, &name), &child_local, options, false, depth + 1).await?; } } Err(error) => { if error.transfer_code() != "FILE_PERMISSION_DENIED" || options["sudo"] != true { return Err(file_error(error.transfer_code(), &format!("Directory {remote} cannot be listed."))); } let listed = self.files.helper(session_id, json!({ "operation": "listdir", "path": remote }), &Authorization::from_options(options).sudo()).await?; for entry in listed["entries"].as_array().cloned().unwrap_or_default() { let raw = entry["name"].as_str().unwrap_or("").to_string(); let Ok(name) = safe_entry_name(&raw) else { plan.skipped.push(json!({ "path": posix::join(remote, &raw), "reason": "name" })); continue; }; let child = posix::join(remote, &name); let child_local = Path::new(local).join(&name).to_string_lossy().to_string(); match entry["type"].as_str() { Some(kind @ ("symlink" | "other")) => plan.skipped.push(json!({ "path": child, "reason": kind })), Some("directory") => self.walk_remote(sftp, session_id, plan, &child, &child_local, options, true, depth + 1).await?, _ => add_download( plan, &child, &child_local, &RemoteInfo { kind: "file", size: number(&entry["size"]).unwrap_or(0.0) as u64, mtime: number(&entry["mtime"]).unwrap_or(0.0) as u64, mode: number(&entry["mode"]).unwrap_or(0.0) as u32 }, ), } } } } Ok(()) } .boxed() } async fn create_directories(&self, plan: &Plan, privileged: bool, password: &str) -> CoreResult<()> { if plan.direction == "download" { for directory in &plan.directories { std::fs::create_dir_all(directory).map_err(io_core)?; #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; let _ = std::fs::set_permissions(directory, std::fs::Permissions::from_mode(0o700)); } } return Ok(()); } if plan.directories.is_empty() { return Ok(()); } if privileged { let auth = Authorization { sudo: true, sudo_password: (!password.is_empty()).then(|| password.to_string()), ..Default::default() }; for directory in &plan.directories { if let Err(error) = self.files.helper(&plan.session_id, json!({ "operation": "mkdir", "path": directory, "parents": true }), &auth).await { if error.code != "FILE_EXISTS" { return Err(error); } } } return Ok(()); } let sftp = self.files.sftp(&plan.session_id).await?; for directory in &plan.directories { if let Err(error) = sftp.mkdir(directory, 0o755).await { if !sftp.stat(directory).await.map(|s| s.is_dir()).unwrap_or(false) { return Err(file_error(error.transfer_code(), &format!("Directory {directory} could not be created."))); } } } Ok(()) } // ---- scheduling ------------------------------------------------------------------------------ fn schedule(self: &Arc) { if self.scheduling.swap(true, Ordering::SeqCst) { return; } let manager = self.clone(); tokio::spawn(async move { tokio::task::yield_now().await; manager.scheduling.store(false, Ordering::SeqCst); let start: Vec = { let mut inner = manager.inner.lock(); if inner.closed { return; } let mut running = inner.runtime.values().filter(|r| r.running).count(); let mut waiting: Vec<(u64, String)> = inner.jobs.values().filter(|j| j.state == "waiting").map(|j| (j.order, j.id.clone())).collect(); waiting.sort(); let max = inner.settings.max_concurrent as usize; let mut start = Vec::new(); for (_, id) in waiting { if running >= max { break; } if inner.runtime.get(&id).is_some_and(|r| r.running) { continue; } running += 1; let transferred = inner.jobs[&id].transferred; inner.runtime.entry(id.clone()).or_insert_with(|| Runtime::new(transferred)).running = true; start.push(id); } start }; for id in start { let manager = manager.clone(); tokio::spawn(async move { manager.run(id).await }); } }); } async fn run(self: Arc, id: String) { let stop_rx = { let mut inner = self.inner.lock(); let transferred = inner.jobs.get(&id).map(|j| j.transferred).unwrap_or(0); let rt = inner.runtime.entry(id.clone()).or_insert_with(|| Runtime::new(transferred)); rt.running = true; rt.stop = watch::channel(String::new()).0; rt.generation += 1; rt.speed = 0.0; rt.samples = 0; rt.last_bytes = transferred; rt.last_time = now(); rt.progress_since_resume = 0; let rx = rt.stop.subscribe(); if let Some(job) = inner.jobs.get_mut(&id) { job.error = None; job.resumable = false; } rx }; self.set(&id, "connecting", None); let outcome: RunResult<()> = async { self.check_host_keys(&id)?; let session_id = self.session_for(&id)?; self.update(&id, |j| j.session_id = json!(session_id)); let direction = self.with_job(&id, |j| j.direction.clone()).unwrap_or_default(); if direction == "upload" { self.run_upload(&id, &session_id, stop_rx.clone()).await } else if direction == "relay" { self.run_relay(&id, &session_id, stop_rx.clone()).await } else { self.run_download(&id, &session_id, stop_rx.clone()).await } } .await; match outcome { Ok(()) => { self.finish_move(&id).await; if let Some(batch) = self.with_job(&id, |j| j.batch_id.clone()) { self.finish_batch_move(&batch).await; } } Err(error) => self.failure(&id, error, &stop_rx).await, } if let Some(rt) = self.inner.lock().runtime.get_mut(&id) { rt.running = false; } self.persist(true, false); self.schedule(); } async fn failure(self: &Arc, id: &str, error: RunError, stop: &watch::Receiver) { let requested = stop.borrow().clone(); let error = match error { RunError::Stop(kind) if kind != "lost" => Err(kind), RunError::Stop(_) => Ok(CoreError::new("SESSION_NOT_CONNECTED", "The SSH connection was lost.")), RunError::Error(_) if !requested.is_empty() && requested != "lost" => Err(requested), RunError::Error(e) => Ok(e), }; let error = match error { Err(kind) => { match kind.as_str() { "cancel" => self.finish_cancel(id).await, "pause" => self.set(id, "paused", None), _ => self.set(id, "interrupted", Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() })), } return; } Ok(error) => error, }; let code = error_code(&error); let message: String = Regex::strip_code(&error.message).chars().take(400).collect(); let (state, transferred, size, direction) = self.with_job(id, |j| (j.state.clone(), j.transferred, j.size, j.direction.clone())).unwrap_or_default(); self.logger.warn("Transfer interrupted.", json!({ "jobId": id, "direction": direction, "state": state, "code": code, "transferred": transferred, "size": size })); if state == "finalizing" && (code == "FILE_RESULT_UNKNOWN" || CONNECTION_CODES.contains(&code.as_str())) { self.update(id, |j| j.finalize["state"] = json!("unknown")); } if AUTH_CODES.contains(&code.as_str()) { self.update(id, |j| j.auth_required = true); self.set(id, "interrupted", Some(JobError { code, message })); return; } if CONNECTION_CODES.contains(&code.as_str()) || code == "FILE_RESULT_UNKNOWN" { self.set(id, "interrupted", Some(JobError { code, message })); let auto = self.inner.lock().settings.auto_retry; let retries = self.with_job(id, |j| j.total_retries).unwrap_or(0); if auto && retries < MAX_TOTAL_RETRIES { self.schedule_retry(id); } return; } let resumable = ["FILE_NO_SPACE", "FILE_EXISTS", "FILE_CONFLICT", "LOCAL_NO_SPACE", "ENOSPC", "FILE_READ_ONLY_FILESYSTEM", "FILE_BUSY"].contains(&code.as_str()); self.update(id, |j| j.resumable = resumable); self.set(id, "failed", Some(JobError { code: if code == "ENOSPC" { "LOCAL_NO_SPACE".into() } else { code }, message })); } fn schedule_retry(self: &Arc, id: &str) { self.clear_retry(id); let delay = { let mut inner = self.inner.lock(); let Some(job) = inner.jobs.get_mut(id) else { return }; if job.attempts as usize >= RETRY_DELAYS_MS.len() { return; // Manual resume from here on: no endless loop. } let delay = RETRY_DELAYS_MS[job.attempts as usize]; job.attempts += 1; job.total_retries += 1; job.next_retry_at = Some(now() + delay as i64); delay }; let error = self.with_job(id, |j| j.error.clone()).flatten(); self.set(id, "reconnecting", error.clone()); let manager = self.clone(); let job_id = id.to_string(); let task = tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(delay)).await; if let Some(rt) = manager.inner.lock().runtime.get_mut(&job_id) { rt.retry_task = None; } if manager.job_state(&job_id) != "reconnecting" { return; } let attempts = manager.with_job(&job_id, |j| j.attempts).unwrap_or(0) as usize; if manager.find_session(&job_id).is_some() { manager.set(&job_id, "waiting", None); manager.schedule(); } else if attempts < RETRY_DELAYS_MS.len() { manager.schedule_retry(&job_id); } else { let error = manager.with_job(&job_id, |j| j.error.clone()).flatten(); manager.set(&job_id, "interrupted", error); } }); let mut inner = self.inner.lock(); let transferred = inner.jobs.get(id).map(|j| j.transferred).unwrap_or(0); inner.runtime.entry(id.to_string()).or_insert_with(|| Runtime::new(transferred)).retry_task = Some(task); } fn clear_retry(&self, id: &str) { let mut inner = self.inner.lock(); if let Some(task) = inner.runtime.get_mut(id).and_then(|r| r.retry_task.take()) { task.abort(); } if let Some(job) = inner.jobs.get_mut(id) { job.next_retry_at = None; } } fn session_state(self: &Arc, status: &Value) { if status["state"] == "disconnected" { // A dropped SSH connection does not answer outstanding SFTP requests: end those jobs now. let session = status["sessionId"].clone(); let ids: Vec = { let inner = self.inner.lock(); inner.jobs.values().filter(|j| (j.session_id == session || j.extra.get("relaySource").is_some_and(|r| r["sessionId"] == session)) && inner.runtime.get(&j.id).is_some_and(|r| r.running)).map(|j| j.id.clone()).collect() }; for id in ids { self.stop(&id, "lost"); } return; } if status["state"] != "connected" { return; } let identity = SessionIdentity::of_status(status).key(); let candidates: Vec = { let inner = self.inner.lock(); inner .jobs .values() .filter(|j| j.identity.key() == identity || relay_source(j).is_some_and(|(_, source)| source.key() == identity)) .filter(|j| { j.state == "reconnecting" || (j.state == "interrupted" && inner.settings.auto_retry && j.error.as_ref().is_some_and(|e| CONNECTION_CODES.contains(&e.code.as_str())) && j.total_retries < MAX_TOTAL_RETRIES) }) .map(|j| j.id.clone()) .collect() }; for id in candidates { self.clear_retry(&id); self.set(&id, "waiting", None); } self.inner.lock().cleaned_identities.remove(&identity); self.cleanup_orphans_later(); self.schedule(); } fn stop(&self, id: &str, kind: &str) { let inner = self.inner.lock(); if let Some(rt) = inner.runtime.get(id) { let _ = rt.stop.send(kind.to_string()); } } // ---- transfer engine ----------------------------------------------------------------------- fn helper_auth(&self, id: &str, privileged_override: bool, timeout: Option) -> Authorization { let inner = self.inner.lock(); let privileged = privileged_override || inner.jobs.get(id).is_some_and(|j| j.mode == "privileged"); let mut auth = if privileged { Authorization { sudo: true, sudo_password: inner.secrets.get(id).map(|s| s.to_string()).filter(|s| !s.is_empty()), ..Default::default() } } else { Authorization::default() }; auth.timeout_ms = timeout; auth } async fn open_channel(&self, session_id: &str, stop: &watch::Receiver) -> RunResult> { let sftp = self.files.sftp(session_id).await?; let kind = stop.borrow().clone(); if !kind.is_empty() { return Err(RunError::Stop(kind)); } Ok(sftp) } async fn run_upload(self: &Arc, id: &str, session_id: &str, stop: watch::Receiver) -> RunResult<()> { let job = self.snapshot(id)?; let meta = std::fs::symlink_metadata(&job.local_path).map_err(|e| file_error(if e.kind() == std::io::ErrorKind::NotFound { "LOCAL_NOT_FOUND" } else { "LOCAL_READ_FAILED" }, &format!("{} cannot be read.", job.local_path)))?; if !meta.is_file() { return Err(file_error("LOCAL_NOT_FILE", &format!("{} is no longer a regular file.", job.local_path)).into()); } let current = LocalIdentity::of(&meta); if !job.local_identity.as_ref().is_some_and(|i| i.same(¤t)) { if job.transferred > 0 { return Err(file_error("FILE_SOURCE_CHANGED", "The local file changed since the transfer started; restart it from the beginning.").into()); } self.update(id, |j| { j.local_identity = Some(current.clone()); j.size = current.size; }); } let job = self.snapshot(id)?; if job.finalize_state() == "unknown" || (job.hash.is_some() && job.transferred == job.size && job.part_path.is_some()) { return self.finalize_upload(id, session_id, &stop).await; } self.prepare_upload_target(id, session_id).await?; let sftp = self.open_channel(session_id, &stop).await?; let local = LocalFile::open(Path::new(&job.local_path), "r").map_err(io_core)?; let handle = self.open_remote_part(&sftp, id).await?; let result: RunResult<()> = async { let job = self.snapshot(id)?; if job.transferred > 0 { let remote = sftp.fstat(&handle).await.map_err(sftp_core)?; let transferred = job.transferred.min(remote.size); self.update(id, |j| j.transferred = transferred); let window = TAIL_CHECK.min(transferred); let position = transferred - window; verify_tail( transferred, async { local.read_at(position, window as usize).await.map_err(io_core) }.boxed(), async { sftp.read_full(&handle, position, window as usize).await.map_err(sftp_core) }.boxed(), window, ) .await?; } self.set(id, "transferring", None); let size = self.with_job(id, |j| j.size).unwrap_or(0); let produce: Producer = { let local = local.clone(); Arc::new(move |pos, len| { let local = local.clone(); Box::pin(async move { let data = local.read_at(pos, len).await.map_err(io_core)?; if data.len() != len { return Err(file_error("FILE_SOURCE_CHANGED", "The local file shrank during the transfer.")); } Ok(data) }) }) }; let consume: Consumer = { let (sftp, handle) = (sftp.clone(), handle.clone()); Arc::new(move |buf, pos| { let (sftp, handle) = (sftp.clone(), handle.clone()); Box::pin(async move { sftp.write_at(&handle, pos, &buf).await.map_err(sftp_core) }) }) }; self.run_pump(id, size, produce, consume, stop.clone()).await?; let remote = sftp.fstat(&handle).await.map_err(sftp_core)?; if remote.size > size { sftp.truncate(&handle, size).await.map_err(sftp_core)?; } let _ = sftp.fsync(&handle).await; let _ = sftp.close_file(&handle).await; let after = local.metadata().map(|m| LocalIdentity::of(&m)).map_err(io_core)?; if !self.with_job(id, |j| j.local_identity.as_ref().is_some_and(|i| i.same(&after))).unwrap_or(false) { return Err(file_error("FILE_SOURCE_CHANGED", "The local file changed during the transfer.").into()); } self.set(id, "verifying", None); let hash = local_hash(Path::new(&job.local_path), stop.clone()).await?; self.update(id, |j| { j.hash = Some(hash.clone()); j.verification = Some(json!({ "method": "sha256", "local": hash })); }); Ok(()) } .await; if result.is_err() { let _ = tokio::time::timeout(Duration::from_secs(3), sftp.close_file(&handle)).await; } drop(sftp); result?; self.finalize_upload(id, session_id, &stop).await } /// Like an upload, with an SFTP file as the source: same part file, publish step and resume rules. async fn run_relay(self: &Arc, id: &str, session_id: &str, stop: watch::Receiver) -> RunResult<()> { let job = self.snapshot(id)?; if job.finalize_state() == "unknown" || (job.hash.is_some() && job.transferred == job.size && job.part_path.is_some()) { return self.finalize_upload(id, session_id, &stop).await; } let source_session = self.relay_session(id)?; let source = self.open_channel(&source_session, &stop).await?; let stat = source.stat(&job.local_path).await.map_err(|e| file_error(e.transfer_code(), &format!("{} is not accessible.", job.local_path)))?; if !stat.is_file() { return Err(file_error("FILE_UNSAFE_TYPE_OR_LINK", &format!("{} is no longer a regular file.", job.local_path)).into()); } let current = RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 }; if job.remote_identity.as_ref() != Some(¤t) { if job.transferred > 0 { return Err(file_error("FILE_SOURCE_CHANGED", "The source file changed since the transfer started; restart it from the beginning.").into()); } self.update(id, |j| { j.remote_identity = Some(current.clone()); j.size = current.size; }); } self.prepare_upload_target(id, session_id).await?; let destination = self.open_channel(session_id, &stop).await?; let input = source.open_file(&job.local_path, "r", None).await.map_err(|e| file_error(e.transfer_code(), &format!("{} cannot be read.", job.local_path)))?; let output = match self.open_remote_part(&destination, id).await { Ok(handle) => handle, Err(error) => { let _ = tokio::time::timeout(Duration::from_secs(3), source.close_file(&input)).await; return Err(error); } }; let result: RunResult<()> = async { let job = self.snapshot(id)?; if job.transferred > 0 { let written = destination.fstat(&output).await.map_err(sftp_core)?; let transferred = job.transferred.min(written.size); self.update(id, |j| j.transferred = transferred); let window = TAIL_CHECK.min(transferred); let position = transferred - window; verify_tail( transferred, async { source.read_full(&input, position, window as usize).await.map_err(sftp_core) }.boxed(), async { destination.read_full(&output, position, window as usize).await.map_err(sftp_core) }.boxed(), window, ) .await?; } self.set(id, "transferring", None); let size = self.with_job(id, |j| j.size).unwrap_or(0); let produce: Producer = { let (source, input) = (source.clone(), input.clone()); Arc::new(move |pos, len| { let (source, input) = (source.clone(), input.clone()); Box::pin(async move { let data = source.read_full(&input, pos, len).await.map_err(sftp_core)?; if data.len() != len { return Err(file_error("FILE_SOURCE_CHANGED", "The source file shrank during the transfer.")); } Ok(data) }) }) }; let consume: Consumer = { let (destination, output) = (destination.clone(), output.clone()); Arc::new(move |buf, pos| { let (destination, output) = (destination.clone(), output.clone()); Box::pin(async move { destination.write_at(&output, pos, &buf).await.map_err(sftp_core) }) }) }; self.run_pump(id, size, produce, consume, stop.clone()).await?; let written = destination.fstat(&output).await.map_err(sftp_core)?; if written.size > size { destination.truncate(&output, size).await.map_err(sftp_core)?; } let _ = destination.fsync(&output).await; let _ = destination.close_file(&output).await; self.set(id, "verifying", None); let hash = self.relay_hash(&source_session, &source, &input, &job.local_path, size, &stop).await?; let after = source.stat(&job.local_path).await.map_err(sftp_core)?; if (RemoteIdentity { size: after.size, mtime: after.mtime as u64 }) != current { return Err(file_error("FILE_SOURCE_CHANGED", "The source file changed during the transfer.").into()); } self.update(id, |j| { j.hash = Some(hash.clone()); j.verification = Some(json!({ "method": "sha256", "local": hash })); }); Ok(()) } .await; let _ = tokio::time::timeout(Duration::from_secs(3), source.close_file(&input)).await; if result.is_err() { let _ = tokio::time::timeout(Duration::from_secs(3), destination.close_file(&output)).await; } drop((source, destination)); if let Err(RunError::Error(_)) = &result { // Either end may drop before its disconnect event arrives: that is a lost connection, not a broken file. if [source_session.as_str(), session_id].iter().any(|s| self.sessions.get(s).is_none_or(|r| *r.state.lock() != "connected")) { return Err(file_error("SESSION_NOT_CONNECTED", "A connection of this transfer ended.").into()); } } result?; self.finalize_upload(id, session_id, &stop).await } fn relay_session(&self, id: &str) -> RunResult { let job = self.snapshot(id)?; let (preferred, identity) = relay_source(&job).ok_or_else(|| RunError::Error(file_error("TRANSFER_INVALID_SELECTION", "The transfer has no source server.")))?; let session = matching_session(&self.sessions.list(), &identity, &preferred).ok_or_else(|| { RunError::Error(file_error("SESSION_NOT_CONNECTED", &format!("Connect to {}@{} to continue.", identity.username.as_str().unwrap_or(""), identity.host.as_str().unwrap_or("")))) })?; self.update(id, |j| { if let Some(relay) = j.extra.get_mut("relaySource") { relay["sessionId"] = json!(session); } }); Ok(session) } /// SHA-256 of the source: on the source server if it can, otherwise read over SFTP here. async fn relay_hash(&self, session: &str, sftp: &Sftp, handle: &FileHandle, path: &str, size: u64, stop: &watch::Receiver) -> RunResult { let auth = Authorization::default().with_timeout(hash_timeout(size)); if let Ok(value) = self.files.helper(session, json!({ "operation": "hash", "path": path }), &auth).await { if let Some(hash) = value["hash"].as_str() { return Ok(hash.to_string()); } } if let Ok(hash) = remote_checksum(&self.sessions, session, path).await { return Ok(hash); } let mut hasher = Sha256::new(); let mut position = 0; while position < size { let kind = stop.borrow().clone(); if !kind.is_empty() { return Err(RunError::Stop(kind)); } let length = (size - position).min(1024 * 1024) as usize; let data = sftp.read_full(handle, position, length).await.map_err(sftp_core)?; if data.len() != length { return Err(file_error("FILE_SOURCE_CHANGED", "The source file shrank during the transfer.").into()); } hasher.update(&data); position += length as u64; } Ok(hex::encode(hasher.finalize())) } /// A move removes its source only after a completed transfer whose SHA-256 matched and only if /// the source is still the file that was copied. In any other case the source stays. async fn finish_move(self: &Arc, id: &str) { let Ok(job) = self.snapshot(id) else { return }; if job.state != "completed" || job.extra.get("moveSource") != Some(&Value::Bool(true)) || job.extra.contains_key("moveResult") { return; } let verified = job.verification.as_ref().is_some_and(|v| v["local"].is_string() && v["local"] == v["remote"]); let result: Result<(), &'static str> = if verified { self.remove_source(&job).await } else { Err("unverified") }; if let Err(reason) = result { self.logger.warn("Moved file kept at its source.", json!({ "jobId": id, "reason": reason })); } self.update(id, |j| { j.extra.insert("moveResult".into(), match result { Ok(()) => json!({ "state": "deleted" }), Err(reason) => json!({ "state": "kept", "reason": reason }), }); }); self.mark_job(id); self.emit_soon(); self.persist(true, false); } async fn remove_source(self: &Arc, job: &Job) -> Result<(), &'static str> { let root = job.extra.get("moveRoot").and_then(Value::as_str).unwrap_or("").to_string(); if job.direction == "upload" { let path = Path::new(&job.local_path); let meta = std::fs::symlink_metadata(path).map_err(|_| "source-gone")?; if !meta.is_file() || !job.local_identity.as_ref().is_some_and(|i| i.same(&LocalIdentity::of(&meta))) { return Err("source-changed"); } std::fs::remove_file(path).map_err(|_| "remove-failed")?; let mut parent = path.parent(); while let Some(directory) = parent.filter(|d| !root.is_empty() && d.starts_with(&root) && *d != Path::new(&root)) { if std::fs::remove_dir(directory).is_err() { break; } parent = directory.parent(); } return Ok(()); } let (session, path) = if job.direction == "relay" { let (preferred, identity) = relay_source(job).ok_or("source-gone")?; (matching_session(&self.sessions.list(), &identity, &preferred).ok_or("source-disconnected")?, job.local_path.clone()) } else { (self.find_session(&job.id).ok_or("source-disconnected")?, job.remote_path.clone()) }; let sftp = self.files.sftp(&session).await.map_err(|_| "source-disconnected")?; let stat = sftp.lstat(&path).await.map_err(|_| "source-gone")?; if !stat.is_file() || job.remote_identity.as_ref() != Some(&RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 }) { return Err("source-changed"); } sftp.unlink(&path).await.map_err(|_| "remove-failed")?; let mut directory = posix::dirname(&path); while !root.is_empty() && directory != root && directory.starts_with(&format!("{}/", root.trim_end_matches('/'))) { if sftp.rmdir(&directory).await.is_err() { break; } directory = posix::dirname(&directory); } Ok(()) } /// After every file of a moved batch is gone from its source, removes the source folders, deepest /// first. Removing a folder fails while it holds anything, such as a skipped symlink or a kept file. async fn finish_batch_move(self: &Arc, batch: &str) { let pending = { let mut inner = self.inner.lock(); let done = inner.jobs.values().filter(|j| j.batch_id == batch).all(|j| j.state == "completed" && j.extra.get("moveResult").is_some_and(|r| r["state"] == "deleted")); if !done { return; } inner.batch_moves.remove(batch) }; let Some(mut pending) = pending else { return }; pending.directories.sort_by_key(|d| std::cmp::Reverse(d.len())); match pending.session { None => { for directory in &pending.directories { let _ = std::fs::remove_dir(directory); } } Some(session) => { if let Ok(sftp) = self.files.sftp(&session).await { for directory in &pending.directories { let _ = sftp.rmdir(directory).await; } } } } } /// A new connection is verified against the trusted key, but that key can have been replaced since /// the job started: then this is possibly another machine, and the job must not write or delete there. fn check_host_keys(&self, id: &str) -> RunResult<()> { let job = self.snapshot(id)?; let session = self.find_session(id); let source = relay_source(&job).and_then(|(preferred, identity)| matching_session(&self.sessions.list(), &identity, &preferred)); for (stored, current) in [("hostKey", session), ("sourceHostKey", source)] { let (Some(expected), Some(current)) = (job.extra.get(stored).and_then(Value::as_str), current) else { continue }; if self.sessions.trusted_host_key(¤t).as_deref() != Some(expected) { return Err(file_error("TRANSFER_HOST_KEY_CHANGED", "The server's host key changed after this transfer started. Start a new transfer if this is the same server.").into()); } } Ok(()) } async fn run_pump(self: &Arc, id: &str, size: u64, produce: Producer, consume: Consumer, stop: watch::Receiver) -> RunResult<()> { let start = self.with_job(id, |j| j.transferred).unwrap_or(0); let manager = self.clone(); let job_id = id.to_string(); let progress = move |confirmed: u64| manager.progress(&job_id, confirmed); match pump(start, size, produce, consume, stop, progress).await { Ok(_) => Ok(()), Err(PumpStop::Stopped(kind)) => Err(RunError::Stop(kind)), Err(PumpStop::Failed(error)) => Err(RunError::Error(error)), } } async fn prepare_upload_target(self: &Arc, id: &str, session_id: &str) -> RunResult<()> { let job = self.snapshot(id)?; if job.resolved_path.is_some() && job.part_path.is_some() && job.target_baseline.is_some() { return Ok(()); } let auth = self.helper_auth(id, false, None); let stat = match self.files.helper(session_id, json!({ "operation": "stat", "path": job.remote_path }), &auth).await { Ok(stat) => { self.update(id, |j| j.helper = Some(true)); stat } Err(error) => { if !NO_HELPER_CODES.contains(&error.code.as_str()) || job.mode == "privileged" { return Err(error.into()); } // No python3 / Windows server: plain SFTP is enough for new files; existing ones need // the metadata-preserving Linux helper. self.update(id, |j| j.helper = Some(false)); let sftp = self.files.sftp(session_id).await?; if sftp.exists(&job.remote_path).await { let message = if error.code == "FEATURE_PLATFORM_UNSUPPORTED" { "Replacing an existing file on this server is not supported (no metadata-preserving helper); remove it first or choose another name." } else { "Replacing an existing file requires python3 on the server to preserve its owner, permissions and ACLs." }; return Err(file_error(&error.code, message).into()); } json!({ "exists": false, "resolvedPath": job.remote_path }) } }; let exists = stat["exists"] == true; if exists && stat["type"] != "file" { return Err(file_error("FILE_UNSAFE_TYPE_OR_LINK", &format!("{} is not a regular file.", job.remote_path)).into()); } if exists && number(&stat["baseline"]["nlink"]).unwrap_or(1.0) > 1.0 { return Err(file_error("FILE_HARDLINK_UNSUPPORTED", &format!("{} has several hard links.", job.remote_path)).into()); } if exists && !job.overwrite { return Err(file_error("FILE_EXISTS", &format!("{} already exists.", job.remote_path)).into()); } let resolved = stat["resolvedPath"].as_str().filter(|s| !s.is_empty()).unwrap_or(&job.remote_path).to_string(); let baseline = if exists { stat["baseline"].clone() } else { Value::Null }; let helper = self.with_job(id, |j| j.helper == Some(true)).unwrap_or(false); let (owner_uid, part_path) = if job.mode == "privileged" { let workspace = self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?; (Some(workspace["uid"].clone()), posix::join(workspace["path"].as_str().unwrap_or(""), &format!("{}.part", job.id))) } else { let uid = if helper { Some(self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?["uid"].clone()) } else { None }; (uid, posix::join(&posix::dirname(&resolved), &part_name(&posix::basename(&resolved), &job.id))) }; self.update(id, |j| { j.resolved_path = Some(resolved.clone()); j.target_baseline = Some(baseline.clone()); if owner_uid.is_some() { j.owner_uid = owner_uid.clone(); } j.part_path = Some(part_path.clone()); }); self.persist(true, false); Ok(()) } async fn open_remote_part(&self, sftp: &Sftp, id: &str) -> RunResult { let job = self.snapshot(id)?; let part = job.part_path.clone().unwrap_or_default(); if job.transferred > 0 { match sftp.lstat(&part).await { Ok(stat) if stat.is_file() => return sftp.open_file(&part, "r+", None).await.map_err(|e| sftp_core(e).into()), _ => self.update(id, |j| j.transferred = 0), // The partial file disappeared; start this file again. } } match sftp.open_file(&part, "wx", Some(0o600)).await { Ok(handle) => Ok(handle), Err(error) => { // Only our own leftover (same job id) can exist here; remove the entry itself, never follow it. if error.transfer_code() == "FILE_PERMISSION_DENIED" { return Err(file_error("FILE_PERMISSION_DENIED", &format!("{} is not writable.", posix::dirname(&part))).into()); } let _ = sftp.unlink(&part).await; sftp.open_file(&part, "wx", Some(0o600)).await.map_err(|e| sftp_core(e).into()) } } } async fn finalize_upload(self: &Arc, id: &str, session_id: &str, stop: &watch::Receiver) -> RunResult<()> { let kind = stop.borrow().clone(); if !kind.is_empty() { return Err(RunError::Stop(kind)); } self.set(id, "finalizing", None); let retry = self.with_job(id, |j| ["unknown", "running"].contains(&j.finalize_state())).unwrap_or(false); self.update(id, |j| j.finalize = json!({ "state": "running", "startedAt": now() })); self.persist(true, false); let job = self.snapshot(id)?; if job.helper == Some(false) { return self.publish_without_helper(id, session_id).await; } let auth = self.helper_auth(id, job.finalize_with_sudo, Some(hash_timeout(job.size))); let request = json!({ "operation": "publish", "source": job.part_path, "uid": job.owner_uid, "path": job.remote_path, "hash": job.hash, "size": job.size, "baseline": job.target_baseline.clone().unwrap_or(Value::Null), "overwrite": job.overwrite, "strategy": if job.mode == "privileged" { "copy" } else { "rename" }, "parents": true, "retry": retry, "mode": 0o644 }); let result = self.files.helper(session_id, request, &auth).await?; self.update(id, |j| { j.finalize = json!({ "state": "done", "resolvedPath": result["resolvedPath"], "created": result["created"] == true, "metadataPreserved": result["metadataPreserved"] == true, "alreadyPublished": result["alreadyPublished"] == true, "finishedAt": now() }); let mut verification = j.verification.clone().unwrap_or_else(|| json!({})); verification["remote"] = result["hash"].clone(); verification["method"] = json!("sha256"); j.verification = Some(verification); }); if result["alreadyPublished"] == true && job.mode == "privileged" { let _ = self.files.helper(session_id, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await; } self.inner.lock().secrets.remove(id); self.set(id, "completed", None); self.logger.info("Transfer completed.", json!({ "jobId": id, "direction": "upload", "bytes": job.size, "mode": job.mode, "path": job.remote_path })); Ok(()) } async fn publish_without_helper(self: &Arc, id: &str, session_id: &str) -> RunResult<()> { let job = self.snapshot(id)?; let part = job.part_path.clone().unwrap_or_default(); let remote_hash = remote_checksum(&self.sessions, session_id, &part).await.ok(); if remote_hash.as_ref().is_some_and(|h| Some(h) != job.hash.as_ref()) { let _ = self.discard_partial(id).await; self.update(id, |j| j.transferred = 0); return Err(file_error("FILE_VERIFY_FAILED", "The uploaded data does not match the local file.").into()); } self.update(id, |j| { let mut verification = j.verification.clone().unwrap_or_else(|| json!({})); verification["remote"] = json!(remote_hash); verification["method"] = json!(if remote_hash.is_some() { "sha256" } else { "size" }); j.verification = Some(verification); }); let sftp = self.files.sftp(session_id).await?; let stat = sftp.stat(&part).await.map_err(sftp_core)?; if stat.size != job.size { return Err(file_error("FILE_VERIFY_FAILED", "The uploaded size does not match the local file.").into()); } if sftp.exists(&job.remote_path).await { return Err(file_error("FILE_EXISTS", &format!("{} appeared during the transfer.", job.remote_path)).into()); } sftp.rename(&part, &job.remote_path).await.map_err(sftp_core)?; self.update(id, |j| j.finalize = json!({ "state": "done", "resolvedPath": j.remote_path, "created": true, "metadataPreserved": false, "finishedAt": now() })); self.set(id, "completed", None); Ok(()) } async fn run_download(self: &Arc, id: &str, session_id: &str, stop: watch::Receiver) -> RunResult<()> { let sftp = self.open_channel(session_id, &stop).await?; let job = self.snapshot(id)?; let mut source = job.remote_path.clone(); if job.mode == "privileged" { source = self.ensure_staged(id, &sftp, session_id).await?; } else { let stat = sftp.stat(&job.remote_path).await.map_err(|e| file_error(e.transfer_code(), &format!("{} is not accessible.", job.remote_path)))?; let current = RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 }; if job.remote_identity.as_ref().is_some_and(|r| *r != current) && job.transferred > 0 { return Err(file_error("FILE_SOURCE_CHANGED", "The remote file changed since the transfer started; restart it from the beginning.").into()); } self.update(id, |j| { j.remote_identity = Some(current.clone()); j.size = current.size; j.remote_mode = Some(json!(stat.mode & 0o777)); }); } let handle = match sftp.open_file(&source, "r", None).await { Ok(handle) => handle, Err(error) if error.transfer_code() == "FILE_PERMISSION_DENIED" && self.with_job(id, |j| j.mode == "privileged-allowed").unwrap_or(false) => { self.update(id, |j| j.mode = "privileged".into()); source = self.ensure_staged(id, &sftp, session_id).await?; sftp.open_file(&source, "r", None).await.map_err(sftp_core)? } Err(error) => return Err(file_error(error.transfer_code(), &format!("{} cannot be read.", job.remote_path)).into()), }; let local_part = { let job = self.snapshot(id)?; job.local_part.clone().unwrap_or_else(|| { let local = Path::new(&job.local_path); local.with_file_name(part_name(&local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(), &job.id)).to_string_lossy().to_string() }) }; self.update(id, |j| j.local_part = Some(local_part.clone())); let result: RunResult<()> = async { let local = self.open_local_part(id, &local_part)?; let transferred = self.with_job(id, |j| j.transferred).unwrap_or(0); if transferred > 0 { let window = TAIL_CHECK.min(transferred); let position = transferred - window; verify_tail( transferred, async { local.read_at(position, window as usize).await.map_err(io_core) }.boxed(), async { sftp.read_full(&handle, position, window as usize).await.map_err(sftp_core) }.boxed(), window, ) .await?; } self.set(id, "transferring", None); let size = self.with_job(id, |j| j.size).unwrap_or(0); let produce: Producer = { let (sftp, handle) = (sftp.clone(), handle.clone()); Arc::new(move |pos, len| { let (sftp, handle) = (sftp.clone(), handle.clone()); Box::pin(async move { let data = sftp.read_full(&handle, pos, len).await.map_err(sftp_core)?; if data.len() != len { return Err(file_error("FILE_SOURCE_CHANGED", "The remote file shrank during the transfer.")); } Ok(data) }) }) }; let consume: Consumer = { let local = local.clone(); Arc::new(move |buf, pos| { let local = local.clone(); Box::pin(async move { local.write_at(pos, buf).await.map_err(io_core) }) }) }; self.run_pump(id, size, produce, consume, stop.clone()).await?; local.finish(size).await.map_err(io_core)?; drop(local); let _ = sftp.close_file(&handle).await; self.set(id, "verifying", None); let hash = local_hash(Path::new(&local_part), stop.clone()).await?; let job = self.snapshot(id)?; let mut verification = json!({ "method": "size", "local": hash }); if job.mode == "privileged" { verification = json!({ "method": "sha256", "local": hash, "remote": job.staged.as_ref().map(|s| s["hash"].clone()).unwrap_or(Value::Null) }); } else { let stat = sftp.stat(&job.remote_path).await.map_err(sftp_core)?; let identity = job.remote_identity.clone().unwrap_or_default(); if stat.size != identity.size || stat.mtime as u64 != identity.mtime { return Err(file_error("FILE_SOURCE_CHANGED", "The remote file changed during the transfer; restart it from the beginning.").into()); } if self.inner.lock().settings.verify { let remote = match self.files.helper(session_id, json!({ "operation": "hash", "path": job.remote_path }), &Authorization::default().with_timeout(hash_timeout(job.size))).await { Ok(value) => value["hash"].as_str().map(str::to_string), Err(_) => remote_checksum(&self.sessions, session_id, &job.remote_path).await.ok(), }; if let Some(remote) = remote { verification = json!({ "method": "sha256", "local": hash, "remote": remote }); } } } self.update(id, |j| { j.hash = Some(hash.clone()); j.verification = Some(verification.clone()); }); if verification["remote"].as_str().is_some_and(|r| r != hash) { let _ = self.discard_partial(id).await; self.update(id, |j| j.transferred = 0); return Err(file_error("FILE_VERIFY_FAILED", "The downloaded data does not match the remote file.").into()); } Ok(()) } .await; if result.is_err() { let _ = tokio::time::timeout(Duration::from_secs(3), sftp.close_file(&handle)).await; } drop(sftp); result?; let kind = stop.borrow().clone(); if !kind.is_empty() { return Err(RunError::Stop(kind)); } self.set(id, "finalizing", None); let job = self.snapshot(id)?; if std::fs::symlink_metadata(&job.local_path).is_ok() && !job.overwrite { return Err(file_error("FILE_EXISTS", &format!("{} already exists.", job.local_path)).into()); } #[cfg(unix)] { use std::os::unix::fs::PermissionsExt; let remote_mode = job.remote_mode.as_ref().and_then(|m| number(m)).map(|m| m as u32).unwrap_or(0o644); let mode = if job.mode == "privileged" { 0o600 } else { (remote_mode & 0o755) | 0o600 }; let _ = std::fs::set_permissions(&local_part, std::fs::Permissions::from_mode(mode)); } std::fs::rename(&local_part, &job.local_path).map_err(io_core)?; if job.mode == "privileged" && self.files.helper(session_id, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await.is_err() { self.update(id, |j| j.cleanup_pending = true); } self.update(id, |j| j.finalize = json!({ "state": "done", "finishedAt": now() })); self.inner.lock().secrets.remove(id); self.set(id, "completed", None); self.logger.info("Transfer completed.", json!({ "jobId": id, "direction": "download", "bytes": job.size, "mode": job.mode })); Ok(()) } async fn ensure_staged(self: &Arc, id: &str, sftp: &Sftp, session_id: &str) -> RunResult { let job = self.snapshot(id)?; if let Some(staged) = &job.staged { let path = staged["path"].as_str().unwrap_or(""); if sftp.stat(path).await.is_ok_and(|s| Some(s.size as f64) == number(&staged["size"])) { return Ok(path.to_string()); } self.update(id, |j| j.transferred = 0); } self.set(id, "connecting", None); let workspace = self.files.helper(session_id, json!({ "operation": "transfer-workspace" }), &Authorization::default()).await?; let auth = self.helper_auth(id, true, Some(hash_timeout(job.size))); let staged = self.files.helper(session_id, json!({ "operation": "stage-out", "path": job.remote_path, "id": job.id, "uid": workspace["uid"] }), &auth).await?; let path = staged["path"].as_str().unwrap_or("").to_string(); self.update(id, |j| { j.staged = Some(json!({ "path": staged["path"], "size": staged["size"], "hash": staged["hash"], "mtimeNs": staged["mtimeNs"] })); j.size = number(&staged["size"]).unwrap_or(0.0) as u64; j.transferred = 0; }); self.persist(true, false); Ok(path) } fn open_local_part(&self, id: &str, part: &str) -> RunResult { let transferred = self.with_job(id, |j| j.transferred).unwrap_or(0); if transferred > 0 { match std::fs::symlink_metadata(part) { Ok(meta) if meta.is_file() => { self.update(id, |j| j.transferred = j.transferred.min(meta.len())); return LocalFile::open(Path::new(part), "r+").map_err(|e| io_core(e).into()); } _ => self.update(id, |j| j.transferred = 0), } } let _ = std::fs::remove_file(part); LocalFile::open(Path::new(part), "wx").map_err(|e| io_core(e).into()) } // ---- cancel / cleanup ------------------------------------------------------------------------ async fn finish_cancel(self: &Arc, id: &str) { self.inner.lock().secrets.remove(id); let cleaned = self.discard_partial(id).await.is_ok(); self.update(id, |j| j.cleanup_pending = !cleaned); self.set(id, "cancelled", None); } /// Removes only temporary data this job created; existing destination files are never touched. async fn discard_partial(&self, id: &str) -> CoreResult<()> { let Some(job) = self.with_job(id, |j| j.clone()) else { return Ok(()) }; if job.direction == "download" { if let Some(part) = &job.local_part { match std::fs::remove_file(part) { Ok(()) => {} Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} Err(e) => return Err(io_core(e)), } } if job.staged.is_some() { let session = self.find_session(id).ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "Staged data is removed after reconnecting."))?; self.files.helper(&session, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await?; self.update(id, |j| j.staged = None); } return Ok(()); } let Some(part) = job.part_path.clone() else { return Ok(()) }; let session = self.find_session(id).ok_or_else(|| file_error("SESSION_NOT_CONNECTED", "Partial data is removed after reconnecting."))?; if job.mode == "privileged" { self.files.helper(&session, json!({ "operation": "transfer-remove", "id": job.id }), &Authorization::default()).await?; } else { let sftp = self.files.sftp(&session).await?; if let Ok(stat) = sftp.lstat(&part).await { if !stat.is_dir() { sftp.unlink(&part).await.map_err(sftp_core)?; } } } self.update(id, |j| j.part_path = None); Ok(()) } fn cleanup_orphans_later(self: &Arc) { if self.cleanup_scheduled.swap(true, Ordering::SeqCst) { return; } let manager = self.clone(); tokio::spawn(async move { tokio::task::yield_now().await; manager.cleanup_scheduled.store(false, Ordering::SeqCst); manager.cleanup_orphans().await; }); } /// After a restart: delete temporary transfer data that no resumable job owns any more. async fn cleanup_orphans(self: &Arc) { for session in self.sessions.list() { if session["state"] != "connected" { continue; } let session_id = session["sessionId"].as_str().unwrap_or("").to_string(); let identity = SessionIdentity::of_status(&session).key(); if !self.inner.lock().cleaned_identities.insert(identity.clone()) { continue; } let pending: Vec = self.inner.lock().jobs.values().filter(|j| j.identity.key() == identity && j.cleanup_pending).map(|j| j.id.clone()).collect(); for id in pending { if self.discard_partial(&id).await.is_ok() { self.update(&id, |j| j.cleanup_pending = false); self.mark_job(&id); } } let keep: HashSet = self.inner.lock().jobs.values().filter(|j| j.identity.key() == identity && !FINAL.contains(&j.state.as_str())).map(|j| j.id.clone()).collect(); let Ok(listed) = self.files.helper(&session_id, json!({ "operation": "transfer-list" }), &Authorization::default()).await else { continue }; for item in listed["items"].as_array().cloned().unwrap_or_default() { let item_id = item["id"].as_str().unwrap_or("").to_string(); let known = self.with_job(&item_id, |j| FINAL.contains(&j.state.as_str())); let abandoned = match known { Some(final_state) => final_state, None => now() - (number(&item["mtime"]).unwrap_or(0.0) * 1000.0) as i64 > ORPHAN_AGE_MS, }; if keep.contains(&item_id) || !abandoned { continue; } let _ = self.files.helper(&session_id, json!({ "operation": "transfer-remove", "id": item_id }), &Authorization::default()).await; self.logger.info("Abandoned transfer data removed.", json!({ "sessionId": session_id, "id": item_id, "bytes": item["size"] })); } self.persist(false, false); } } // ---- state, events, persistence ---------------------------------------------------------------- fn authorize(&self, id: &str, options: &Value) { let mut inner = self.inner.lock(); let password = options["sudoPassword"].as_str().filter(|s| !s.is_empty()).map(str::to_string); if let Some(job) = inner.jobs.get_mut(id) { if options["sudo"] == true || password.is_some() { // Data is already next to the target: only the metadata-preserving replacement needs root. if job.direction != "download" && job.mode == "direct" && job.transferred > 0 && job.transferred == job.size { job.finalize_with_sudo = true; } if job.direction == "download" && job.mode == "direct" { job.mode = "privileged-allowed".into(); } if job.direction != "download" && job.mode == "direct" && job.transferred == 0 && job.error.as_ref().is_some_and(|e| AUTH_CODES.contains(&e.code.as_str())) { job.mode = "privileged".into(); job.part_path = None; job.target_baseline = None; } } } if let Some(password) = password { inner.secrets.insert(id.to_string(), Zeroizing::new(password)); } } fn set(self: &Arc, id: &str, state: &str, error: Option) { if !STATES.contains(&state) { return; } { let mut inner = self.inner.lock(); let t = now(); let Inner { jobs, runtime, .. } = &mut *inner; let Some(job) = jobs.get_mut(id) else { return }; let rt = runtime.get_mut(id); if job.state == "transferring" { if let Some(rt) = rt.as_deref() { if let Some(since) = rt.active_since { job.active_ms += (t - since).max(0) as u64; } } } if let Some(rt) = rt { if job.state == "transferring" { rt.active_since = None; } if state == "transferring" { rt.active_since = Some(t); rt.last_time = t; rt.last_bytes = job.transferred; } } job.state = state.into(); job.error = error; job.updated_at = t; if state == "transferring" && job.started_at.is_none() { job.started_at = Some(t); } if FINAL.contains(&state) { job.finished_at = Some(t); } let owner = job.owner.clone(); Self::mark(&mut inner, id, &owner); } self.emit_soon(); self.persist(!RUNNING.contains(&state), false); } fn progress(self: &Arc, id: &str, confirmed: u64) { { let mut inner = self.inner.lock(); let t = now(); let Inner { jobs, runtime, .. } = &mut *inner; let Some(job) = jobs.get_mut(id) else { return }; let moved = confirmed.saturating_sub(job.transferred); job.transferred = confirmed; job.moved_bytes += moved; if let Some(rt) = runtime.get_mut(id) { rt.progress_since_resume += moved; if rt.progress_since_resume > 1024 * 1024 { job.attempts = 0; } let elapsed = t - rt.last_time; if elapsed >= 500 { let instant = (job.transferred.saturating_sub(rt.last_bytes)) as f64 * 1000.0 / elapsed as f64; rt.speed = if rt.samples > 0 { 0.3 * instant + 0.7 * rt.speed } else { instant }; rt.samples += 1; rt.last_time = t; rt.last_bytes = job.transferred; } } job.updated_at = t; let owner = job.owner.clone(); Self::mark(&mut inner, id, &owner); } self.emit_soon(); self.persist(false, false); } fn public(inner: &Inner, job: &Job) -> Value { let rt = inner.runtime.get(&job.id); let t = now(); let mut speed = 0.0; let mut eta = Value::Null; if job.state == "transferring" { if let Some(rt) = rt { let idle = t - rt.last_time; speed = if idle > 2000 { rt.speed * 2000.0 / idle as f64 } else { rt.speed }; if rt.samples >= 3 && speed > 1.0 { eta = json!((job.size.saturating_sub(job.transferred) as f64 / speed).ceil() as u64); } } } let active_ms = job.active_ms + if job.state == "transferring" { rt.and_then(|r| r.active_since).map(|s| (t - s).max(0) as u64).unwrap_or(0) } else { 0 }; let state = job.state.as_str(); let overwrite_error = job.error.as_ref().is_some_and(|e| e.code == "FILE_EXISTS" || e.code == "FILE_CONFLICT"); json!({ "id": job.id, "batchId": job.batch_id, "sessionId": job.session_id, "identity": job.identity, "direction": job.direction, "name": job.name, "state": state, "source": if job.direction != "download" { &job.local_path } else { &job.remote_path }, "target": if job.direction != "download" { &job.remote_path } else { &job.local_path }, "resolvedTarget": job.finalize.get("resolvedPath").filter(|v| v.is_string()).cloned().or_else(|| job.resolved_path.clone().map(Value::from)).unwrap_or(Value::Null), "mode": if job.mode == "privileged" { "privileged" } else { "direct" }, "size": job.size, "transferred": job.transferred, "percent": percent(job.transferred, job.size), "speed": speed.round() as u64, "averageSpeed": if active_ms > 1000 { json!((job.moved_bytes as f64 * 1000.0 / active_ms as f64).round() as u64) } else { Value::Null }, "eta": eta, "finalize": { "state": job.finalize_state(), "metadataPreserved": job.finalize["metadataPreserved"] == true, "created": job.finalize["created"] == true }, "verification": job.verification.as_ref().map(|v| json!({ "method": v["method"], "matched": v["remote"].is_string() && v["remote"] == v["local"] })), "error": job.error.as_ref().map(|e| json!({ "code": e.code, "message": e.message })), "authRequired": job.auth_required, "attempts": job.attempts, "nextRetryAt": job.next_retry_at, "cleanupPending": job.cleanup_pending, "sourceIdentity": job.extra.get("relaySource").map(|r| r["identity"].clone()).unwrap_or(Value::Null), "sourceSessionId": job.extra.get("relaySource").map(|r| r["sessionId"].clone()).unwrap_or(Value::Null), "move": job.extra.get("moveSource") == Some(&Value::Bool(true)), "moveResult": job.extra.get("moveResult").cloned().unwrap_or(Value::Null), "actions": { "pause": (["waiting", "reconnecting", "interrupted", "connecting", "transferring", "verifying"].contains(&state)), "resume": (["paused", "interrupted", "reconnecting"].contains(&state) || (state == "failed" && job.resumable)), "retry": (["failed", "interrupted", "paused", "cancelled"].contains(&state)), "cancel": (!FINAL.contains(&state) && state != "finalizing"), "overwrite": (overwrite_error && ["failed", "interrupted"].contains(&state)), "remove": (FINAL.contains(&state) && !job.cleanup_pending) }, "createdAt": job.created_at, "updatedAt": job.updated_at, "finishedAt": job.finished_at }) } fn mark(inner: &mut Inner, id: &str, owner: &str) { inner.changed.insert(id.to_string()); inner.pending_owners.insert(owner.to_string()); } fn mark_job(self: &Arc, id: &str) { { let mut inner = self.inner.lock(); let Some(owner) = inner.jobs.get(id).map(|j| j.owner.clone()) else { return }; Self::mark(&mut inner, id, &owner); } self.emit_soon(); } fn emit_soon(self: &Arc) { { let mut inner = self.inner.lock(); if inner.emit_scheduled { return; } inner.emit_scheduled = true; } let manager = self.clone(); tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(EMIT_MS)).await; let updates: Vec<(String, Value)> = { let mut inner = manager.inner.lock(); inner.emit_scheduled = false; let owners: Vec = inner.pending_owners.drain().collect(); let full: HashSet = inner.full_owners.drain().collect(); let changed: HashSet = inner.changed.drain().collect(); owners .into_iter() .map(|owner| { let all: Vec = Self::owned(&inner, &owner).iter().map(|j| Self::public(&inner, j)).collect(); let is_full = full.contains(&owner); let jobs: Vec = if is_full { all.clone() } else { all.iter().filter(|j| changed.contains(j["id"].as_str().unwrap_or("")) || RUNNING.contains(&j["state"].as_str().unwrap_or(""))).cloned().collect() }; let removed: Vec<&String> = changed.iter().filter(|id| !inner.jobs.contains_key(*id)).collect(); let update = json!({ "owner": owner, "jobs": jobs, "removed": removed, "full": is_full, "summary": summarize(&all), "settings": inner.settings }); (owner, update) }) .collect() }; for (owner, update) in updates { (manager.emitter)(&owner, update); } }); } fn owned<'a>(inner: &'a Inner, owner: &str) -> Vec<&'a Job> { let mut jobs: Vec<&Job> = inner.jobs.values().filter(|j| j.owner == owner).collect(); jobs.sort_by_key(|j| j.order); jobs } fn select_ids(inner: &Inner, owner: &str, ids: &Value) -> Vec { let owned = Self::owned(inner, owner); if *ids == "all" { return owned.iter().map(|j| j.id.clone()).collect(); } let wanted: HashSet = match ids { Value::Array(list) => list.iter().take(20000).map(|v| v.as_str().map(str::to_string).unwrap_or_else(|| v.to_string())).collect(), Value::Null => HashSet::new(), other => [other.as_str().map(str::to_string).unwrap_or_else(|| other.to_string())].into_iter().collect(), }; owned.iter().filter(|j| wanted.contains(&j.id)).map(|j| j.id.clone()).collect() } fn select(&self, owner: &str, ids: &Value) -> Vec { Self::select_ids(&self.inner.lock(), owner, ids) } /// Moves a server-to-server job to a new connection to its source server, on request only and /// only when the old connection is gone. fn bind_relay_source(&self, id: &str, options: &Value) -> bool { let Some(requested) = options["sourceSessionId"].as_str() else { return true }; let Ok(identity) = self.identity(requested) else { return false }; let Some(Some((previous, expected))) = self.with_job(id, relay_source) else { return false }; if identity != expected || (previous != requested && self.identity(&previous).is_ok()) { return false; } if let Some(relay) = self.inner.lock().jobs.get_mut(id).and_then(|job| job.extra.get_mut("relaySource")) { relay["sessionId"] = json!(requested); } true } fn bind_session(&self, id: &str, options: &Value) -> bool { if !self.bind_relay_source(id, options) { return false; } let Some(requested) = options["sessionId"].as_str() else { return true }; let Ok(identity) = self.identity(requested) else { return false }; let Some((previous, expected)) = self.with_job(id, |j| (j.session(), j.identity.clone())) else { return false }; if identity != expected || (previous != requested && self.identity(&previous).is_ok()) { return false; } if let Some(job) = self.inner.lock().jobs.get_mut(id) { job.session_id = json!(requested); } true } fn find_session(&self, id: &str) -> Option { let (identity, preferred) = self.with_job(id, |j| (j.identity.clone(), j.session()))?; matching_session(&self.sessions.list(), &identity, &preferred) } fn session_for(&self, id: &str) -> RunResult { self.find_session(id).ok_or_else(|| { let (user, host) = self.with_job(id, |j| (j.identity.username.as_str().unwrap_or("").to_string(), j.identity.host.as_str().unwrap_or("").to_string())).unwrap_or_default(); RunError::Error(file_error("SESSION_NOT_CONNECTED", &format!("Connect to {user}@{host} to continue."))) }) } fn trim_history(inner: &mut Inner) { let mut finished: Vec<(i64, String)> = inner.jobs.values().filter(|j| FINAL.contains(&j.state.as_str()) && !j.cleanup_pending).map(|j| (j.finished_at.unwrap_or(0), j.id.clone())).collect(); if finished.len() <= MAX_HISTORY { return; } finished.sort(); let excess = finished.len() - MAX_HISTORY; for (_, id) in finished.into_iter().take(excess) { inner.jobs.remove(&id); inner.changed.insert(id); } } fn with_job(&self, id: &str, f: impl FnOnce(&Job) -> T) -> Option { self.inner.lock().jobs.get(id).map(f) } fn snapshot(&self, id: &str) -> RunResult { self.with_job(id, |j| j.clone()).ok_or_else(|| RunError::Stop("cancel".into())) } fn update(&self, id: &str, f: impl FnOnce(&mut Job)) { if let Some(job) = self.inner.lock().jobs.get_mut(id) { f(job); } } fn job_state(&self, id: &str) -> String { self.with_job(id, |j| j.state.clone()).unwrap_or_default() } fn snapshot_file(&self) -> Option<(PathBuf, Value)> { let file = self.file.clone()?; let mut inner = self.inner.lock(); if inner.closed { return None; } Self::trim_history(&mut inner); // Persisted state never contains passwords, keys or runtime handles. let mut jobs: Vec<&Job> = inner.jobs.values().collect(); jobs.sort_by_key(|j| j.order); Some((file, json!({ "version": 1, "settings": inner.settings, "order": inner.order, "jobs": jobs }))) } fn persist(self: &Arc, soon: bool, sync: bool) { if self.file.is_none() { return; } let write = |manager: &TransferManager| { if let Some((file, data)) = manager.snapshot_file() { if let Err(error) = crate::core::storage::atomic_write_json(&file, &data) { manager.logger.warn("Transfer state could not be saved.", json!({ "message": error.message })); } } }; if sync { write(self); return; } if !soon && self.persist_pending.swap(true, Ordering::SeqCst) { return; } if soon { self.persist_pending.store(true, Ordering::SeqCst); } let manager = self.clone(); tokio::spawn(async move { tokio::time::sleep(Duration::from_millis(if soon { 50 } else { PERSIST_MS })).await; if !manager.persist_pending.swap(false, Ordering::SeqCst) { return; } let _ = tokio::task::spawn_blocking(move || write(&manager)).await; }); } fn load(&self) { let Some(file) = &self.file else { return }; let Ok(text) = std::fs::read_to_string(file) else { return }; let Ok(data) = serde_json::from_str::(&text) else { return }; if data["version"] != 1 || !data["jobs"].is_array() { return; } let mut inner = self.inner.lock(); if let Some(settings) = data["settings"].as_object() { if let Some(v) = settings.get("maxConcurrent").and_then(number) { inner.settings.max_concurrent = if v.fract() == 0.0 && (1.0..=8.0).contains(&v) { v as u32 } else { Settings::default().max_concurrent }; } if let Some(v) = settings.get("autoRetry").and_then(Value::as_bool) { inner.settings.auto_retry = v; } if let Some(v) = settings.get("verify").and_then(Value::as_bool) { inner.settings.verify = v; } } inner.order = number(&data["order"]).unwrap_or(0.0) as u64; let jobs = data["jobs"].as_array().cloned().unwrap_or_default(); let skip = jobs.len().saturating_sub(MAX_JOBS); for raw in jobs.into_iter().skip(skip) { let Ok(mut job) = serde_json::from_value::(raw) else { continue }; if job.id.len() != 32 || !job.id.chars().all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c)) || !STATES.contains(&job.state.as_str()) { continue; } // Nothing restarts on its own after the application was closed: the user is offered to resume. if RUNNING.contains(&job.state.as_str()) || job.state == "waiting" || job.state == "reconnecting" { if job.state == "finalizing" || job.finalize_state() == "running" { if !job.finalize.is_object() { job.finalize = json!({}); } job.finalize["state"] = json!("unknown"); } job.state = "interrupted".into(); job.error = Some(JobError { code: "APP_RESTARTED".into(), message: "The application was closed during the transfer.".into() }); } job.next_retry_at = None; job.attempts = 0; inner.order = inner.order.max(job.order); inner.jobs.insert(job.id.clone(), job); } } } struct Regex; impl Regex { /// `String(message).replace(/^[A-Z_]+: /, "")`. fn strip_code(message: &str) -> String { if let Some((prefix, rest)) = message.split_once(": ") { if !prefix.is_empty() && prefix.chars().all(|c| c.is_ascii_uppercase() || c == '_') { return rest.to_string(); } } message.to_string() } } #[derive(Debug, Clone)] struct RemoteInfo { kind: &'static str, size: u64, mtime: u64, mode: u32, } fn relay_source(job: &Job) -> Option<(String, SessionIdentity)> { let relay = job.extra.get("relaySource")?; Some((relay["sessionId"].as_str().unwrap_or("").to_string(), serde_json::from_value(relay["identity"].clone()).ok()?)) } /// Every name is checked: a hostile source server cannot place a file outside the destination. /// Symlinks and special files are skipped and listed, never followed. fn walk_relay<'a>(sftp: &'a Sftp, plan: &'a mut Plan, source: &'a str, destination: &'a str, depth: u32) -> BoxFuture<'a, CoreResult<()>> { async move { if plan.files.len() + plan.directories.len() > MAX_PLAN_ENTRIES { return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries."))); } let stat = sftp.lstat(source).await.map_err(|e| file_error(e.transfer_code(), &format!("{source} is not accessible.")))?; if stat.is_symlink() || (!stat.is_dir() && !stat.is_file()) { plan.skipped.push(json!({ "path": source, "reason": if stat.is_symlink() { "symlink" } else { "type" } })); return Ok(()); } if stat.is_file() { plan.bytes += stat.size; plan.files.push(PlanFile { name: posix::basename(destination), local_path: source.to_string(), remote_path: destination.to_string(), size: stat.size, local_identity: None, remote_identity: Some(RemoteIdentity { size: stat.size, mtime: stat.mtime as u64 }), mode: None, conflict: false, overwrite_confirmed: false, }); return Ok(()); } if depth > 64 { return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep.")); } plan.directories.push(destination.to_string()); plan.source_directories.push(source.to_string()); let entries = sftp.read_dir(source).await.map_err(|e| file_error(e.transfer_code(), &format!("{source} cannot be read.")))?; for entry in entries { if entry.name == "." || entry.name == ".." { continue; } let Ok(name) = safe_entry_name(&entry.name) else { plan.skipped.push(json!({ "path": posix::join(source, &entry.name), "reason": "name" })); continue; }; walk_relay(sftp, plan, &posix::join(source, &name), &posix::join(destination, &name), depth + 1).await?; } Ok(()) } .boxed() } fn add_download(plan: &mut Plan, remote: &str, local: &str, info: &RemoteInfo) { plan.bytes += info.size; plan.files.push(PlanFile { name: posix::basename(remote), local_path: local.to_string(), remote_path: remote.to_string(), size: info.size, local_identity: None, remote_identity: Some(RemoteIdentity { size: info.size, mtime: info.mtime }), mode: Some(info.mode & 0o777), conflict: false, overwrite_confirmed: false, }); } #[derive(Default)] struct LocalWalk { files: Vec, directories: Vec, source_directories: Vec, skipped: Vec, bytes: u64, } impl LocalWalk { fn visit(&mut self, local: &Path, remote: &str, depth: u32) -> CoreResult<()> { if self.files.len() + self.directories.len() > MAX_PLAN_ENTRIES { return Err(file_error("TRANSFER_TOO_MANY_ENTRIES", &format!("A transfer can contain at most {MAX_PLAN_ENTRIES} entries."))); } let meta = std::fs::symlink_metadata(local).map_err(io_core)?; let shown = local.to_string_lossy().to_string(); if meta.file_type().is_symlink() { self.skipped.push(json!({ "path": shown, "reason": "symlink" })); return Ok(()); } if meta.is_dir() { self.directories.push(remote.to_string()); self.source_directories.push(shown); if depth > 64 { return Err(file_error("TRANSFER_TOO_DEEP", "The selected directory tree is too deep.")); } for entry in std::fs::read_dir(local).map_err(io_core)? { let entry = entry.map_err(io_core)?; let raw = entry.file_name().to_string_lossy().to_string(); let Ok(name) = safe_entry_name(&raw) else { self.skipped.push(json!({ "path": local.join(&raw).to_string_lossy(), "reason": "name" })); continue; }; self.visit(&local.join(&name), &posix::join(remote, &name), depth + 1)?; } return Ok(()); } if !meta.is_file() { self.skipped.push(json!({ "path": shown, "reason": "type" })); return Ok(()); } self.bytes += meta.len(); self.files.push(PlanFile { name: local.file_name().map(|n| n.to_string_lossy().to_string()).unwrap_or_default(), local_path: shown, remote_path: remote.to_string(), size: meta.len(), local_identity: Some(LocalIdentity::of(&meta)), remote_identity: None, mode: None, conflict: false, overwrite_confirmed: false, }); Ok(()) } } /// Writable by the SSH account: direct SFTP; otherwise the privileged staging route. async fn directory_mode(sftp: &Sftp, directory: &str) -> CoreResult<&'static str> { let stat: Stat = sftp.stat(directory).await.map_err(|e| { let code = e.transfer_code(); file_error(if code == "FILE_NOT_FOUND" { "FILE_NOT_FOUND" } else { code }, &format!("{directory} is not accessible.")) })?; if !stat.is_dir() { return Err(file_error("FILE_NOT_DIRECTORY", &format!("{directory} is not a directory."))); } let probe = posix::join(directory, &format!(".chj-probe-{}", hex_id(6))); match sftp.open_file(&probe, "wx", Some(0o600)).await { Ok(handle) => { let _ = sftp.close_file(&handle).await; let _ = sftp.unlink(&probe).await; Ok("direct") } Err(error) if error.transfer_code() == "FILE_PERMISSION_DENIED" => Ok("privileged"), Err(error) => { let code = error.transfer_code(); Err(file_error(if code == "TRANSFER_FAILED" { "FILE_REMOTE_IO_FAILED" } else { code }, &format!("{directory} is not writable."))) } } } async fn remote_conflicts(sftp: &Arc, plan: &mut Plan) { let paths: Vec<(usize, String)> = plan.files.iter().enumerate().map(|(i, f)| (i, f.remote_path.clone())).collect(); let checks = futures::stream::iter(paths.into_iter().map(|(i, path)| { let sftp = sftp.clone(); async move { (i, sftp.lstat(&path).await.ok()) } })) .buffer_unordered(16) .collect::>() .await; let mut blocked = HashSet::new(); for (i, stat) in checks { let Some(stat) = stat else { continue }; if stat.is_dir() { blocked.insert(i); plan.skipped.push(json!({ "path": plan.files[i].remote_path, "reason": "directory-exists" })); } else { plan.files[i].conflict = true; } } let mut index = 0; plan.files.retain(|_| { let keep = !blocked.contains(&index); index += 1; keep }); } async fn local_hash(path: &Path, stop: watch::Receiver) -> RunResult { let path = path.to_path_buf(); let result = tokio::task::spawn_blocking(move || -> Result { use std::io::Read; let mut file = std::fs::File::open(&path).map_err(|e| RunError::Error(io_core(e)))?; let mut digest = Sha256::new(); let mut buffer = vec![0u8; 1024 * 1024]; loop { let kind = stop.borrow().clone(); if !kind.is_empty() { return Err(RunError::Stop(kind)); } let n = file.read(&mut buffer).map_err(|e| RunError::Error(io_core(e)))?; if n == 0 { break; } digest.update(&buffer[..n]); } Ok(hex::encode(digest.finalize())) }) .await .map_err(|_| RunError::Error(CoreError::plain("Local hashing failed.")))?; result } #[cfg(test)] mod tests { use super::*; #[test] fn transfers_never_choose_another_connection_to_the_same_server() { let a = json!({ "sessionId": "A", "state": "connected", "profileId": "p", "host": "h", "port": 22, "username": "u" }); let mut b = a.clone(); b["sessionId"] = json!("B"); let identity = SessionIdentity::of_status(&a); assert_eq!(matching_session(&[a.clone(), b.clone()], &identity, "A"), Some("A".into())); assert_eq!(matching_session(&[b.clone()], &identity, "A"), None); assert_eq!(matching_session(&[a.clone(), b.clone()], &identity, "B"), Some("B".into())); b["host"] = json!("another-host"); assert_eq!(matching_session(&[b], &identity, "B"), None); let mut disconnected = a; disconnected["state"] = json!("disconnected"); assert_eq!(matching_session(&[disconnected], &identity, "A"), None); } /// A `transfers.json` written by 0.1.0 loads, keeps unknown fields and comes back unchanged /// (except the restart transition 0.1.0 applies on load as well). #[test] fn loads_and_rewrites_0_1_0_state() { let legacy = json!({ "version": 1, "settings": { "maxConcurrent": 3, "autoRetry": false, "verify": true }, "order": 7, "jobs": [{ "id": "0123456789abcdef0123456789abcdef", "owner": "chj.file-manager", "batchId": "b", "direction": "upload", "identity": { "profileId": "p", "host": "h", "port": 22, "username": "u" }, "sessionId": "s1", "order": 7, "name": "a.bin", "localPath": "/tmp/a.bin", "remotePath": "/srv/a.bin", "size": 10, "transferred": 4, "movedBytes": 4, "activeMs": 12, "state": "transferring", "mode": "direct", "overwrite": false, "localIdentity": { "size": 10, "mtimeMs": 1700000000123.0, "ino": 42, "dev": 16777220 }, "remoteIdentity": null, "remoteMode": null, "finalize": { "state": "pending" }, "attempts": 2, "totalRetries": 3, "error": null, "createdAt": 1, "updatedAt": 2, "resolvedPath": "/srv/a.bin", "partPath": "/srv/.a.bin.chj-part-x", "targetBaseline": null, "helper": true, "ownerUid": 1000, "futureField": { "kept": true } }] }); let dir = tempfile::tempdir().unwrap(); std::fs::write(dir.path().join("transfers.json"), serde_json::to_string(&legacy).unwrap()).unwrap(); let job: Job = serde_json::from_value(legacy["jobs"][0].clone()).unwrap(); assert_eq!(job.target_baseline, Some(Value::Null)); assert_eq!(job.local_identity.as_ref().unwrap().mtime_ms, 1700000000123); let back = serde_json::to_value(&job).unwrap(); assert_eq!(back["futureField"], json!({ "kept": true })); assert_eq!(back["targetBaseline"], Value::Null); assert_eq!(back["ownerUid"], 1000); let mut fresh = job.clone(); fresh.target_baseline = None; assert!(serde_json::to_value(&fresh).unwrap().get("targetBaseline").is_none()); assert_eq!(Regex::strip_code("FILE_EXISTS: /x already exists."), "/x already exists."); assert_eq!(part_name("abc", "id1"), ".abc.chj-part-id1"); } #[test] fn summary_matches_0_1_0_rules() { let jobs = vec![json!({ "state": "completed", "size": 10, "transferred": 10 }), json!({ "state": "transferring", "size": 10, "transferred": 5, "speed": 5, "eta": 1 })]; let s = summarize(&jobs); assert_eq!(s["percent"], 75.0); assert_eq!(s["eta"], 1); let done = summarize(&[json!({ "state": "completed", "size": 0, "transferred": 0 })]); assert_eq!(done["percent"], 100.0); } } #[cfg(test)] #[path = "../../../tests/support/transfer_route_identity.rs"] mod route_identity_tests; #[cfg(test)] #[path = "../../../tests/support/transfer_safety.rs"] mod safety_tests;