fix: prevent Xtream refresh deadlocks (#814)

* **New Features**
  * Added staged, atomic publication for playlist databases and category refreshes.
  * Refreshes preserve existing playlist details and report update outcomes.
* **Bug Fixes**
  * Improved recovery from interrupted writes and pending changes.
  * Prevented conflicts between staging and published data, including path aliases.
  * Improved cleanup of temporary, stale, and orphaned refresh artifacts.
* **Reliability**
  * Enhanced handling and reporting of locking, synchronization, and publication errors.
  * Improved behavior during concurrent refreshes and repeated updates.

backend/src/repository/bplustree/common.rs

@@ -4,7 +4,7 @@ use memmap2::Mmap;
 pub(crate) use memmap2::Advice;
 use std::{
     ffi::OsString,
-    fs::File,
+    fs::{self, File},
     io,
     path::{Path, PathBuf},
 };
@@ -117,6 +117,99 @@ pub(crate) fn sidecar_lock_path(filepath: &Path) -> PathBuf {
     }
 }
 
+/// Resolves an existing path, or its existing parent plus the not-yet-created leaf.
+///
+/// B+Tree staging names are validated before every artifact exists. Canonicalizing the
+/// parent still collapses `..` components and symlinked directory aliases without
+/// requiring callers to create cleanup-owned files first.
+pub(crate) fn resolved_path_identity(path: &Path) -> io::Result<PathBuf> {
+    match fs::canonicalize(path) {
+        Ok(resolved) => Ok(resolved),
+        Err(error) if error.kind() == io::ErrorKind::NotFound => {
+            let leaf = path.file_name().ok_or_else(|| {
+                io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    format!("path has no file name for identity resolution: {}", path.display()),
+                )
+            })?;
+            let parent = path
+                .parent()
+                .filter(|parent| !parent.as_os_str().is_empty())
+                .unwrap_or_else(|| Path::new("."));
+            fs::canonicalize(parent)
+                .map(|resolved_parent| resolved_parent.join(leaf))
+                .map_err(|parent_error| {
+                    io::Error::new(
+                        parent_error.kind(),
+                        format!(
+                            "failed to resolve parent directory for path identity {}: {parent_error}",
+                            path.display()
+                        ),
+                    )
+                })
+        }
+        Err(error) => Err(io::Error::new(
+            error.kind(),
+            format!("failed to resolve path identity {}: {error}", path.display()),
+        )),
+    }
+}
+
+pub(crate) fn ensure_distinct_sidecar_lock_domains(published: &Path, staging: &Path) -> io::Result<()> {
+    let published_lock = sidecar_lock_path(published);
+    let staging_lock = sidecar_lock_path(staging);
+    let published_identity = resolved_path_identity(&published_lock)?;
+    let staging_identity = resolved_path_identity(&staging_lock)?;
+    if published_identity == staging_identity {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!(
+                "published database {} and staging database {} share resolved sidecar lock {}",
+                published.display(),
+                staging.display(),
+                published_identity.display()
+            ),
+        ));
+    }
+    Ok(())
+}
+
+pub(crate) fn remove_file_if_exists(path: &Path) -> io::Result<()> {
+    match fs::remove_file(path) {
+        Ok(()) => Ok(()),
+        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
+        Err(error) => Err(io::Error::new(
+            error.kind(),
+            format!("failed to remove file {}: {error}", path.display()),
+        )),
+    }
+}
+
+pub(crate) fn parent_or_dot(path: &Path) -> &Path {
+    path.parent()
+        .filter(|parent| !parent.as_os_str().is_empty())
+        .unwrap_or_else(|| Path::new("."))
+}
+
+pub(crate) fn same_parent_directory(left: &Path, right: &Path) -> bool {
+    parent_or_dot(left) == parent_or_dot(right)
+}
+
+pub(crate) fn require_same_parent_directory(staging: &Path, published: &Path) -> io::Result<()> {
+    if same_parent_directory(staging, published) {
+        Ok(())
+    } else {
+        Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!(
+                "staging path {} and published path {} must share one parent directory",
+                staging.display(),
+                published.display()
+            ),
+        ))
+    }
+}
+
 #[derive(Debug)]
 pub enum BPlusTreeError {
     Io(io::Error),

backend/src/repository/bplustree/v3/mod.rs

@@ -1,9 +1,12 @@
 mod format;
 pub(crate) mod migration;
 mod page;
+mod publish;
 mod tree;
 mod wal;
 
+pub(crate) use publish::{publish_staged_database, BPlusTreeStagingArtifacts};
+
 #[allow(unused_imports)]
 pub use tree::{
     BPlusTree, BPlusTreeDiskIterator, BPlusTreeDiskIteratorOwned, BPlusTreeQuery, BPlusTreeRangeIterator,

backend/src/repository/bplustree/v3/publish.rs

@@ -0,0 +1,555 @@
+use super::{
+    tree::{publish_database_and_invalidate_sorted_index, BPlusTreeUpdate},
+    wal::{
+        recover_pending_under_existing_lock, sync_parent_directory, wal_path, wal_temporary_path,
+        ExclusiveSidecarGuard,
+    },
+};
+use crate::repository::{
+    bplustree::common::{
+        ensure_distinct_sidecar_lock_domains, remove_file_if_exists, require_same_parent_directory,
+        resolved_path_identity, sidecar_lock_path,
+    },
+    storage::get_file_path_for_db_index,
+};
+use serde::{de::DeserializeOwned, Serialize};
+use std::{
+    io,
+    path::{Path, PathBuf},
+};
+
+#[derive(Debug, Eq, PartialEq)]
+struct BPlusTreeArtifactPaths {
+    database: PathBuf,
+    index: PathBuf,
+    wal: PathBuf,
+    wal_temporary: PathBuf,
+    sidecar_lock: PathBuf,
+}
+
+impl BPlusTreeArtifactPaths {
+    fn for_database(database: &Path) -> Self {
+        Self {
+            database: database.to_path_buf(),
+            index: get_file_path_for_db_index(database),
+            wal: wal_path(database),
+            wal_temporary: wal_temporary_path(database),
+            sidecar_lock: sidecar_lock_path(database),
+        }
+    }
+
+    fn remove_all(&self) -> io::Result<()> {
+        let mut first_error = None;
+        for path in self.paths() {
+            if let Err(error) = remove_file_if_exists(path) {
+                first_error.get_or_insert(error);
+            }
+        }
+        first_error.map_or(Ok(()), Err)
+    }
+
+    fn paths(&self) -> [&Path; 5] {
+        [&self.database, &self.index, &self.wal, &self.wal_temporary, &self.sidecar_lock]
+    }
+
+    fn ensure_disjoint_from(&self, published: &Self) -> io::Result<()> {
+        let mut published_identities = Vec::with_capacity(5);
+        for path in published.paths() {
+            published_identities.push((path, resolved_path_identity(path)?));
+        }
+        for staging_path in self.paths() {
+            let staging_identity = resolved_path_identity(staging_path)?;
+            if let Some((published_path, _)) = published_identities
+                .iter()
+                .find(|(_, published_identity)| published_identity.as_path() == staging_identity.as_path())
+            {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    format!(
+                        "staging artifact {} aliases published artifact {} at {}",
+                        staging_path.display(),
+                        published_path.display(),
+                        staging_identity.display()
+                    ),
+                ));
+            }
+        }
+        Ok(())
+    }
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub(crate) struct BPlusTreeStagingArtifacts(BPlusTreeArtifactPaths);
+
+impl BPlusTreeStagingArtifacts {
+    pub(crate) fn new(published: &Path, staging: &Path) -> io::Result<Self> {
+        ensure_distinct_sidecar_lock_domains(published, staging)?;
+        let published_artifacts = BPlusTreeArtifactPaths::for_database(published);
+        let staging_artifacts = BPlusTreeArtifactPaths::for_database(staging);
+        staging_artifacts.ensure_disjoint_from(&published_artifacts)?;
+        Ok(Self(staging_artifacts))
+    }
+
+    pub(crate) fn remove_owned_staging_artifacts(&self) -> io::Result<()> { self.0.remove_all() }
+
+    #[cfg(test)]
+    pub(crate) fn owned_paths(&self) -> [&Path; 5] {
+        [
+            &self.0.database,
+            &self.0.index,
+            &self.0.wal,
+            &self.0.wal_temporary,
+            &self.0.sidecar_lock,
+        ]
+    }
+}
+
+fn publish_staged_database_inner<K, V>(
+    staging: &Path,
+    published: &Path,
+    acquire_final_lock: impl FnOnce(&Path) -> io::Result<ExclusiveSidecarGuard>,
+    sync_published_directory: impl FnOnce(&Path) -> io::Result<()>,
+) -> io::Result<()>
+where
+    K: Ord + Serialize + DeserializeOwned + Clone,
+    V: Serialize + DeserializeOwned,
+{
+    let staging_tree = BPlusTreeUpdate::<K, V>::try_new(staging).map_err(|error| {
+        io::Error::new(
+            error.kind(),
+            format!("failed to verify staging B+Tree {} before publish: {error}", staging.display()),
+        )
+    })?;
+    drop(staging_tree);
+
+    let final_guard = acquire_final_lock(published).map_err(|error| {
+        io::Error::new(
+            error.kind(),
+            format!("failed to acquire published B+Tree sidecar lock for {}: {error}", published.display()),
+        )
+    })?;
+    recover_pending_under_existing_lock(published).map_err(|error| {
+        io::Error::new(
+            error.kind(),
+            format!("failed to recover published B+Tree {} before replacement: {error}", published.display()),
+        )
+    })?;
+    publish_database_and_invalidate_sorted_index(staging, published, sync_published_directory).map_err(|error| {
+        io::Error::new(
+            error.kind(),
+            format!(
+                "failed to publish staging B+Tree {} as {}: {error}",
+                staging.display(),
+                published.display()
+            ),
+        )
+    })?;
+    drop(final_guard);
+    Ok(())
+}
+
+fn publish_staged_database_with_lock_acquirer<K, V>(
+    staging: &Path,
+    published: &Path,
+    acquire_final_lock: impl FnOnce(&Path) -> io::Result<ExclusiveSidecarGuard>,
+    sync_published_directory: impl FnOnce(&Path) -> io::Result<()>,
+) -> io::Result<()>
+where
+    K: Ord + Serialize + DeserializeOwned + Clone,
+    V: Serialize + DeserializeOwned,
+{
+    let staging_artifacts = BPlusTreeStagingArtifacts::new(published, staging)?;
+    require_same_parent_directory(staging, published)?;
+    let publish_result = publish_staged_database_inner::<K, V>(
+        staging,
+        published,
+        acquire_final_lock,
+        sync_published_directory,
+    );
+    let cleanup_result = staging_artifacts.remove_owned_staging_artifacts();
+    match (publish_result, cleanup_result) {
+        (Ok(()), Ok(())) => Ok(()),
+        (Ok(()), Err(cleanup_error)) => Err(io::Error::new(
+            cleanup_error.kind(),
+            format!(
+                "B+Tree {} was published, but staging cleanup failed: {cleanup_error}",
+                published.display()
+            ),
+        )),
+        (Err(publish_error), Ok(())) => Err(publish_error),
+        (Err(publish_error), Err(cleanup_error)) => Err(io::Error::new(
+            publish_error.kind(),
+            format!("{publish_error}; staging cleanup also failed: {cleanup_error}"),
+        )),
+    }
+}
+
+pub(crate) fn publish_staged_database<K, V>(staging: &Path, published: &Path) -> io::Result<()>
+where
+    K: Ord + Serialize + DeserializeOwned + Clone,
+    V: Serialize + DeserializeOwned,
+{
+    publish_staged_database_with_lock_acquirer::<K, V>(
+        staging,
+        published,
+        ExclusiveSidecarGuard::acquire,
+        sync_parent_directory,
+    )
+}
+
+#[cfg(test)]
+fn publish_staged_database_after_observed_final_lock_contention<K, V>(
+    staging: &Path,
+    published: &Path,
+    on_contention: impl FnOnce() -> io::Result<()>,
+) -> io::Result<()>
+where
+    K: Ord + Serialize + DeserializeOwned + Clone,
+    V: Serialize + DeserializeOwned,
+{
+    publish_staged_database_with_lock_acquirer::<K, V>(
+        staging,
+        published,
+        |database| ExclusiveSidecarGuard::acquire_after_observed_contention(database, on_contention),
+        sync_parent_directory,
+    )
+}
+
+#[cfg(test)]
+fn publish_staged_database_with_directory_sync<K, V>(
+    staging: &Path,
+    published: &Path,
+    sync_published_directory: impl FnOnce(&Path) -> io::Result<()>,
+) -> io::Result<()>
+where
+    K: Ord + Serialize + DeserializeOwned + Clone,
+    V: Serialize + DeserializeOwned,
+{
+    publish_staged_database_with_lock_acquirer::<K, V>(
+        staging,
+        published,
+        ExclusiveSidecarGuard::acquire,
+        sync_published_directory,
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        publish_staged_database, publish_staged_database_after_observed_final_lock_contention,
+        publish_staged_database_with_directory_sync, BPlusTreeArtifactPaths, BPlusTreeStagingArtifacts,
+    };
+    use super::super::wal::{leave_uncommitted_test_wal_after_database_write, wal_path};
+    use crate::repository::{get_file_path_for_db_index, BPlusTree, BPlusTreeError, BPlusTreeQuery};
+    use std::{
+        env, fs, io,
+        process::{Child, Command, ExitStatus, Stdio},
+        thread,
+        time::{Duration, Instant},
+    };
+
+    struct KillChildOnDrop(Option<Child>);
+
+    impl KillChildOnDrop {
+        fn spawn(command: &mut Command) -> io::Result<Self> { command.spawn().map(|child| Self(Some(child))) }
+
+        fn try_wait(&mut self) -> io::Result<Option<ExitStatus>> {
+            self.0.as_mut().map_or(Ok(None), Child::try_wait)
+        }
+
+        fn wait_until(&mut self, deadline: Instant) -> io::Result<ExitStatus> {
+            loop {
+                if let Some(status) = self.try_wait()? {
+                    self.0 = None;
+                    return Ok(status);
+                }
+                if Instant::now() >= deadline {
+                    return Err(io::Error::new(io::ErrorKind::TimedOut, "publish child did not finish"));
+                }
+                thread::sleep(Duration::from_millis(10));
+            }
+        }
+    }
+
+    impl Drop for KillChildOnDrop {
+        fn drop(&mut self) {
+            if let Some(mut child) = self.0.take() {
+                if child.try_wait().ok().flatten().is_none() {
+                    let _ = child.kill();
+                }
+                let _ = child.wait();
+            }
+        }
+    }
+
+    fn wait_for_path(path: &std::path::Path, deadline: Instant) -> io::Result<()> {
+        while !path.exists() {
+            if Instant::now() >= deadline {
+                return Err(io::Error::new(io::ErrorKind::TimedOut, "publish child did not reach lock boundary"));
+            }
+            thread::sleep(Duration::from_millis(10));
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn publish_staged_database_child() -> io::Result<()> {
+        let Some(staging) = env::var_os("TULIPROX_BPLUS_PUBLISH_STAGING") else {
+            return Ok(());
+        };
+        let published = env::var_os("TULIPROX_BPLUS_PUBLISH_FINAL")
+            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing published path"))?;
+        let marker = env::var_os("TULIPROX_BPLUS_PUBLISH_MARKER")
+            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing marker path"))?;
+        publish_staged_database_after_observed_final_lock_contention::<u32, String>(
+            std::path::Path::new(&staging),
+            std::path::Path::new(&published),
+            || fs::write(marker, b"contended"),
+        )
+    }
+
+    #[test]
+    fn publish_waits_for_final_reader_then_replaces_database_and_cleans_staging() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let published = directory.path().join("live.db");
+        let staging = directory.path().join("live.refresh-fixed.db");
+        let marker = directory.path().join("publish.final-lock-contended");
+
+        let mut final_tree = BPlusTree::new();
+        final_tree.insert(1u32, String::from("published"));
+        final_tree.store(&published)?;
+        fs::write(get_file_path_for_db_index(&published), b"stale index")?;
+
+        let mut staged_tree = BPlusTree::new();
+        staged_tree.insert(2u32, String::from("staged"));
+        staged_tree.store(&staging)?;
+        let staging_artifacts = BPlusTreeArtifactPaths::for_database(&staging);
+        fs::write(&staging_artifacts.wal_temporary, b"abandoned")?;
+        fs::write(&staging_artifacts.index, b"staging index")?;
+
+        let mut final_query = BPlusTreeQuery::<u32, String>::try_new(&published)?;
+        let published_artifacts = BPlusTreeArtifactPaths::for_database(&published);
+        fs::write(&published_artifacts.wal_temporary, b"abandoned final WAL staging file")?;
+        let mut child = KillChildOnDrop::spawn(
+            Command::new(env::current_exe()?)
+                .arg("--exact")
+                .arg("repository::bplustree::v3::publish::tests::publish_staged_database_child")
+                .arg("--nocapture")
+                .env("TULIPROX_BPLUS_PUBLISH_STAGING", &staging)
+                .env("TULIPROX_BPLUS_PUBLISH_FINAL", &published)
+                .env("TULIPROX_BPLUS_PUBLISH_MARKER", &marker)
+                .stdout(Stdio::null())
+                .stderr(Stdio::null()),
+        )?;
+        wait_for_path(&marker, Instant::now() + Duration::from_secs(5))?;
+        assert_eq!(
+            final_query.query(&1).map_err(BPlusTreeError::to_io)?,
+            Some(String::from("published"))
+        );
+        assert!(staging.exists(), "staging database must exist while the final lock is contended");
+        assert!(child.try_wait()?.is_none(), "publish must wait for the final shared sidecar lock");
+
+        drop(final_query);
+        let status = child.wait_until(Instant::now() + Duration::from_secs(5))?;
+        assert!(status.success(), "publish child failed with {status}");
+
+        let mut query = BPlusTreeQuery::<u32, String>::try_new(&published)?;
+        assert_eq!(query.query(&1).map_err(BPlusTreeError::to_io)?, None);
+        assert_eq!(query.query(&2).map_err(BPlusTreeError::to_io)?, Some(String::from("staged")));
+        assert!(!get_file_path_for_db_index(&published).exists());
+        assert!(!published_artifacts.wal_temporary.exists());
+        for artifact in [
+            staging_artifacts.database,
+            staging_artifacts.index,
+            staging_artifacts.wal,
+            staging_artifacts.wal_temporary,
+            staging_artifacts.sidecar_lock,
+        ] {
+            assert!(!artifact.exists(), "staging artifact survived: {}", artifact.display());
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn publish_recovers_active_final_wal_before_replacing_database() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let published = directory.path().join("live.db");
+        let staging = directory.path().join("live.refresh-wal.db");
+
+        let mut published_tree = BPlusTree::new();
+        published_tree.insert(1u32, String::from("published"));
+        published_tree.store(&published)?;
+        let mut staging_tree = BPlusTree::new();
+        staging_tree.insert(2u32, String::from("staged"));
+        staging_tree.store(&staging)?;
+
+        leave_uncommitted_test_wal_after_database_write(&published)?;
+        assert!(wal_path(&published).exists(), "test setup must leave an active final WAL");
+
+        publish_staged_database::<u32, String>(&staging, &published)?;
+
+        assert!(!wal_path(&published).exists(), "active final WAL must be recovered before publish");
+        let mut query = BPlusTreeQuery::<u32, String>::try_new(&published)?;
+        assert_eq!(query.query(&1).map_err(BPlusTreeError::to_io)?, None);
+        assert_eq!(query.query(&2).map_err(BPlusTreeError::to_io)?, Some(String::from("staged")));
+        Ok(())
+    }
+
+    #[test]
+    fn sequential_staged_publications_leave_only_the_final_lock_domain() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let published = directory.path().join("live.db");
+        let first_staging = directory.path().join("live.refresh-first.db");
+        let second_staging = directory.path().join("live.refresh-second.db");
+
+        let mut published_tree = BPlusTree::new();
+        published_tree.insert(1u32, String::from("published"));
+        published_tree.store(&published)?;
+        let mut first_tree = BPlusTree::new();
+        first_tree.insert(2u32, String::from("first"));
+        first_tree.store(&first_staging)?;
+        publish_staged_database::<u32, String>(&first_staging, &published)?;
+
+        let mut first_query = BPlusTreeQuery::<u32, String>::try_new(&published)?;
+        assert_eq!(
+            first_query.query(&2).map_err(BPlusTreeError::to_io)?,
+            Some(String::from("first"))
+        );
+        drop(first_query);
+
+        let mut second_tree = BPlusTree::new();
+        second_tree.insert(3u32, String::from("second"));
+        second_tree.store(&second_staging)?;
+        publish_staged_database::<u32, String>(&second_staging, &published)?;
+
+        let mut second_query = BPlusTreeQuery::<u32, String>::try_new(&published)?;
+        assert_eq!(second_query.query(&2).map_err(BPlusTreeError::to_io)?, None);
+        assert_eq!(
+            second_query.query(&3).map_err(BPlusTreeError::to_io)?,
+            Some(String::from("second"))
+        );
+        assert!(BPlusTreeArtifactPaths::for_database(&published).sidecar_lock.exists());
+        assert!(!BPlusTreeArtifactPaths::for_database(&first_staging).sidecar_lock.exists());
+        assert!(!BPlusTreeArtifactPaths::for_database(&second_staging).sidecar_lock.exists());
+        Ok(())
+    }
+
+    #[test]
+    fn colliding_publish_path_fails_without_cleaning_published_artifacts() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let published = directory.path().join("live.db");
+        let colliding_staging = directory.path().join("live.tmp");
+        let published_artifacts = BPlusTreeArtifactPaths::for_database(&published);
+
+        let mut published_tree = BPlusTree::new();
+        published_tree.insert(1u32, String::from("published"));
+        published_tree.store(&published)?;
+        fs::write(&published_artifacts.index, b"published index")?;
+        fs::write(&colliding_staging, b"unverified staging")?;
+
+        let error = publish_staged_database::<u32, String>(&colliding_staging, &published)
+            .expect_err("colliding sidecar domains must fail before cleanup");
+        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
+        assert!(published.exists());
+        assert!(published_artifacts.index.exists());
+        assert!(published_artifacts.sidecar_lock.exists());
+        assert!(colliding_staging.exists());
+        Ok(())
+    }
+
+    #[test]
+    fn staging_artifacts_reject_a_published_lock_domain_alias() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let published = directory.path().join("live.db");
+        let colliding_staging = directory.path().join("live.tmp");
+
+        let error = BPlusTreeStagingArtifacts::new(&published, &colliding_staging)
+            .expect_err("staging artifacts must not represent the published lock domain");
+
+        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
+        Ok(())
+    }
+
+    #[test]
+    fn staging_artifacts_reject_parent_component_alias() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let nested = directory.path().join("nested");
+        fs::create_dir(&nested)?;
+        let published = directory.path().join("live.db");
+        let aliased_staging = nested.join("..").join("live.db");
+        fs::write(&published, b"published database")?;
+
+        let error = BPlusTreeStagingArtifacts::new(&published, &aliased_staging)
+            .expect_err("resolved parent aliases must not receive staging ownership");
+
+        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
+        assert_eq!(fs::read(&published)?, b"published database");
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn staging_artifacts_reject_symlinked_parent_alias() -> io::Result<()> {
+        use std::os::unix::fs::symlink;
+
+        let directory = tempfile::tempdir()?;
+        let published_parent = directory.path().join("published");
+        let staging_parent_alias = directory.path().join("staging-alias");
+        fs::create_dir(&published_parent)?;
+        symlink(&published_parent, &staging_parent_alias)?;
+        let published = published_parent.join("live.db");
+        let aliased_staging = staging_parent_alias.join("live.db");
+
+        let error = BPlusTreeStagingArtifacts::new(&published, &aliased_staging)
+            .expect_err("symlinked parent aliases must not receive staging ownership");
+
+        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
+        Ok(())
+    }
+
+    #[test]
+    fn staging_artifacts_reject_cross_artifact_alias() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let published = directory.path().join("live.db");
+        let staging_database = wal_path(&published);
+        fs::write(&staging_database, b"published WAL")?;
+
+        let error = BPlusTreeStagingArtifacts::new(&published, &staging_database)
+            .expect_err("staging database must not alias a published WAL artifact");
+
+        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
+        assert_eq!(fs::read(&staging_database)?, b"published WAL");
+        Ok(())
+    }
+
+    #[test]
+    fn post_rename_sync_failure_still_invalidates_published_index() -> io::Result<()> {
+        let directory = tempfile::tempdir()?;
+        let published = directory.path().join("live.db");
+        let staging = directory.path().join("live.refresh-sync-failure.db");
+        let published_index = get_file_path_for_db_index(&published);
+
+        let mut published_tree = BPlusTree::new();
+        published_tree.insert(1u32, String::from("published"));
+        published_tree.store(&published)?;
+        fs::write(&published_index, b"stale sorted index")?;
+        let mut staging_tree = BPlusTree::new();
+        staging_tree.insert(2u32, String::from("staged"));
+        staging_tree.store(&staging)?;
+
+        let error = publish_staged_database_with_directory_sync::<u32, String>(
+            &staging,
+            &published,
+            |_| Err(io::Error::other("injected post-rename directory sync failure")),
+        )
+        .expect_err("post-rename sync failure must remain visible");
+
+        assert!(error.to_string().contains("publication durability remains unknown"));
+        assert!(!published_index.exists(), "old sorted index must be invalidated after rename");
+        assert!(!staging.exists(), "renamed staging database must not reappear during cleanup");
+        let mut query = BPlusTreeQuery::<u32, String>::try_new(&published)?;
+        assert_eq!(query.query(&1).map_err(BPlusTreeError::to_io)?, None);
+        assert_eq!(query.query(&2).map_err(BPlusTreeError::to_io)?, Some(String::from("staged")));
+        Ok(())
+    }
+}

