graft batch 65: 2 lands from fluent-1

.github/workflows/container.yml

@@ -115,7 +115,7 @@ jobs:
           deadline=$((SECONDS + 30))
           setup_token=''
           while (( SECONDS < deadline )); do
-            if ! setup_token=$(docker exec "$container" cat /home/rtbit/db/session/setup-token 2>/dev/null); then
+            if ! setup_token=$(docker exec "$container" cat /home/rtbit/db/com.rtbit.session/setup-token 2>/dev/null); then
               setup_token=''
             fi
             if [[ -n "$setup_token" ]] && curl --fail --silent --show-error \
@@ -140,4 +140,4 @@ jobs:
           docker stop --time 30 "$container" >/dev/null
           [[ "$(docker inspect --format '{{.State.ExitCode}}' "$container")" == 0 ]]
           docker run --rm --volume "$volume:/data" --entrypoint sh "$IMAGE" \
-            -c 'test -s /data/session/session.json'
+            -c 'test -s /data/com.rtbit.session/session.json'

crates/librtbit-core/Cargo.toml

@@ -43,6 +43,7 @@ encoding_rs = "0.8"
 
 [dev-dependencies]
 serde_json = "1"
+tempfile.workspace = true
 
 [lints.clippy]
 all = { level = "warn", priority = -1 }

crates/librtbit-core/src/directories.rs

@@ -1,6 +1,32 @@
 use anyhow::Context;
+use std::path::Path;
 use tracing::{info, warn};
 
+fn migrate_directory(old: &Path, new: &Path) {
+    match (old.exists(), new.exists()) {
+        (true, true) => {
+            warn!(
+                ?old,
+                ?new,
+                "can't migrate configuration as both directories exist, not sure what to do"
+            )
+        }
+        (true, false) => {
+            info!(
+                ?old,
+                ?new,
+                "migrating configuration directories as rtbit was upgraded"
+            );
+            if let Err(e) = std::fs::rename(old, new) {
+                warn!(?old, ?new, "error migrating: {e:#}");
+            }
+        }
+        // In these cases, nothing to migrate, so do nothing.
+        (false, true) => {}
+        (false, false) => {}
+    }
+}
+
 pub fn get_configuration_directory_linux(
     application: &str,
 ) -> anyhow::Result<directories::ProjectDirs> {
@@ -21,32 +47,33 @@ pub fn get_configuration_directory_linux(
         (legacy_dir.config_dir(), new_dir.config_dir()),
         (legacy_dir.data_dir(), new_dir.data_dir()),
     ] {
-        match (old.exists(), new.exists()) {
-            (true, true) => {
-                warn!(
-                    ?old,
-                    ?new,
-                    "can't migrate configuration as both directories exist, not sure what to do"
-                )
-            }
-            (true, false) => {
-                info!(
-                    ?old,
-                    ?new,
-                    "migrating configuration directories as rtbit was upgraded"
-                );
-                if let Err(e) = std::fs::rename(old, new) {
-                    warn!(?old, ?new, "error migrating: {e:#}");
-                }
-            }
-            // In these cases, nothing to migrate, so do nothing.
-            (false, true) => {}
-            (false, false) => {}
-        }
+        migrate_directory(old, new);
     }
     Ok(new_dir)
 }
 
+#[cfg(test)]
+mod tests {
+    use super::migrate_directory;
+
+    #[test]
+    fn migrates_legacy_directory_when_current_directory_is_absent() {
+        let temp = tempfile::tempdir().unwrap();
+        let legacy = temp.path().join("session");
+        let current = temp.path().join("com.rtbit.session");
+        std::fs::create_dir(&legacy).unwrap();
+        std::fs::write(legacy.join("session.json"), "legacy state").unwrap();
+
+        migrate_directory(&legacy, &current);
+
+        assert!(!legacy.exists());
+        assert_eq!(
+            std::fs::read_to_string(current.join("session.json")).unwrap(),
+            "legacy state"
+        );
+    }
+}
+
 pub fn get_configuration_directory(application: &str) -> anyhow::Result<directories::ProjectDirs> {
     #[cfg(not(any(target_os = "macos", target_os = "windows")))]
     {

crates/librtbit/src/session/mod.rs

@@ -1648,6 +1648,16 @@ impl Session {
         for torrent in &torrents {
             self.try_update_persistence_metadata(torrent).await;
         }
+        // A fresh session has no torrent metadata to trigger a persistence
+        // flush. Still write the session database during graceful shutdown so
+        // an empty, healthy state is durable and observable.
+        if torrents.is_empty()
+            && let Some(persistence) = self.persistence.as_ref()
+            && let Ok(settings) = persistence.load_session_settings().await
+            && let Err(error) = persistence.store_session_settings(&settings).await
+        {
+            warn!(?error, "error flushing empty session database");
+        }
     }
 
     /// Save every torrent's state on a timer, so upload totals reach disk even if

crates/rtbit/src/main.rs

@@ -1,3 +1,4 @@
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::{
     collections::HashSet,
     io,
@@ -788,9 +789,11 @@ fn main() -> anyhow::Result<()> {
         .build()?;
 
     let token = tokio_util::sync::CancellationToken::new();
+    let shutdown_complete = Arc::new(AtomicBool::new(false));
     #[cfg(not(target_os = "windows"))]
     {
         let token = token.clone();
+        let shutdown_complete = shutdown_complete.clone();
         let shutdown_timeout = opts.shutdown_timeout;
         use signal_hook::{consts::SIGINT, consts::SIGTERM, iterator::Signals};
         let mut signals = Signals::new([SIGINT, SIGTERM])?;
@@ -808,19 +811,23 @@ fn main() -> anyhow::Result<()> {
                 token.cancel();
                 cancel_triggered = true;
 
+                let shutdown_complete = shutdown_complete.clone();
                 std::thread::spawn(move || {
                     std::thread::sleep(shutdown_timeout);
-                    warn!(
-                        "could not shut down within {:?}, killing myself",
-                        shutdown_timeout
-                    );
-                    std::process::exit(1)
+                    if !shutdown_complete.load(Ordering::Acquire) {
+                        warn!(
+                            "could not shut down within {:?}, killing myself",
+                            shutdown_timeout
+                        );
+                        std::process::exit(1)
+                    }
                 });
             }
         });
     }
 
     let result = rt.block_on(async_main(opts, token.clone()));
+    shutdown_complete.store(true, Ordering::Release);
     match result.as_ref() {
         Err(e) if librtbit_core::spawn_utils::is_cancelled(e) => {
             info!("rtbit shut down: cancelled");
@@ -831,6 +838,11 @@ fn main() -> anyhow::Result<()> {
     rt.shutdown_timeout(Duration::from_secs(1));
     match result {
         Ok(_) => std::process::exit(0),
+        // SIGTERM/SIGINT is the normal container shutdown path.  The signal
+        // cancels the HTTP server, which is reported as an internal
+        // cancellation error, but it must still produce a successful process
+        // exit after the session has flushed its state.
+        Err(e) if librtbit_core::spawn_utils::is_cancelled(&e) => std::process::exit(0),
         Err(_) => std::process::exit(1),
     }
 }