diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 2dabdfe..28bd923 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -12,6 +12,7 @@ mod shortcut; #[path = "../common/window.rs"] mod window; +mod local_fs; mod notifications; mod scheduler; mod zookeeper; @@ -819,6 +820,7 @@ pub fn run() { } let mut app = builder + .manage(local_fs::LocalFsState::default()) .manage::(std::sync::Mutex::new( zookeeper::DaemonState::default(), )) @@ -853,6 +855,8 @@ pub fn run() { is_flatpak, is_linux_mint, has_flatpak_permissions, + local_fs::list_local_directory, + local_fs::cancel_local_directory, open_full_window, open_window, open_small_window, diff --git a/src-tauri/src/local_fs.rs b/src-tauri/src/local_fs.rs new file mode 100644 index 0000000..3c20676 --- /dev/null +++ b/src-tauri/src/local_fs.rs @@ -0,0 +1,489 @@ +use chrono::{DateTime, Utc}; +use serde::Serialize; +use std::collections::{HashMap, HashSet}; +use std::fs; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; +use tauri::{ipc::Channel, State}; + +const CANCELLED: &str = "Local directory listing cancelled"; +const MAX_SCAN_DIRECTORIES: usize = 10_000; +const MAX_SCAN_ENTRIES: usize = 100_000; +const MAX_SCAN_DURATION: Duration = Duration::from_millis(3_500); + +#[derive(Default)] +pub struct LocalFsState { + requests: Arc>>>, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub struct LocalEntry { + name: String, + full_path: String, + is_dir: bool, + size: Option, + mod_time: Option, +} + +#[derive(Serialize)] +#[serde( + tag = "event", + rename_all = "camelCase", + rename_all_fields = "camelCase" +)] +pub enum LocalListingEvent { + Entries { + entries: Vec, + }, + Size { + full_path: String, + size: Option, + }, + Error { + message: String, + }, + Complete, +} + +#[derive(Debug)] +enum ScanError { + BudgetExceeded, + Cancelled, + Incomplete, +} + +#[cfg(unix)] +type DirectoryIdentity = (u64, u64); +#[cfg(unix)] +type DeviceIdentity = u64; + +#[cfg(unix)] +fn directory_info(path: &Path) -> std::io::Result<(DirectoryIdentity, DeviceIdentity)> { + use std::os::unix::fs::MetadataExt; + + let metadata = fs::metadata(path)?; + Ok(((metadata.dev(), metadata.ino()), metadata.dev())) +} + +#[cfg(windows)] +type DirectoryIdentity = PathBuf; +#[cfg(windows)] +type DeviceIdentity = std::ffi::OsString; + +#[cfg(windows)] +fn directory_info(path: &Path) -> std::io::Result<(DirectoryIdentity, DeviceIdentity)> { + use std::io::{Error, ErrorKind}; + + let canonical = fs::canonicalize(path)?; + let device = canonical + .components() + .next() + .ok_or_else(|| Error::new(ErrorKind::InvalidInput, "path has no volume prefix"))? + .as_os_str() + .to_os_string(); + Ok((canonical, device)) +} + +#[cfg(not(any(unix, windows)))] +type DirectoryIdentity = PathBuf; +#[cfg(not(any(unix, windows)))] +type DeviceIdentity = (); + +#[cfg(not(any(unix, windows)))] +fn directory_info(path: &Path) -> std::io::Result<(DirectoryIdentity, DeviceIdentity)> { + fs::canonicalize(path).map(|path| (path, ())) +} + +struct ScanBudget { + started: Instant, + directories: usize, + entries: usize, +} + +impl ScanBudget { + fn new() -> Self { + Self { + started: Instant::now(), + directories: 0, + entries: 0, + } + } + + fn visit_directory(&mut self) -> Result<(), ScanError> { + if self.started.elapsed() >= MAX_SCAN_DURATION || self.directories >= MAX_SCAN_DIRECTORIES { + return Err(ScanError::BudgetExceeded); + } + self.directories += 1; + Ok(()) + } + + fn visit_entry(&mut self) -> Result<(), ScanError> { + if self.started.elapsed() >= MAX_SCAN_DURATION || self.entries >= MAX_SCAN_ENTRIES { + return Err(ScanError::BudgetExceeded); + } + self.entries += 1; + Ok(()) + } +} + +fn directory_size( + path: &Path, + allowed_device: &DeviceIdentity, + cancelled: &AtomicBool, + budget: &mut ScanBudget, +) -> Result { + let mut total = 0_u64; + let mut pending = vec![path.to_path_buf()]; + let mut visited = HashSet::new(); + + while let Some(directory) = pending.pop() { + if cancelled.load(Ordering::Relaxed) { + return Err(ScanError::Cancelled); + } + budget.visit_directory()?; + + let (identity, device) = directory_info(&directory).map_err(|_| ScanError::Incomplete)?; + if &device != allowed_device { + if directory == path { + return Err(ScanError::Incomplete); + } + continue; + } + if !visited.insert(identity) { + continue; + } + + let entries = fs::read_dir(directory).map_err(|_| ScanError::Incomplete)?; + for entry in entries { + if cancelled.load(Ordering::Relaxed) { + return Err(ScanError::Cancelled); + } + budget.visit_entry()?; + + let entry = entry.map_err(|_| ScanError::Incomplete)?; + let file_type = entry.file_type().map_err(|_| ScanError::Incomplete)?; + if file_type.is_symlink() { + continue; + } + if file_type.is_dir() { + pending.push(entry.path()); + } else if file_type.is_file() { + let size = entry.metadata().map_err(|_| ScanError::Incomplete)?.len(); + total = total.checked_add(size).ok_or(ScanError::Incomplete)?; + } + } + } + + Ok(total) +} + +fn list_local_entries( + path: &Path, + cancelled: &AtomicBool, +) -> Result<(Vec, Vec), String> { + if cancelled.load(Ordering::Relaxed) { + return Err(CANCELLED.to_string()); + } + + let entries = fs::read_dir(path) + .map_err(|error| format!("Failed to read {}: {}", path.display(), error))?; + let mut result = Vec::new(); + let mut directories = Vec::new(); + + for entry in entries { + if cancelled.load(Ordering::Relaxed) { + return Err(CANCELLED.to_string()); + } + + let Ok(entry) = entry else { + continue; + }; + let name = entry.file_name().to_string_lossy().into_owned(); + if name.starts_with('.') { + continue; + } + let Ok(file_type) = entry.file_type() else { + continue; + }; + if file_type.is_symlink() { + continue; + } + + let full_path = entry.path(); + let metadata = entry.metadata().ok(); + let is_dir = file_type.is_dir(); + let size = if file_type.is_file() { + metadata.as_ref().map(fs::Metadata::len) + } else { + None + }; + if is_dir { + directories.push(full_path.clone()); + } + let mod_time = metadata + .and_then(|metadata| metadata.modified().ok()) + .map(|time| DateTime::::from(time).to_rfc3339()); + + result.push(LocalEntry { + name, + full_path: full_path.to_string_lossy().into_owned(), + is_dir, + size, + mod_time, + }); + } + + directories.sort(); + Ok((result, directories)) +} + +#[tauri::command] +pub fn list_local_directory( + state: State<'_, LocalFsState>, + path: String, + request_id: String, + on_event: Channel, +) -> Result<(), String> { + let cancelled = Arc::new(AtomicBool::new(false)); + { + let mut requests = state.requests.lock().map_err(|error| error.to_string())?; + if let Some(previous) = requests.insert(request_id.clone(), Arc::clone(&cancelled)) { + previous.store(true, Ordering::Relaxed); + } + } + + let requests = Arc::clone(&state.requests); + let scan_cancelled = Arc::clone(&cancelled); + let _ = tauri::async_runtime::spawn_blocking(move || { + let allowed_device = directory_info(Path::new(&path)) + .ok() + .map(|(_, device)| device); + match list_local_entries(Path::new(&path), &scan_cancelled) { + Ok((entries, directories)) => { + if on_event + .send(LocalListingEvent::Entries { entries }) + .is_ok() + { + let mut budget = ScanBudget::new(); + for directory in directories { + let size = match allowed_device.as_ref() { + Some(device) => { + match directory_size( + &directory, + device, + &scan_cancelled, + &mut budget, + ) { + Ok(size) => Some(size), + Err(ScanError::BudgetExceeded | ScanError::Incomplete) => None, + Err(ScanError::Cancelled) => break, + } + } + None => None, + }; + if on_event + .send(LocalListingEvent::Size { + full_path: directory.to_string_lossy().into_owned(), + size, + }) + .is_err() + { + break; + } + } + } + } + Err(message) if message != CANCELLED => { + let _ = on_event.send(LocalListingEvent::Error { message }); + } + Err(_) => {} + } + + let _ = on_event.send(LocalListingEvent::Complete); + + if let Ok(mut requests) = requests.lock() { + if requests + .get(&request_id) + .is_some_and(|active| Arc::ptr_eq(active, &cancelled)) + { + requests.remove(&request_id); + } + } + }); + + Ok(()) +} + +#[tauri::command] +pub fn cancel_local_directory(state: State<'_, LocalFsState>, request_id: String) { + if let Ok(requests) = state.requests.lock() { + if let Some(cancelled) = requests.get(&request_id) { + cancelled.store(true, Ordering::Relaxed); + } + } +} + +#[cfg(test)] +mod tests { + use super::{ + directory_info, directory_size, list_local_entries, LocalListingEvent, ScanBudget, + ScanError, CANCELLED, MAX_SCAN_DIRECTORIES, + }; + use std::fs; + use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + use std::time::{SystemTime, UNIX_EPOCH}; + + static NEXT_TEST_ID: AtomicU64 = AtomicU64::new(0); + + fn test_directory() -> std::path::PathBuf { + let suffix = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let id = NEXT_TEST_ID.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!( + "s-tray-local-fs-{}-{suffix}-{id}", + std::process::id() + )) + } + + #[test] + fn lists_file_and_recursive_folder_sizes() { + let root = test_directory(); + fs::create_dir_all(root.join("folder/nested")).unwrap(); + fs::create_dir(root.join("empty")).unwrap(); + fs::write(root.join("file.txt"), b"12345").unwrap(); + fs::write(root.join("folder/one.bin"), b"123").unwrap(); + fs::write(root.join("folder/nested/two.bin"), b"1234567").unwrap(); + + let cancelled = AtomicBool::new(false); + let allowed_device = directory_info(&root).unwrap().1; + let (entries, _) = list_local_entries(&root, &cancelled).unwrap(); + let mut budget = ScanBudget::new(); + assert_eq!( + entries + .iter() + .find(|entry| entry.name == "file.txt") + .unwrap() + .size, + Some(5) + ); + assert_eq!( + directory_size( + &root.join("folder"), + &allowed_device, + &cancelled, + &mut budget, + ) + .unwrap(), + 10 + ); + assert_eq!( + directory_size( + &root.join("empty"), + &allowed_device, + &cancelled, + &mut budget, + ) + .unwrap(), + 0 + ); + + fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn stops_before_scanning_when_cancelled() { + let root = test_directory(); + fs::create_dir(&root).unwrap(); + let cancelled = AtomicBool::new(true); + + let result = list_local_entries(&root, &cancelled); + assert!(matches!(result, Err(error) if error == CANCELLED)); + + fs::remove_dir(root).unwrap(); + } + + #[test] + fn serializes_channel_events_for_the_frontend() { + let event = serde_json::to_value(LocalListingEvent::Size { + full_path: "/tmp/folder".to_string(), + size: Some(42), + }) + .unwrap(); + + assert_eq!(event["event"], "size"); + assert_eq!(event["fullPath"], "/tmp/folder"); + assert_eq!(event["size"], 42); + } + + #[test] + fn stops_when_the_scan_budget_is_exhausted() { + let mut budget = ScanBudget { + started: std::time::Instant::now(), + directories: MAX_SCAN_DIRECTORIES, + entries: 0, + }; + + assert!(matches!( + budget.visit_directory(), + Err(ScanError::BudgetExceeded) + )); + } + + #[cfg(any(unix, windows))] + #[test] + fn rejects_a_directory_on_another_device() { + let root = test_directory(); + fs::create_dir(&root).unwrap(); + #[cfg(unix)] + let device = directory_info(&root).unwrap().1.wrapping_add(1); + #[cfg(windows)] + let device = { + let mut device = directory_info(&root).unwrap().1; + device.push("-other"); + device + }; + let result = directory_size( + &root, + &device, + &AtomicBool::new(false), + &mut ScanBudget::new(), + ); + + assert!(matches!(result, Err(ScanError::Incomplete))); + + fs::remove_dir(root).unwrap(); + } + + #[cfg(unix)] + #[test] + fn excludes_symlinks_from_folder_sizes() { + use std::os::unix::fs::symlink; + + let root = test_directory(); + fs::create_dir_all(root.join("folder")).unwrap(); + fs::write(root.join("folder/file.bin"), b"123").unwrap(); + symlink(&root, root.join("folder/loop")).unwrap(); + + let cancelled = AtomicBool::new(false); + let allowed_device = directory_info(&root).unwrap().1; + let _ = list_local_entries(&root, &cancelled).unwrap(); + let mut budget = ScanBudget::new(); + assert_eq!( + directory_size( + &root.join("folder"), + &allowed_device, + &cancelled, + &mut budget, + ) + .unwrap(), + 3 + ); + + fs::remove_dir_all(root).unwrap(); + } +} diff --git a/src/components/navigator/FileList.tsx b/src/components/navigator/FileList.tsx index cd2f66b..ca57c03 100644 --- a/src/components/navigator/FileList.tsx +++ b/src/components/navigator/FileList.tsx @@ -250,7 +250,7 @@ export default function FileList({
- {!entry.isDir && typeof entry.size === 'number' + {typeof entry.size === 'number' && entry.size >= 0 ? formatBytes(entry.size) : '—'}
diff --git a/src/components/navigator/useFileNavigation.ts b/src/components/navigator/useFileNavigation.ts index eab87a5..673624c 100644 --- a/src/components/navigator/useFileNavigation.ts +++ b/src/components/navigator/useFileNavigation.ts @@ -1,6 +1,6 @@ import { useQuery } from '@tanstack/react-query' +import { Channel, invoke } from '@tauri-apps/api/core' import { homeDir } from '@tauri-apps/api/path' -import { readDir } from '@tauri-apps/plugin-fs' import { startTransition, useCallback, useEffect, useMemo, useRef, useState } from 'react' import rclone from '../../../lib/rclone/client.ts' import { useHostStore } from '../../../store/host.ts' @@ -74,10 +74,14 @@ export default function useFileNavigation({ const canShowRemotes = useMemo(() => allowedKeys.includes('REMOTES'), [allowedKeys]) const cacheRef = useRef>(new Map()) + const localCompleteCacheRef = useRef>(new Set()) const entryByKeyRef = useRef>(new Map()) const abortControllerRef = useRef(null) const loadingTimerRef = useRef | null>(null) const isNavigatingRef = useRef(false) + const localRequestIdRef = useRef(null) + const localRequestPrefixRef = useRef(Math.random().toString(36).slice(2)) + const localRequestSequenceRef = useRef(0) const [selectedPaths, setSelectedPaths] = useState>(new Set()) const selectedTypesRef = useRef>(new Map()) @@ -314,6 +318,7 @@ export default function useFileNavigation({ const refresh = useCallback(() => { const cKey = cacheKey(selectedRemote, cwd) cacheRef.current.delete(cKey) + localCompleteCacheRef.current.delete(cKey) startTransition(() => setItems([])) setIsLoading(true) setRefreshKey((k) => k + 1) @@ -380,6 +385,7 @@ export default function useFileNavigation({ // biome-ignore lint/correctness/useExhaustiveDependencies: refreshKey is an intentional re-run trigger the body doesn't read — refresh() evicts the cacheRef entry, clears items, and bumps it to force a refetch of the current directory; removing it breaks the Refresh button (empty panel, isLoading stuck true) useEffect(() => { if (!isActive) return + const cKey = cacheKey(selectedRemote, cwd) async function loadDir() { log('loadDir: start', { selectedRemote, cwd, isRemote }) @@ -410,12 +416,23 @@ export default function useFileNavigation({ } }, 200) - const cKey = cacheKey(selectedRemote, cwd) if (cacheRef.current.has(cKey)) { log('loadDir: cache hit', cKey) const cached = cacheRef.current.get(cKey)! startTransition(() => setItems(cached)) isNavigatingRef.current = false + if ( + selectedRemote === 'UI_LOCAL_FS' && + localCompleteCacheRef.current.has(cKey) + ) { + finished = true + if (loadingTimerRef.current) { + clearTimeout(loadingTimerRef.current) + loadingTimerRef.current = null + } + setIsLoading(false) + return + } } let nextItems: Entry[] = [] @@ -527,10 +544,130 @@ export default function useFileNavigation({ }) .filter((e) => !e.name.startsWith('.')) } else if (cwd) { - let entries: Awaited> | null = null + const requestId = `${localRequestPrefixRef.current}-${++localRequestSequenceRef.current}` + localRequestIdRef.current = requestId + let streamedItems: Entry[] = [] + let streamedIndexes = new Map() + const pendingSizes = new Map() + let sizeFrame: number | null = null + let streamFailed = false + const channel = new Channel< + | { + event: 'entries' + entries: { + name: string + fullPath: string + isDir: boolean + size: number | null + modTime: string | null + }[] + } + | { event: 'size'; fullPath: string; size: number | null } + | { event: 'error'; message: string } + | { event: 'complete' } + >() + const cancelSizeFrame = () => { + if (sizeFrame !== null) cancelAnimationFrame(sizeFrame) + sizeFrame = null + } + const flushSizeUpdates = () => { + if (controller.signal.aborted || pendingSizes.size === 0) { + pendingSizes.clear() + return + } + + const nextItems = [...streamedItems] + for (const [fullPath, size] of pendingSizes) { + const index = streamedIndexes.get(fullPath) + if (index === undefined) continue + const updated = { ...nextItems[index], size } + nextItems[index] = updated + entryByKeyRef.current.set(updated.key, updated) + } + pendingSizes.clear() + streamedItems = nextItems + cacheRef.current.set(cKey, streamedItems) + startTransition(() => setItems(streamedItems)) + } try { - entries = await readDir(cwd) + await new Promise((resolve, reject) => { + channel.onmessage = (event) => { + if (event.event === 'complete') { + cancelSizeFrame() + flushSizeUpdates() + if (!controller.signal.aborted && !streamFailed) { + localCompleteCacheRef.current.add(cKey) + } + if (localRequestIdRef.current === requestId) { + localRequestIdRef.current = null + } + resolve() + return + } + if (controller.signal.aborted) return + if (event.event === 'error') { + streamFailed = true + cancelSizeFrame() + pendingSizes.clear() + cacheRef.current.delete(cKey) + localCompleteCacheRef.current.delete(cKey) + reject(new Error(event.message)) + return + } + if (event.event === 'entries') { + streamedItems = event.entries.map( + (entry) => + ({ + key: entry.fullPath, + name: entry.name, + isDir: entry.isDir, + size: entry.size ?? undefined, + modTime: entry.modTime ?? undefined, + remote: 'UI_LOCAL_FS', + fullPath: entry.fullPath, + }) as Entry + ) + streamedItems.sort((a, b) => { + if (a.isDir && !b.isDir) return -1 + if (!a.isDir && b.isDir) return 1 + return a.name.localeCompare(b.name) + }) + streamedIndexes = new Map( + streamedItems.map((entry, index) => [entry.fullPath, index]) + ) + finished = true + if (loadingTimerRef.current) { + clearTimeout(loadingTimerRef.current) + loadingTimerRef.current = null + } + cacheRef.current.set(cKey, streamedItems) + localCompleteCacheRef.current.delete(cKey) + const map = entryByKeyRef.current + for (const entry of streamedItems) map.set(entry.key, entry) + setItems(streamedItems) + setIsLoading(false) + isNavigatingRef.current = false + return + } + + pendingSizes.set(event.fullPath, event.size ?? undefined) + if (sizeFrame === null) { + sizeFrame = requestAnimationFrame(() => { + sizeFrame = null + flushSizeUpdates() + }) + } + } + + invoke('list_local_directory', { + path: cwd, + requestId, + onEvent: channel, + }).catch(reject) + }) } catch { + cacheRef.current.delete(cKey) + localCompleteCacheRef.current.delete(cKey) if (!controller.signal.aborted) { finished = true if (loadingTimerRef.current) { @@ -543,22 +680,15 @@ export default function useFileNavigation({ isNavigatingRef.current = false } return + } finally { + cancelSizeFrame() + pendingSizes.clear() + if (localRequestIdRef.current === requestId) { + localRequestIdRef.current = null + } } - if (controller.signal.aborted) return - - const processedEntries: Entry[] = [] - for (const e of entries || []) { - if (e.isSymlink) continue - const full = await joinLocal(cwd, e.name) - processedEntries.push({ - key: full, - name: e.name, - isDir: e.isDirectory, - fullPath: full, - } as Entry) - } - nextItems = processedEntries.filter((e) => !e.name.startsWith('.')) + return } else { nextItems = [] } @@ -590,6 +720,14 @@ export default function useFileNavigation({ return () => { if (abortControllerRef.current) abortControllerRef.current.abort() + if (localRequestIdRef.current) { + cacheRef.current.delete(cKey) + localCompleteCacheRef.current.delete(cKey) + invoke('cancel_local_directory', { + requestId: localRequestIdRef.current, + }).catch(() => {}) + localRequestIdRef.current = null + } if (loadingTimerRef.current) { clearTimeout(loadingTimerRef.current) loadingTimerRef.current = null