backend/src/repository/bplustree/v3/tree.rs

@@ -684,27 +684,90 @@ fn temporary_path(filepath: &Path) -> io::Result<PathBuf> {
     Ok(filepath.with_file_name(format!("{name}.{}.v3.tmp", uuid::Uuid::new_v4())))
 }
 
-fn publish_database(
+#[derive(Debug)]
+pub(super) enum PublishDatabaseError {
+    NotPublished(io::Error),
+    PublishedDurabilityUnknown(io::Error),
+}
+
+impl PublishDatabaseError {
+    pub(super) const fn database_was_published(&self) -> bool {
+        matches!(self, Self::PublishedDurabilityUnknown(_))
+    }
+
+    fn into_io(self) -> io::Error {
+        match self {
+            Self::NotPublished(error) | Self::PublishedDurabilityUnknown(error) => error,
+        }
+    }
+}
+
+impl From<PublishDatabaseError> for io::Error {
+    fn from(error: PublishDatabaseError) -> Self { error.into_io() }
+}
+
+pub(super) fn publish_database(
     temporary: &Path,
     destination: &Path,
     sync_directory: impl FnOnce(&Path) -> io::Result<()>,
-) -> io::Result<()> {
+) -> Result<(), PublishDatabaseError> {
     let temporary = match tempfile::TempPath::try_from_path(temporary) {
         Ok(path) => path,
         Err(error) => {
             let _ = std::fs::remove_file(temporary);
-            return Err(error);
+            return Err(PublishDatabaseError::NotPublished(error));
         }
     };
-    temporary.persist(destination).map_err(io::Error::from)?;
+    temporary
+        .persist(destination)
+        .map_err(io::Error::from)
+        .map_err(PublishDatabaseError::NotPublished)?;
     sync_directory(destination).map_err(|error| {
-        io::Error::new(
+        PublishDatabaseError::PublishedDurabilityUnknown(io::Error::new(
             error.kind(),
             format!("database published but directory sync failed; durability unknown: {error}"),
-        )
+        ))
     })
 }
 
+/// Invalidates the sorted index whenever publication made the replacement database visible.
+pub(super) fn publish_database_and_invalidate_sorted_index(
+    temporary: &Path,
+    destination: &Path,
+    sync_directory: impl FnOnce(&Path) -> io::Result<()>,
+) -> io::Result<()> {
+    let publish_error = match publish_database(temporary, destination, sync_directory) {
+        Ok(()) => None,
+        Err(error) if error.database_was_published() => Some(io::Error::from(error)),
+        Err(error) => return Err(io::Error::from(error)),
+    };
+    let invalidation_error = invalidate_sorted_index(destination).err();
+    match (publish_error, invalidation_error) {
+        (Some(publish_error), Some(invalidation_error)) => Err(io::Error::new(
+            publish_error.kind(),
+            format!(
+                "database {} was published with unknown directory durability: {publish_error}; its previous sorted index could not be invalidated: {invalidation_error}",
+                destination.display()
+            ),
+        )),
+        (Some(publish_error), None) => Err(io::Error::new(
+            publish_error.kind(),
+            format!(
+                "database {} was published and its previous sorted index was invalidated, but publication durability remains unknown: {publish_error}",
+                destination.display()
+            ),
+        )),
+        (None, Some(error)) => Err(io::Error::new(
+            error.kind(),
+            format!(
+                "database {} was published, but its previous sorted index could not be invalidated: {error}",
+                destination.display()
+            ),
+        )),
+        (None, None) => Ok(()),
+    }
+}
+
 impl<K, V> BPlusTree<K, V>
 where
     K: Ord + Serialize + for<'de> Deserialize<'de> + Clone,
@@ -732,6 +795,14 @@ where
     }
 
     fn store_exclusive(&mut self, filepath: &Path) -> io::Result<StoredDatabase> {
+        self.store_exclusive_with_directory_sync(filepath, sync_parent_directory)
+    }
+
+    fn store_exclusive_with_directory_sync(
+        &mut self,
+        filepath: &Path,
+        sync_published_directory: impl FnOnce(&Path) -> io::Result<()>,
+    ) -> io::Result<StoredDatabase> {
         let temp_path = temporary_path(filepath)?;
         // The builder streams into the temp file, so it exists from here on — every error
         // path below has to remove it again.
@@ -762,8 +833,7 @@ where
                 return Err(error);
             }
         };
