From f02d034c10afe8ce23b90e3d102f3ee43536719d Mon Sep 17 00:00:00 2001 From: Ammar Abou Zor Date: Wed, 9 Sep 2026 18:05:32 +0200 Subject: [PATCH] Improve temp files handling and cleanup Replace the age-based sweep of leftover session files with explicit ownership, so temp files are removed when their owner goes away instead of after 30 days. To achieve this we are using OS lock files, creating one lock and directory per running chipmunk instance. On start we scan chipmunk tmp dir for leftover and try to claim them which means that there are no other chipmunk instance that is using them right now. So they are safe to clean. The current solution still support the previous time-based approach for the remaining files and it will fallback to it in case lock files aren't supported in the running environment. --- crates/app/src/host/service/cleanup.rs | 75 +--- crates/app/src/host/service/mod.rs | 39 +- crates/app/src/session/service/export.rs | 21 +- crates/app/src/session/service/mod.rs | 16 +- crates/core/session/src/lib.rs | 1 + crates/core/session/src/session.rs | 8 +- crates/core/session/src/state/mod.rs | 8 +- crates/core/session/src/state/session_file.rs | 26 +- crates/core/session/src/state/tests_nested.rs | 16 +- crates/core/session/src/temp_dir.rs | 384 ++++++++++++++++++ crates/core/session/tests/nested_search.rs | 8 +- .../core/session/tests/snapshot_tests/utls.rs | 14 +- crates/core/session/tests/text_files.rs | 35 +- 13 files changed, 543 insertions(+), 108 deletions(-) create mode 100644 crates/core/session/src/temp_dir.rs diff --git a/crates/app/src/host/service/cleanup.rs b/crates/app/src/host/service/cleanup.rs index 09e233da1..32864e409 100644 --- a/crates/app/src/host/service/cleanup.rs +++ b/crates/app/src/host/service/cleanup.rs @@ -9,17 +9,20 @@ use std::{ use log::error; use semver::Version; -use session_core::paths::{get_chipmunk_downloads_dir, get_streams_dir}; +use session_core::{paths::get_chipmunk_downloads_dir, temp_dir::InstanceTempDir}; use stypes::{NativeError, NativeErrorKind, Severity}; use crate::{common::app_info, host::service::update::version_dir_prefix}; -/// Cleans up temporary files that were generated by Chipmunk on previous sessions. +/// Cleans up temporary files that were left behind by instances which didn't shut down normally, +/// using `temp_dir` to tell this instance's files from theirs. /// /// Both cleanups are always attempted, failures are logged only since the caller /// has no way to recover from them. -pub fn cleanup_temp_files() { - if let Err(err) = cleanup_temp_dir() { +pub fn cleanup_temp_files(temp_dir: Option<&InstanceTempDir>) { + if let Some(temp_dir) = temp_dir + && let Err(err) = cleanup_temp_dir(temp_dir) + { error!("Error while cleaning up temporary files. Error: {err:?}"); } @@ -28,13 +31,12 @@ pub fn cleanup_temp_files() { } } -/// Iterates through chipmunk temporary directory and remove the entries older -/// than one months. -fn cleanup_temp_dir() -> Result<(), NativeError> { +/// Removes the temp files of instances that aren't running anymore, falling back to the age +/// limit for entries no instance lock can decide about. +fn cleanup_temp_dir(temp_dir: &InstanceTempDir) -> Result<(), NativeError> { const ONE_MONTH: Duration = Duration::from_secs(30 * 24 * 60 * 60); - let tmp_dir = get_streams_dir()?; - cleanup_dir(&tmp_dir, stale_before(ONE_MONTH)?)?; + temp_dir.sweep_leftovers(stale_before(ONE_MONTH)?)?; Ok(()) } @@ -71,20 +73,6 @@ fn cleanup_download_dir() -> Result<(), NativeError> { Ok(()) } -/// Clean files and directory within the given path that have a modified time older than -/// the given modified date limit -fn cleanup_dir(path: &Path, modified_date_limit: SystemTime) -> io::Result<()> { - if !path.exists() { - return Ok(()); - } - - fs::read_dir(path)? - .flat_map(Result::ok) - .filter(|p| is_stale_entry(p, modified_date_limit)) - .map(|entry| entry.path()) - .try_for_each(remove_entry) -} - /// Clean stale entries within the given path, plus the staging directories of the /// given version. fn cleanup_download_entries( @@ -136,13 +124,12 @@ fn remove_entry(path: PathBuf) -> io::Result<()> { mod tests { use std::{ fs::{self, File}, - thread, time::{Duration, SystemTime}, }; use semver::Version; - use super::{cleanup_dir, cleanup_download_entries}; + use super::cleanup_download_entries; #[test] fn cleanup_download_entries_removes_matching_prefix_dirs() { @@ -169,42 +156,4 @@ mod tests { assert!(other_update.exists()); assert!(matching_file.exists()); } - - #[test] - fn test_cleanup_dir() { - // Create temporary directory with some entries - let tempdir = tempfile::tempdir().unwrap(); - let temp_path = tempdir.path(); - - let dir = temp_path.join("dir"); - fs::create_dir(&dir).unwrap(); - let sub_file = dir.join("sub_file"); - _ = File::create(&sub_file).unwrap(); - let file = temp_path.join("file"); - _ = File::create(&file).unwrap(); - - let entries = [dir, sub_file, file]; - - // Make sure there differences in modification time and current time. - thread::sleep(Duration::from_millis(50)); - - // Cleaning up with time stamp one day ago must not remove anything. - let past = SystemTime::now() - .checked_sub(Duration::from_secs(3600)) - .unwrap(); - cleanup_dir(temp_path, past).unwrap(); - for entry in &entries { - assert!(entry.exists()); - } - - // Cleaning up with now must remove all files and directories. - cleanup_dir(temp_path, SystemTime::now()).unwrap(); - - // Temp directory itself shouldn't be removed. - assert!(temp_path.exists()); - - for entry in entries { - assert!(!entry.exists()); - } - } } diff --git a/crates/app/src/host/service/mod.rs b/crates/app/src/host/service/mod.rs index e2d5b2bc3..4fed01890 100644 --- a/crates/app/src/host/service/mod.rs +++ b/crates/app/src/host/service/mod.rs @@ -2,16 +2,18 @@ use std::{ collections::HashMap, ops::Not, path::{Path, PathBuf}, + sync::Arc, thread, }; use anyhow::Result; use itertools::Itertools; -use log::trace; +use log::{error, trace}; use tokio::{runtime::Handle, select, sync::mpsc}; use uuid::Uuid; use parsers::dlt::DltFilterConfig; +use session_core::temp_dir::InstanceTempDir; use stypes::{ DltParserSettings, FileFormat, NativeError, NativeErrorKind, ObserveOptions, ObserveOrigin, ParserType, Severity, SomeIpParserSettings, SomeipFilterConfig, Transport, @@ -72,6 +74,9 @@ pub struct HostService { async_event_rx: mpsc::Receiver, storage: StorageService, plugins: PluginService, + /// Temp directory of this instance, shared with every spawned session. Taken on shutdown so + /// its content is removed once the last session released it. + temp_dir: Option>, } /// Startup data returned after the host service runtime has initialized. @@ -143,14 +148,27 @@ impl HostService { let storage = StorageService::init(async_event_tx.clone()); let plugins = PluginService::init(async_event_tx); + let temp_dir = match InstanceTempDir::in_streams_dir() { + Ok(temp_dir) => Some(Arc::new(temp_dir)), + Err(err) => { + error!( + "Claiming a temp directory for this instance failed. Error: {err:?}" + ); + None + } + }; + let host = Self { communication, async_event_rx, storage, plugins, + temp_dir: temp_dir.clone(), }; - tokio::task::spawn_blocking(cleanup::cleanup_temp_files); + tokio::task::spawn_blocking(move || { + cleanup::cleanup_temp_files(temp_dir.as_deref()) + }); let previous_version = storage::app_version::sync_current_version(); update::spawn_check( host.communication.senders.clone(), @@ -288,7 +306,9 @@ impl HostService { .await; } HostCommand::OnShutdown { confirm_tx } => { - // Cleanup on shutdown goes here. + // Releasing the last reference removes the instance directory. Sessions are closed + // before this command arrives, so this is usually the last one. + self.temp_dir.take(); let _ = confirm_tx.send(()); } HostCommand::CopyFiles { copy_file_infos } => file::copy_files(copy_file_infos).await?, @@ -370,6 +390,15 @@ impl HostService { .await; } + /// Temp directory of this instance, shared with every session it spawns. + fn temp_dir(&self) -> Result, HostError> { + self.temp_dir.clone().ok_or_else(|| { + HostError::InitSessionError(InitSessionError::Other(String::from( + "No temp directory available for session files", + ))) + }) + } + async fn send_plugins_state(&self, state: PluginsState) { self.communication .senders @@ -395,6 +424,7 @@ impl HostService { schema_spec, additional_sources, restore_state, + self.temp_dir()?, ) .await?; @@ -475,6 +505,7 @@ impl HostService { LogSchemaSpec::Text, Vec::new(), None, + self.temp_dir()?, ) .await?; @@ -698,6 +729,7 @@ impl HostService { LogSchemaSpec::Text, Vec::new(), None, + self.temp_dir()?, ) .await { @@ -1077,6 +1109,7 @@ impl HostService { schema_spec, Vec::new(), None, + self.temp_dir()?, ) .await?; diff --git a/crates/app/src/session/service/export.rs b/crates/app/src/session/service/export.rs index c716d893d..cb62a2b75 100644 --- a/crates/app/src/session/service/export.rs +++ b/crates/app/src/session/service/export.rs @@ -1,11 +1,12 @@ //! Session export command handling and generated search-results tab flow. -use std::{ops::RangeInclusive, path::PathBuf}; +use std::{ops::RangeInclusive, path::PathBuf, sync::Arc}; use itertools::Itertools; use uuid::Uuid; use parsers::COLUMN_SEPARATOR; +use session_core::temp_dir::InstanceTempDir; use stypes::{ComputationError, FileFormat, ObserveOptions, ObserveOrigin, ParserType}; use super::{SessionService, SessionStartup, cleanup_temp_source}; @@ -126,7 +127,7 @@ impl SessionService { .map_err(SessionError::NativeError)? .executed; let mode = resolve_mode_from_executed(&executed); - let destination = new_search_results_path(operation_id, mode)?; + let destination = new_search_results_path(operation_id, mode, &self.temp_dir); // Track ownership until the backend confirms export completion. On success the // generated path is transferred to the newly spawned session service. @@ -280,7 +281,9 @@ impl SessionService { let child_session_id = Uuid::new_v4(); let (child_session, child_callback_rx) = - match session_core::session::Session::new(child_session_id).await { + match session_core::session::Session::new(child_session_id, Arc::clone(&self.temp_dir)) + .await + { Ok(session_parts) => session_parts, Err(error) => { cleanup_temp_source(&operation.destination); @@ -302,6 +305,7 @@ impl SessionService { options, schema_spec, additional_sources, + Arc::clone(&self.temp_dir), ) .with_restore_state(Some(restore_state)) .with_temp_source(temp_source) @@ -391,17 +395,16 @@ fn resolve_mode_from_executed(executed: &[ObserveOptions]) -> SearchResultsTabMo fn new_search_results_path( operation_id: Uuid, mode: SearchResultsTabMode, -) -> Result { + temp_dir: &InstanceTempDir, +) -> PathBuf { let extension = match mode { SearchResultsTabMode::PreserveDltBinary => "dlt", SearchResultsTabMode::PreserveText | SearchResultsTabMode::Text => "txt", }; - let path = session_core::paths::get_streams_dir() - .map_err(SessionError::NativeError)? - .join(format!("search-results-{operation_id}.{extension}")); - - Ok(path) + temp_dir + .path() + .join(format!("search-results-{operation_id}.{extension}")) } /// Converts startup failures into the session error channel used by live sessions. diff --git a/crates/app/src/session/service/mod.rs b/crates/app/src/session/service/mod.rs index df44b7792..83edd8156 100644 --- a/crates/app/src/session/service/mod.rs +++ b/crates/app/src/session/service/mod.rs @@ -12,7 +12,7 @@ use tokio::{select, sync::mpsc, task}; use uuid::Uuid; use processor::{grabber::LineRange, search::filter}; -use session_core::session::Session; +use session_core::{session::Session, temp_dir::InstanceTempDir}; use stypes::{ CallbackEvent, ComputationError, NativeError, NativeErrorKind, ObserveOptions, ObserveOrigin, SdeRequest, Severity, Transport, @@ -74,6 +74,9 @@ pub struct SessionService { tracker: OperationTracker, /// Temp sources owned and cleaned up when this service closes. owned_temp_sources: Vec, + /// Temp directory of the running instance, holding it alive while this service owns files in + /// it, and used for the sources it generates. + temp_dir: Arc, /// Optional shutdown confirmation sent after service-owned cleanup runs. shutdown_ack: Option>, } @@ -97,6 +100,8 @@ struct SessionStartup { restore_state: Option, /// Temp sources owned and cleaned up by this service. owned_temp_sources: Vec, + /// Temp directory of the running instance the session writes its files into. + temp_dir: Arc, /// Whether this session participates in recent-session storage. recent_session_policy: RecentSessionPolicy, } @@ -115,15 +120,17 @@ impl SessionService { /// * `schema_spec`: Schema data used to render log rows. /// * `additional_sources`: Additional sources to be attached to session during startup. /// * `restore_state`: State to be restored once session is loaded. + /// * `temp_dir`: Temp directory of the running instance to write session files into. pub async fn spawn( shared_senders: SharedSenders, options: ObserveOptions, schema_spec: LogSchemaSpec, additional_sources: Vec, restore_state: Option, + temp_dir: Arc, ) -> Result { let session_id = Uuid::new_v4(); - let (session, callback_rx) = Session::new(session_id).await?; + let (session, callback_rx) = Session::new(session_id, Arc::clone(&temp_dir)).await?; let startup = SessionStartup::new( shared_senders, @@ -132,6 +139,7 @@ impl SessionService { options, schema_spec, additional_sources, + temp_dir, ) .with_restore_state(restore_state); @@ -148,6 +156,7 @@ impl SessionService { additional_sources, restore_state, owned_temp_sources, + temp_dir, recent_session_policy, } = startup; @@ -205,6 +214,7 @@ impl SessionService { callback_rx, tracker: OperationTracker::default(), owned_temp_sources, + temp_dir, shutdown_ack: None, }; @@ -945,6 +955,7 @@ impl SessionStartup { options: ObserveOptions, schema_spec: LogSchemaSpec, additional_sources: Vec, + temp_dir: Arc, ) -> Self { Self { shared_senders, @@ -955,6 +966,7 @@ impl SessionStartup { additional_sources, restore_state: None, owned_temp_sources: Vec::new(), + temp_dir, recent_session_policy: RecentSessionPolicy::Register, } } diff --git a/crates/core/session/src/lib.rs b/crates/core/session/src/lib.rs index 803e73b57..45c6478db 100644 --- a/crates/core/session/src/lib.rs +++ b/crates/core/session/src/lib.rs @@ -4,4 +4,5 @@ pub mod paths; pub mod session; pub mod state; pub mod tail; +pub mod temp_dir; pub mod tracker; diff --git a/crates/core/session/src/session.rs b/crates/core/session/src/session.rs index d8405b43f..078af42db 100644 --- a/crates/core/session/src/session.rs +++ b/crates/core/session/src/session.rs @@ -6,13 +6,14 @@ use crate::{ operations::Operation, state, state::{IndexedNavigation, NestedMatch, SessionStateAPI}, + temp_dir::InstanceTempDir, tracker, tracker::OperationTrackerAPI, }; use futures::Future; use log::{debug, error, warn}; use processor::{grabber::LineRange, search::filter::SearchFilter}; -use std::{ops::RangeInclusive, path::PathBuf}; +use std::{ops::RangeInclusive, path::PathBuf, sync::Arc}; use tokio::{ join, sync::{ @@ -48,8 +49,11 @@ impl Session { /// The operations loop is the entry point to pass operations requests from an outside thread. /// The state loop is responsible for all state manipulations of the session. /// + /// The session writes its files into `temp_dir` and keeps that directory alive until it is + /// closed. pub async fn new( uuid: Uuid, + temp_dir: Arc, ) -> Result<(Self, UnboundedReceiver), stypes::ComputationError> { let (tx_operations, rx_operations): OperationsChannel = unbounded_channel(); let (tracker_api, rx_tracker_api) = OperationTrackerAPI::new(); @@ -109,7 +113,7 @@ impl Session { &tx_operations, &destroying, "state", - state::run(rx_state_api, tx_callback_events_state) + state::run(rx_state_api, tx_callback_events_state, temp_dir) ), Self::run( &tx_operations, diff --git a/crates/core/session/src/state/mod.rs b/crates/core/session/src/state/mod.rs index 4a188b29a..6d2cac340 100644 --- a/crates/core/session/src/state/mod.rs +++ b/crates/core/session/src/state/mod.rs @@ -18,6 +18,7 @@ use std::{ io::{BufWriter, Write}, ops::RangeInclusive, path::PathBuf, + sync::Arc, }; use tokio::sync::{ mpsc::{self, UnboundedReceiver, UnboundedSender}, @@ -37,6 +38,7 @@ mod source_ids; mod tests_nested; pub(crate) mod values; +use crate::temp_dir::InstanceTempDir; pub use api::{Api, SessionStateAPI}; pub use attachments::{Attachments, AttachmentsError}; pub use indexes::{ @@ -89,9 +91,10 @@ impl SessionState { fn new( tx_callback_events: UnboundedSender, searcher_tx: mpsc::Sender, + temp_dir: Arc, ) -> Self { Self { - session_file: SessionFile::new(), + session_file: SessionFile::new(temp_dir), observed: Observed::new(), search_map: SearchMap::new(), attachments: Attachments::new(), @@ -550,9 +553,10 @@ impl SessionState { pub async fn run( mut rx_api: UnboundedReceiver, tx_callback_events: UnboundedSender, + temp_dir: Arc, ) -> Result<(), stypes::NativeError> { let (search_req_tx, mut search_res_rx) = searchers::spawn(); - let mut state = SessionState::new(tx_callback_events.clone(), search_req_tx); + let mut state = SessionState::new(tx_callback_events.clone(), search_req_tx, temp_dir); let state_cancellation_token = CancellationToken::new(); debug!("task is started"); loop { diff --git a/crates/core/session/src/state/session_file.rs b/crates/core/session/src/state/session_file.rs index 52b443d95..64dcd88cc 100644 --- a/crates/core/session/src/state/session_file.rs +++ b/crates/core/session/src/state/session_file.rs @@ -3,7 +3,7 @@ //! during the session. use super::source_ids::SourceIDs; -use crate::paths; +use crate::temp_dir::InstanceTempDir; use log::debug; use processor::{ grabber::{Grabber, LineRange}, @@ -13,6 +13,7 @@ use std::{ fs::File, io::{BufWriter, Write}, path::PathBuf, + sync::Arc, time::Instant, }; use stypes::GrabbedElement; @@ -36,16 +37,20 @@ pub struct SessionFile { pub writer: Option>, pub last_message_timestamp: Instant, pub sources: SourceIDs, + /// Temp directory the session file is created in. Holding it keeps that directory alive for + /// as long as this session needs its files. + temp_dir: Arc, } impl SessionFile { - pub fn new() -> Self { + pub fn new(temp_dir: Arc) -> Self { SessionFile { grabber: None, filename: None, writer: None, last_message_timestamp: Instant::now(), sources: SourceIDs::new(), + temp_dir, } } @@ -55,8 +60,10 @@ impl SessionFile { if self.grabber.is_some() { return Ok(()); } - let streams = paths::get_streams_dir()?; - let filename = streams.join(format!("{}.{SESSION_FILE_EXTENSION}", Uuid::new_v4())); + let filename = self + .temp_dir + .path() + .join(format!("{}.{SESSION_FILE_EXTENSION}", Uuid::new_v4())); debug!("Session file setup: {}", filename.to_string_lossy()); self.writer = Some(BufWriter::new(File::create(&filename).map_err(|e| { stypes::NativeError { @@ -300,19 +307,16 @@ impl SessionFile { } } -impl Default for SessionFile { - fn default() -> Self { - Self::new() - } -} - #[cfg(test)] mod tests { use std::{ fs::{self, File}, io::BufWriter, + sync::Arc, }; + use crate::temp_dir::InstanceTempDir; + use super::SessionFile; #[test] @@ -323,7 +327,7 @@ mod tests { fs::create_dir(&attachments_dir).unwrap(); fs::write(attachments_dir.join("attachment.txt"), "content").unwrap(); - let mut session_file = SessionFile::new(); + let mut session_file = SessionFile::new(Arc::new(InstanceTempDir::claim(temp_dir.path()))); session_file.filename = Some(session_path.clone()); session_file.writer = Some(BufWriter::new(File::create(&session_path).unwrap())); diff --git a/crates/core/session/src/state/tests_nested.rs b/crates/core/session/src/state/tests_nested.rs index 61e51c0a7..34415e02b 100644 --- a/crates/core/session/src/state/tests_nested.rs +++ b/crates/core/session/src/state/tests_nested.rs @@ -1,13 +1,20 @@ //! Nested-search state contract tests. +use std::sync::Arc; + use processor::search::filter::SearchFilter; +use tempfile::TempDir; use tokio::sync::mpsc::{channel, unbounded_channel}; use tokio_util::sync::CancellationToken; +use crate::temp_dir::InstanceTempDir; + use super::{IndexedNavigation, NestedMatch, SessionState}; struct Fixture { state: SessionState, + /// Streams directory of this fixture, dropped after the state that writes into it. + _streams_dir: TempDir, } impl Fixture { @@ -16,7 +23,9 @@ impl Fixture { fn new(lines: &[&str]) -> Self { let (callback_tx, _callback_rx) = unbounded_channel(); let (search_tx, _search_rx) = channel(1); - let mut state = SessionState::new(callback_tx, search_tx); + let streams_dir = tempfile::tempdir().unwrap(); + let temp_dir = Arc::new(InstanceTempDir::claim(streams_dir.path())); + let mut state = SessionState::new(callback_tx, search_tx, temp_dir); state.session_file.init().unwrap(); let content = lines.iter().map(|line| format!("{line}\n")).collect(); state @@ -28,7 +37,10 @@ impl Fixture { .flush(CancellationToken::new(), true) .unwrap(); - Self { state } + Self { + state, + _streams_dir: streams_dir, + } } fn set_primary(&mut self, rows: &[u64]) { diff --git a/crates/core/session/src/temp_dir.rs b/crates/core/session/src/temp_dir.rs new file mode 100644 index 000000000..f99c38683 --- /dev/null +++ b/crates/core/session/src/temp_dir.rs @@ -0,0 +1,384 @@ +//! Ownership of the temp directory one running instance writes its session files to. +//! +//! An instance locks `inst-.lock` and keeps its files in the sibling `inst-/`; both are +//! removed when [`InstanceTempDir`] is dropped. [`InstanceTempDir::sweep_leftovers`] exists only +//! for abnormal exits: the OS releases the lock when the owning process dies, so another instance +//! can prove a leftover directory is orphaned instead of guessing from modification times. + +use std::{ + fs::{self, File, OpenOptions, TryLockError}, + io, + path::{Path, PathBuf}, + time::SystemTime, +}; + +use log::{debug, warn}; +use uuid::Uuid; + +use crate::paths; + +/// Marks the lock file and the directory of one instance within the streams directory. +const INSTANCE_PREFIX: &str = "inst-"; +const LOCK_EXTENSION: &str = "lock"; + +/// Temp directory of one running instance, removed with its content once dropped. +/// +/// Everything writing into the directory must keep a reference to this value, since the directory +/// lives exactly as long as the last owner of it. +#[derive(Debug)] +pub struct InstanceTempDir { + /// Streams directory holding the directories of all instances. + root: PathBuf, + /// The path of the temp directory for the running instance. + path: PathBuf, + /// `None` when the lock couldn't be acquired, so `path` is the shared streams directory + /// itself, never removed, with leftovers reclaimed by age only. + lock: Option, +} + +/// Lock file proving that the owner of an instance directory is still running. +#[derive(Debug)] +struct InstanceLock { + path: PathBuf, + /// Kept open for the instance's lifetime; never locked again on this handle, since relocking + /// is platform specific. + file: File, +} + +impl InstanceTempDir { + /// Claims a directory within Chipmunk's streams directory. + pub fn in_streams_dir() -> Result { + let streams_dir = paths::get_streams_dir()?; + + let instance = Self::claim(&streams_dir); + + Ok(instance) + } + + /// Claims a directory within `root`, falling back to `root` itself when the lock can't be + /// acquired, e.g. on filesystems without locking support. + pub fn claim(root: &Path) -> Self { + match Self::try_claim(root) { + Ok(temp_dir) => temp_dir, + Err(err) => { + warn!( + "Claiming an instance directory in {} failed: {err}. Temp files of this run \ + stay in the streams directory and are cleaned up by age only.", + root.display() + ); + + Self { + root: root.to_path_buf(), + path: root.to_path_buf(), + lock: None, + } + } + } + } + + /// Directory to write session files and other temp sources of this instance to. + pub fn path(&self) -> &Path { + &self.path + } + + /// Removes leftovers of other instances: directories whose lock proves the owner is gone, + /// plus entries outside the instance layout which were modified before `stale_before`. + pub fn sweep_leftovers(&self, stale_before: SystemTime) -> io::Result<()> { + let own_lock = self.lock.as_ref().map(|lock| lock.path.as_path()); + + sweep_dir(&self.root, own_lock, stale_before) + } + + fn try_claim(root: &Path) -> io::Result { + let id = Uuid::new_v4(); + let lock_path = root.join(format!("{INSTANCE_PREFIX}{id}.{LOCK_EXTENSION}")); + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&lock_path)?; + + // Lock before creating the directory: a sweeper only removes directories whose lock it + // could acquire, so there is no moment in which ours looks orphaned. + file.try_lock().map_err(|err| match err { + TryLockError::Error(err) => err, + TryLockError::WouldBlock => { + io::Error::other(format!("lock {} is already held", lock_path.display())) + } + })?; + + let path = root.join(format!("{INSTANCE_PREFIX}{id}")); + fs::create_dir(&path)?; + debug!("Instance directory claimed: {}", path.display()); + + let lock = InstanceLock { + path: lock_path, + file, + }; + let instance = Self { + root: root.to_path_buf(), + path, + lock: Some(lock), + }; + + Ok(instance) + } +} + +impl Drop for InstanceTempDir { + fn drop(&mut self) { + // Without a lock the directory is the shared streams directory, which is not ours. + let Some(lock) = self.lock.take() else { + return; + }; + + debug!("Removing instance directory: {}", self.path.display()); + // Session files and attachments are already removed one by one when their session closes, + // so this is normally an empty tree. + remove_entry(&self.path); + + // Close first; Windows refuses to remove a locked file. + drop(lock.file); + remove_entry(&lock.path); + } +} + +/// Removes orphaned instance directories and stale entries of the flat layout within `dir`. +/// +/// `own_lock` is the lock file of the sweeping instance, which must be skipped instead of probed. +fn sweep_dir(dir: &Path, own_lock: Option<&Path>, stale_before: SystemTime) -> io::Result<()> { + if !dir.exists() { + return Ok(()); + } + + for entry in fs::read_dir(dir)?.flatten() { + let path = entry.path(); + if is_lock_file(&path) { + // Probing our own lock would only prove that this process is alive. + if own_lock.is_some_and(|own| own == path) { + continue; + } + + // A lock left behind by the probe may only be reclaimed by age once it guards + // nothing: a live instance's lock still has its directory next to it. + if !reclaim_if_orphan(&path) + && !guarded_dir(&path).exists() + && is_stale(&entry, stale_before) + { + remove_entry(&path); + } + continue; + } + + // Entries no lock can decide about fall back to the age rule: legacy flat layout and + // runs without locking support. + if lock_path_of(&path).is_none_or(|lock| !lock.exists()) && is_stale(&entry, stale_before) { + remove_entry(&path); + } + } + + Ok(()) +} + +/// Removes the instance directory guarded by `lock_path` when its lock can be acquired, which +/// proves the owning process is gone. +/// +/// Returns `true` when this lock and the directory it guarded are gone, and `false` when either +/// survived, be it because a live instance holds the lock, because the lock couldn't be probed at +/// all, or because removing the directory failed. +fn reclaim_if_orphan(lock_path: &Path) -> bool { + let lock = match OpenOptions::new().read(true).write(true).open(lock_path) { + Ok(file) => file, + // Another sweeper may have reclaimed this instance in the meantime. + Err(err) if err.kind() == io::ErrorKind::NotFound => return true, + Err(err) => { + warn!("Opening lock {} failed: {err}", lock_path.display()); + return false; + } + }; + + match lock.try_lock() { + Ok(()) => {} + // The lock is held, so a live instance owns this directory. + Err(TryLockError::WouldBlock) => return false, + Err(TryLockError::Error(err)) => { + warn!("Probing lock {} failed: {err}", lock_path.display()); + return false; + } + } + + debug!("Reclaiming leftovers of instance {}", lock_path.display()); + // Keep the lock when its directory survived: the next sweep probes it again and retries, + // instead of leaving the directory unguarded and reclaimable by age only. + if !remove_entry(&guarded_dir(lock_path)) { + return false; + } + + // Unlock and close first; Windows refuses to remove a locked file. + if let Err(err) = lock.unlock() { + warn!("Unlocking {} failed: {err}", lock_path.display()); + } + drop(lock); + remove_entry(lock_path); + + true +} + +fn is_lock_file(path: &Path) -> bool { + has_instance_prefix(path) && path.extension().is_some_and(|ext| ext == LOCK_EXTENSION) +} + +/// Lock file guarding the given entry, or `None` when the entry isn't part of the instance +/// layout. +fn lock_path_of(entry: &Path) -> Option { + has_instance_prefix(entry).then(|| entry.with_extension(LOCK_EXTENSION)) +} + +/// Instance directory the given lock file guards. +fn guarded_dir(lock_path: &Path) -> PathBuf { + lock_path.with_extension("") +} + +fn has_instance_prefix(path: &Path) -> bool { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| name.starts_with(INSTANCE_PREFIX)) +} + +fn is_stale(entry: &fs::DirEntry, stale_before: SystemTime) -> bool { + entry.metadata().is_ok_and(|meta| { + meta.modified() + .is_ok_and(|modified| modified < stale_before) + }) +} + +/// Removes a file or directory tree and reports whether it is gone, logging failures instead of +/// stopping the sweep or panicking on drop. +fn remove_entry(path: &Path) -> bool { + let removed = if path.is_dir() { + fs::remove_dir_all(path) + } else { + fs::remove_file(path) + }; + + match removed { + Ok(()) => true, + Err(err) if err.kind() == io::ErrorKind::NotFound => true, + Err(err) => { + warn!("Removing leftover {} failed: {err}", path.display()); + false + } + } +} + +#[cfg(test)] +mod tests { + use std::{ + fs::{self, File}, + path::{Path, PathBuf}, + time::{Duration, SystemTime}, + }; + + use super::{InstanceTempDir, sweep_dir}; + + /// Time limit that makes every freshly created entry count as stale. + fn everything_stale() -> SystemTime { + SystemTime::now() + .checked_add(Duration::from_secs(60)) + .unwrap() + } + + /// Time limit that makes no entry count as stale. + fn nothing_stale() -> SystemTime { + SystemTime::now() + .checked_sub(Duration::from_secs(60)) + .unwrap() + } + + /// Creates the layout of an instance which is no longer running. + fn leftover_instance(root: &Path, id: &str) -> (PathBuf, PathBuf) { + let lock = root.join(format!("inst-{id}.lock")); + File::create(&lock).unwrap(); + let dir = root.join(format!("inst-{id}")); + fs::create_dir(&dir).unwrap(); + File::create(dir.join("session.session")).unwrap(); + + (lock, dir) + } + + #[test] + fn drop_removes_own_directory_and_lock() { + let temp = tempfile::tempdir().unwrap(); + let temp_dir = InstanceTempDir::claim(temp.path()); + File::create(temp_dir.path().join("session.session")).unwrap(); + + drop(temp_dir); + + assert_eq!( + fs::read_dir(temp.path()).unwrap().count(), + 0, + "neither the directory nor the lock file may survive the instance" + ); + } + + #[test] + fn sweep_reclaims_orphan_instance() { + let temp = tempfile::tempdir().unwrap(); + let temp_dir = InstanceTempDir::claim(temp.path()); + let (lock, dir) = leftover_instance(temp.path(), "gone"); + + temp_dir.sweep_leftovers(nothing_stale()).unwrap(); + + assert!(!dir.exists(), "orphaned directory must be removed"); + assert!(!lock.exists(), "orphaned lock file must be removed"); + assert!(temp_dir.path().exists(), "own directory must survive"); + } + + #[test] + fn sweep_keeps_locked_instance() { + let temp = tempfile::tempdir().unwrap(); + let live = InstanceTempDir::claim(temp.path()); + let dir = live.path().to_path_buf(); + File::create(dir.join("session.session")).unwrap(); + + // Sweeping without knowing the lock is ours is what another instance does: the held lock + // alone must protect the directory. + sweep_dir(temp.path(), None, everything_stale()).unwrap(); + + assert!(dir.exists(), "live instance directory must survive"); + drop(live); + } + + #[test] + fn sweep_removes_stale_entries_of_flat_layout() { + let temp = tempfile::tempdir().unwrap(); + let session = temp.path().join("some-uuid.session"); + File::create(&session).unwrap(); + let attachments = temp.path().join("some-uuid"); + fs::create_dir(&attachments).unwrap(); + File::create(attachments.join("attachment.txt")).unwrap(); + + sweep_dir(temp.path(), None, nothing_stale()).unwrap(); + assert!(session.exists(), "fresh entries must survive"); + assert!(attachments.exists(), "fresh entries must survive"); + + sweep_dir(temp.path(), None, everything_stale()).unwrap(); + assert!(!session.exists()); + assert!(!attachments.exists()); + assert!(temp.path().exists(), "the swept directory itself must stay"); + } + + #[test] + fn sweep_ages_out_instance_directory_without_lock() { + let temp = tempfile::tempdir().unwrap(); + let dir = temp.path().join("inst-lockless"); + fs::create_dir(&dir).unwrap(); + + sweep_dir(temp.path(), None, nothing_stale()).unwrap(); + assert!(dir.exists()); + + sweep_dir(temp.path(), None, everything_stale()).unwrap(); + assert!(!dir.exists()); + } +} diff --git a/crates/core/session/tests/nested_search.rs b/crates/core/session/tests/nested_search.rs index bd335ba0b..9b0179a49 100644 --- a/crates/core/session/tests/nested_search.rs +++ b/crates/core/session/tests/nested_search.rs @@ -1,9 +1,12 @@ //! Public nested-search API contract tests. +use std::sync::Arc; + use processor::search::filter::SearchFilter; use session::{ session::Session, state::{IndexedNavigation, NestedMatch}, + temp_dir::InstanceTempDir, }; use uuid::Uuid; @@ -22,7 +25,10 @@ async fn public_api_returns_named_nested_coordinates() { .collect(); let session_id = Uuid::new_v4(); - let (session, _events) = Session::new(session_id).await.unwrap(); + // Session files go to a throwaway instance directory instead of the user's Chipmunk home. + let streams_dir = tempfile::tempdir().unwrap(); + let temp_dir = Arc::new(InstanceTempDir::claim(streams_dir.path())); + let (session, _events) = Session::new(session_id, temp_dir).await.unwrap(); session.state.create_session_file().await.unwrap(); session.state.write_session_file(0, content).await.unwrap(); session.state.flush_session_file().await.unwrap(); diff --git a/crates/core/session/tests/snapshot_tests/utls.rs b/crates/core/session/tests/snapshot_tests/utls.rs index 10ab9be54..cdbb2a792 100644 --- a/crates/core/session/tests/snapshot_tests/utls.rs +++ b/crates/core/session/tests/snapshot_tests/utls.rs @@ -1,6 +1,9 @@ use serde::{Deserialize, Serialize}; -use session::session::Session; -use std::path::{Path, PathBuf}; +use session::{session::Session, temp_dir::InstanceTempDir}; +use std::{ + path::{Path, PathBuf}, + sync::Arc, +}; use uuid::Uuid; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -102,7 +105,12 @@ pub async fn run_observe_session>( ); let uuid = Uuid::new_v4(); - let (session, mut receiver) = Session::new(uuid).await.expect("Session should be created"); + // Session files go to a throwaway instance directory instead of the user's Chipmunk home. + let streams_dir = tempfile::tempdir().unwrap(); + let temp_dir = Arc::new(InstanceTempDir::claim(streams_dir.path())); + let (session, mut receiver) = Session::new(uuid, temp_dir) + .await + .expect("Session should be created"); session .observe( diff --git a/crates/core/session/tests/text_files.rs b/crates/core/session/tests/text_files.rs index 05184749d..b545d3d45 100644 --- a/crates/core/session/tests/text_files.rs +++ b/crates/core/session/tests/text_files.rs @@ -1,11 +1,11 @@ //! Contract tests for text files, which are parsed by the tokenizer like any other source //! instead of being read as session content directly. -use std::{fs::read_to_string, io::Write, path::Path, time::Duration}; +use std::{fs::read_to_string, io::Write, path::Path, sync::Arc, time::Duration}; -use session::session::Session; +use session::{session::Session, temp_dir::InstanceTempDir}; use stypes::{CallbackEvent, FileFormat, ObserveOptions, ObserveOrigin, ParserType}; -use tempfile::NamedTempFile; +use tempfile::{NamedTempFile, TempDir}; use tokio::sync::mpsc::UnboundedReceiver; use uuid::Uuid; @@ -33,11 +33,23 @@ async fn wait_for_event( .expect("Expected event must arrive"); } +/// Throwaway streams directory with the instance directory of one test session in it, so tests +/// never write into the user's Chipmunk home. The guard must outlive the session. +fn test_temp_dir() -> (TempDir, Arc) { + let streams_dir = tempfile::tempdir().unwrap(); + let temp_dir = Arc::new(InstanceTempDir::claim(streams_dir.path())); + + (streams_dir, temp_dir) +} + /// Observes `path` with the text parser and waits until its content has been read into the -/// session file. -async fn observe_text_file(path: &Path) -> (Session, UnboundedReceiver) { +/// session file. The streams directory is returned first, so it outlives the session. +async fn observe_text_file(path: &Path) -> (TempDir, Session, UnboundedReceiver) { let uuid = Uuid::new_v4(); - let (session, mut receiver) = Session::new(uuid).await.expect("Session should be created"); + let (streams_dir, temp_dir) = test_temp_dir(); + let (session, mut receiver) = Session::new(uuid, temp_dir) + .await + .expect("Session should be created"); session .observe( @@ -51,7 +63,7 @@ async fn observe_text_file(path: &Path) -> (Session, UnboundedReceiver