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