-        publish_database(&temp_path, filepath, sync_parent_directory)?;
-        invalidate_sorted_index(filepath)?;
+        publish_database_and_invalidate_sorted_index(&temp_path, filepath, sync_published_directory)?;
         self.dirty = false;
         Ok(StoredDatabase { root_page_id, verification })
     }
@@ -849,7 +919,7 @@ where
             let _ = std::fs::remove_file(&temporary);
             return Err(error);
         }
-        publish_database(&temporary, &index_path, sync_parent_directory)
+        publish_database(&temporary, &index_path, sync_parent_directory).map_err(io::Error::from)
     }
 
 }
@@ -5886,6 +5956,8 @@ mod tests {
         })
         .err()
         .ok_or_else(|| io::Error::other("post-commit sync failure was hidden"))?;
+        assert!(error.database_was_published());
+        let error = io::Error::from(error);
         assert!(error.to_string().contains("database published but directory sync failed; durability unknown"));
         assert!(!temporary.exists());
         let mut query = BPlusTreeQuery::<u32, Vec<u8>>::try_new(&destination)?;
@@ -5894,6 +5966,64 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn store_sync_failure_after_rename_invalidates_previous_sorted_index() -> io::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let destination = dir.path().join("database.db");
+        let index = crate::repository::storage::get_file_path_for_db_index(&destination);
+        let mut old = BPlusTree::new();
+        old.insert(1u32, String::from("old"));
+        let _ = old.store_with_index(&destination, String::clone)?;
+        assert!(index.is_file());
+
+        let mut replacement = BPlusTree::new();
+        replacement.insert(2u32, String::from("new"));
+        let error = replacement
+            .store_exclusive_with_directory_sync(&destination, |_| {
+                Err(io::Error::other("injected directory sync failure"))
+            })
+            .err()
+            .ok_or_else(|| io::Error::other("post-rename sync failure was hidden"))?;
+
+        assert!(error.to_string().contains("publication durability remains unknown"));
+        assert!(!index.exists());
+        assert!(replacement.dirty);
+        let mut query = BPlusTreeQuery::<u32, String>::try_new(&destination)?;
+        assert_eq!(query.query(&1).map_err(BPlusTreeError::to_io)?, None);
+        assert_eq!(query.query(&2).map_err(BPlusTreeError::to_io)?, Some(String::from("new")));
+        Ok(())
+    }
+
+    #[test]
+    fn store_reports_sync_and_index_invalidation_failures_together() -> io::Result<()> {
+        let dir = tempfile::tempdir()?;
+        let destination = dir.path().join("database.db");
+        let index = crate::repository::storage::get_file_path_for_db_index(&destination);
+        let mut old = BPlusTree::new();
+        old.insert(1u32, String::from("old"));
+        let _ = old.store(&destination)?;
+        fs::create_dir(&index)?;
+
+        let mut replacement = BPlusTree::new();
+        replacement.insert(2u32, String::from("new"));
+        let error = replacement
+            .store_exclusive_with_directory_sync(&destination, |_| {
+                Err(io::Error::other("injected directory sync failure"))
+            })
+            .err()
+            .ok_or_else(|| io::Error::other("publish and index invalidation failures were hidden"))?;
+        let message = error.to_string();
+
+        assert!(message.contains("unknown directory durability"));
+        assert!(message.contains("previous sorted index could not be invalidated"));
+        assert!(index.is_dir());
+        assert!(replacement.dirty);
+        let mut query = BPlusTreeQuery::<u32, String>::try_new(&destination)?;
+        assert_eq!(query.query(&1).map_err(BPlusTreeError::to_io)?, None);
+        assert_eq!(query.query(&2).map_err(BPlusTreeError::to_io)?, Some(String::from("new")));
+        Ok(())
+    }
+
     #[test]
     fn publish_failure_removes_the_temporary_file() -> io::Result<()> {
         let dir = tempfile::tempdir()?;
@@ -5902,7 +6032,9 @@ mod tests {
         let temporary = dir.path().join("database.tx.v3.tmp");
         fs::write(&temporary, b"new")?;
 
-        assert!(publish_database(&temporary, &destination, sync_parent_directory).is_err());
+        let error = publish_database(&temporary, &destination, sync_parent_directory)
+            .expect_err("publishing over a directory must fail");
+        assert!(!error.database_was_published());
         assert!(!temporary.exists());
         Ok(())
     }

backend/src/repository/bplustree/v3/wal.rs

@@ -762,6 +762,43 @@ fn commit_prepared_pages_with_hook(
     })
 }
 
