Improve temp files handling and cleanup

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());
-        }
-    }
 }

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<HostAsyncEvent>,
     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<Arc<InstanceTempDir>>,
 }
 
 /// 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<Arc<InstanceTempDir>, 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?;
 

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<PathBuf, SessionError> {
+    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.

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<PathBuf>,
+    /// 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<InstanceTempDir>,
     /// Optional shutdown confirmation sent after service-owned cleanup runs.
     shutdown_ack: Option<StdSender<()>>,
 }
@@ -97,6 +100,8 @@ struct SessionStartup {
     restore_state: Option<RecentSessionStateSnapshot>,
     /// Temp sources owned and cleaned up by this service.
     owned_temp_sources: Vec<PathBuf>,
+    /// Temp directory of the running instance the session writes its files into.
+    temp_dir: Arc<InstanceTempDir>,
     /// 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<ObserveOrigin>,
         restore_state: Option<RecentSessionStateSnapshot>,
+        temp_dir: Arc<InstanceTempDir>,
     ) -> Result<SpawnedSession, InitSessionError> {
         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<ObserveOrigin>,
+        temp_dir: Arc<InstanceTempDir>,
     ) -> 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,
         }
     }

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;