+#[cfg(test)]
+pub(super) fn leave_uncommitted_test_wal_after_database_write(database: &Path) -> io::Result<()> {
+    let mut header_page = [0u8; PAGE_SIZE];
+    File::open(database)?.read_exact(&mut header_page)?;
+    let mut updated_header = DatabaseHeader::decode(&header_page)?;
+    updated_header.generation = updated_header
+        .generation
+        .checked_add(1)
+        .ok_or_else(|| io::Error::other("test database generation overflow"))?;
+    let prepared = [(0, updated_header.encode()?)];
+    let commit_result = commit_prepared_pages_with_hook(database, &prepared, |boundary| {
+        if boundary == CommitBoundary::DatabaseWritten {
+            Err(io::Error::other("test fault after database write"))
+        } else {
+            Ok(())
+        }
+    });
+    match commit_result {
+        Err(error) if !error.to_string().contains("test fault after database write") => Err(io::Error::new(
+            error.kind(),
+            format!("test commit for {} failed before the requested boundary: {error}", database.display()),
+        )),
+        Err(error) if !wal_path(database).try_exists()? => Err(io::Error::new(
+            error.kind(),
+            format!(
+                "test commit for {} reached the requested boundary without leaving an active WAL: {error}",
+                database.display()
+            ),
+        )),
+        Err(_) => Ok(()),
+        Ok(()) => Err(io::Error::other(format!(
+            "test commit for {} unexpectedly completed",
+            database.display()
+        ))),
+    }
+}
+
 #[cfg(test)]
 fn commit_prepared_pages_with_hook_under_existing_lock(
     database_path: &Path,
@@ -947,6 +984,26 @@ impl ExclusiveSidecarGuard {
         file.lock_exclusive()?;
         Ok(Self { _file: file })
     }
+
+    #[cfg(test)]
+    pub(super) fn acquire_after_observed_contention(
+        database: &Path,
+        on_contention: impl FnOnce() -> io::Result<()>,
+    ) -> io::Result<Self> {
+        let file = open_sidecar(database)?;
+        match file.try_lock_exclusive() {
+            Ok(()) => Err(io::Error::other(format!(
+                "expected final B+Tree sidecar lock contention for {}",
+                database.display()
+            ))),
+            Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                on_contention()?;
+                file.lock_exclusive()?;
+                Ok(Self { _file: file })
+            }
+            Err(error) => Err(error),
+        }
+    }
 }
 
 pub(crate) fn with_exclusive_sidecar<T>(database: &Path, operation: impl FnOnce() -> io::Result<T>) -> io::Result<T> {

backend/src/repository/storage.rs

@@ -1,9 +1,11 @@
 use crate::model::Config;
+use crate::repository::bplustree::common::sidecar_lock_path;
 use crate::repository::storage_const;
 use crate::utils;
+use fs2::FileExt;
+use shared::concat_string;
 use shared::error::TuliproxError;
 use std::path::{Path, PathBuf};
-use shared::concat_string;
 
 pub fn get_target_id_mapping_file(target_path: &Path) -> PathBuf {
     // Join directly with &str to avoid an intermediate PathBuf allocation
@@ -80,12 +82,140 @@ pub async fn get_input_storage_path(input_name: &str, storage_dir: &str) -> std:
 }
 
 pub async fn ensure_input_storage_path(cfg: &Config, input_name: &str) -> Result<PathBuf, TuliproxError> {
-    get_input_storage_path(input_name, &cfg.storage_dir).await
+    let path = get_input_storage_path(input_name, &cfg.storage_dir).await
         .map_err(|err| {
             TuliproxError::RepositoryStorage(format!("Failed to save input data, can't create directory for input {input_name}: {err}"))
-        })
+        })?;
+    cleanup_orphaned_staging_artifacts(&path, ORPHAN_STAGING_MIN_AGE);
+    Ok(path)
 }
 
+/// Age threshold below which a `.refresh-<uuid>.<ext>` file is treated as
+/// still in flight and left alone. Refreshes on large Xtream providers can
+/// run 5–15 minutes, so the threshold is set to 30 minutes — comfortably
+/// above the slowest legitimate refresh, while still bounding stale-file
+/// disk usage to roughly half an hour after a crash.
+#[allow(clippy::duration_suboptimal_units)]
+const ORPHAN_STAGING_MIN_AGE: std::time::Duration = std::time::Duration::from_secs(30 * 60);
+
+/// Removes staging files left behind by aborted refreshes (`.refresh-<uuid>.<ext>`).
+///
+/// Best-effort: only the leaf filename pattern and an mtime check decide what
+/// to delete, never the file contents, so a successfully published file is
+/// never deleted and an in-flight parallel refresh is never interrupted.
+/// Errors are logged and otherwise ignored — this is a cleanup helper, not a
+/// guard.
+pub(crate) fn cleanup_orphaned_staging_artifacts(storage_path: &Path, min_age: std::time::Duration) {
+    let now = std::time::SystemTime::now();
+    let entries = match std::fs::read_dir(storage_path) {
+        Ok(entries) => entries,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return,
+        Err(error) => {
+            log::warn!(
+                "Failed to scan storage path {} for orphaned refresh artifacts: {error}",
+                storage_path.display()
+            );
+            return;
+        }
+    };
+    for entry in entries.flatten() {
+        let leaf = entry.file_name();
+        let Some(name) = leaf.to_str() else { continue };
+        if !is_orphan_staging_name(name) {
+            continue;
+        }
+        let metadata = match entry.metadata() {
+            Ok(metadata) => metadata,
+            Err(error) => {
+                log::warn!(
+                    "Failed to stat refresh artifact {}: {error}",
+                    entry.path().display()
+                );
+                continue;
+            }
+        };
+        let age = match metadata.modified() {
+            Ok(modified) => now.duration_since(modified).unwrap_or_default(),
+            Err(error) => {
+                log::warn!(
+                    "Failed to read mtime of refresh artifact {}: {error}",
+                    entry.path().display()
+                );
+                continue;
+            }
+        };
+        if age < min_age {
+            log::debug!(
+                "Skipping recent refresh artifact {} (age {:?} < {:?})",
+                entry.path().display(),
+                age,
+                min_age
+            );
+            continue;
+        }
+        // Cross-process safety: an active refresh holds an exclusive `flock` on the
+        // staging sidecar lock file. A non-blocking attempt that fails means another
+        // process (or a still-active older lease in this process) owns the artifact,
+        // even if the mtime looks stale. The lock handle is held through `remove_file`
+        // so a new refresh cannot acquire the sidecar and start writing before the
+        // deletion lands. We never create or truncate the sidecar: a missing sidecar
+        // means nothing owns the artifact, and creating it here would leave a stale
+        // `.lock` behind for every probed orphan.
+        let lock_path = sidecar_lock_path(&entry.path());
+        let lock_file = match std::fs::OpenOptions::new().read(true).write(true).open(&lock_path) {
+            Ok(file) => Some(file),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+            Err(error) => {
+                log::debug!(
+                    "Skipping refresh artifact {}: sidecar lock probe failed: {error}",
+                    entry.path().display()
+                );
+                continue;
+            }
+        };
+        if let Some(lock_file) = lock_file {
+            if lock_file.try_lock_exclusive().is_err() {
+                log::debug!(
+                    "Skipping refresh artifact {} owned by an active generation",
+                    entry.path().display()
+                );
+                continue;
+            }
+        }
+        let remove_result = std::fs::remove_file(entry.path());
+        debug_assert!(
+            !lock_path.exists(),
+            "cleanup must not leave a generated sidecar behind at {}",
+            lock_path.display()
+        );
+        if let Err(error) = remove_result {
+            log::warn!(
+                "Failed to remove orphaned refresh artifact {}: {error}",
+                entry.path().display()
+            );
+        } else {
+            log::info!("Removed orphaned refresh artifact {}", entry.path().display());
+        }
+    }
+}
+
+fn is_orphan_staging_name(leaf: &str) -> bool {
+    // `<stem>.refresh-<uuid>.<ext>`. The UUID is 32 lowercase hex chars via
+    // `Uuid::simple()`; requiring the dot boundary after the 32 hex chars and
+    // lowercase-only hex keeps operators safe from accidental matches on
+    // filenames that happen to contain `.refresh-`.
+    leaf.match_indices(".refresh-").any(|(index, _)| {
+        let after = &leaf[index + ".refresh-".len()..];
+        let uuid_bytes = after.as_bytes();
+        if uuid_bytes.len() <= 33 || uuid_bytes[32] != b'.' {
+            return false;
+        }
+        let uuid = &uuid_bytes[..32];
+        uuid.iter().all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
+    })
+}
+
+
 pub fn get_geoip_path(storage_dir: &str) -> PathBuf {
     Path::new(storage_dir).join("geoip.db")
 }
@@ -93,3 +223,108 @@ pub fn get_geoip_path(storage_dir: &str) -> PathBuf {
 pub fn get_file_path_for_db_index(db_path: &Path) -> PathBuf {
     db_path.with_extension(storage_const::FILE_SUFFIX_INDEX)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{cleanup_orphaned_staging_artifacts, is_orphan_staging_name};
+    use fs2::FileExt;
+    use std::{
+        fs,
+        path::Path,
+        time::{Duration, SystemTime},
+    };
+
+    #[test]
+    fn refresh_artifacts_match_orphan_pattern() {
+        assert!(is_orphan_staging_name("live.refresh-0d8f0d8f0d8f0d8f0d8f0d8f0d8f0d8f.db"));
+        assert!(is_orphan_staging_name("cat_live.refresh-0d8f0d8f0d8f0d8f0d8f0d8f0d8f0d8f.json"));
+        assert!(!is_orphan_staging_name("live.db"));
+        assert!(!is_orphan_staging_name("cat_live.json"));
+        assert!(!is_orphan_staging_name("live.refresh-fixed.db"));
+    }
+
+    #[test]
+    fn orphan_pattern_rejects_missing_extension_boundary() {
+        // 32 hex chars but no dot boundary after them — must not match.
+        assert!(!is_orphan_staging_name("live.refresh-0d8f0d8f0d8f0d8f0d8f0d8f0d8f0d8fdb"));
+        // 32 hex chars plus extra suffix with no extension — must not match.
+        assert!(!is_orphan_staging_name("live.refresh-0d8f0d8f0d8f0d8f0d8f0d8f0d8f0d8fX"));
+        // 32 hex chars plus dot but no extension after the dot — must not match.
+        assert!(!is_orphan_staging_name("live.refresh-0d8f0d8f0d8f0d8f0d8f0d8f0d8f0d8f."));
+    }
+
+    #[test]
+    fn orphan_pattern_rejects_uppercase_hex_identifiers() {
+        // Uppercase hex must be rejected — `Uuid::simple()` always emits lowercase.
+        assert!(!is_orphan_staging_name("live.refresh-0D8F0D8F0D8F0D8F0D8F0D8F0D8F0D8F.db"));
+        assert!(!is_orphan_staging_name("live.refresh-0d8f0d8f0d8f0d8f0d8f0D8F0D8F0D8F.db"));
+    }
+
+    fn touch_with_age(path: &Path, age: Duration) {
+        fs::write(path, b"fixture").expect("write fixture");
+        let mtime = SystemTime::now()
+            .checked_sub(age)
+            .expect("age fits in system time");
+        let file = fs::OpenOptions::new()
+            .write(true)
+            .open(path)
+            .expect("reopen fixture");
+        file.set_modified(mtime).expect("set mtime");
+    }
+
+    #[test]
+    fn cleanup_leaves_recent_staging_alone() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let active = dir.path().join("live.refresh-0d8f0d8f0d8f0d8f0d8f0d8f0d8f0d8f.db");
+        // 5 seconds old — well under the 10-minute threshold, an in-flight refresh
+        // would write here.
+        touch_with_age(&active, Duration::from_secs(5));
+
+        cleanup_orphaned_staging_artifacts(dir.path(), Duration::from_secs(600));
+
+        assert!(active.exists(), "recent staging file must not be deleted by a parallel refresh");
+    }
+
+    #[test]
+    fn cleanup_removes_only_old_refresh_artifacts() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let published = dir.path().join("live.db");
+        let active = dir.path().join("live.refresh-11111111111111111111111111111111.db");
+        let orphan_db = dir.path().join("live.refresh-22222222222222222222222222222222.db");
+        let orphan_cat = dir.path().join("cat_live.refresh-33333333333333333333333333333333.json");
+
+        fs::write(&published, b"kept").expect("write published");
+        touch_with_age(&active, Duration::from_secs(5));
+        touch_with_age(&orphan_db, Duration::from_secs(3600));
+        touch_with_age(&orphan_cat, Duration::from_secs(3600));
+
+        cleanup_orphaned_staging_artifacts(dir.path(), Duration::from_secs(600));
+
+        assert!(published.exists());
+        assert!(active.exists(), "in-flight staging must survive");
+        assert!(!orphan_db.exists());
+        assert!(!orphan_cat.exists());
+    }
+
+    #[test]
+    fn cleanup_skips_active_generation_older_than_min_age() {
+        // An in-flight refresh that exceeds `min_age` because it is unusually slow
+        // must not be reaped: the sidecar lock proves another owner is still alive.
+        let dir = tempfile::tempdir().expect("tempdir");
+        let active = dir.path().join("live.refresh-44444444444444444444444444444444.db");
+        let sidecar = dir.path().join(".live.refresh-44444444444444444444444444444444.lock");
+
+        touch_with_age(&active, Duration::from_secs(3600));
+        let lock_file = fs::OpenOptions::new()
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(&sidecar)
+            .expect("open sidecar");
+        lock_file.lock_exclusive().expect("acquire active lease");
+
+        cleanup_orphaned_staging_artifacts(dir.path(), Duration::from_secs(600));
+
+        assert!(active.exists(), "active generation older than min_age must survive");
+    }
+}