diff --git a/backend/src/repository/bplustree/common.rs b/backend/src/repository/bplustree/common.rs index 53211fd1b..869dfc1ed 100644 --- a/backend/src/repository/bplustree/common.rs +++ b/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 { + 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), diff --git a/backend/src/repository/bplustree/v3/mod.rs b/backend/src/repository/bplustree/v3/mod.rs index d7d772b5e..c80cf7801 100644 --- a/backend/src/repository/bplustree/v3/mod.rs +++ b/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, diff --git a/backend/src/repository/bplustree/v3/publish.rs b/backend/src/repository/bplustree/v3/publish.rs new file mode 100644 index 000000000..bca6448cb --- /dev/null +++ b/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 { + 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( + staging: &Path, + published: &Path, + acquire_final_lock: impl FnOnce(&Path) -> io::Result, + sync_published_directory: impl FnOnce(&Path) -> io::Result<()>, +) -> io::Result<()> +where + K: Ord + Serialize + DeserializeOwned + Clone, + V: Serialize + DeserializeOwned, +{ + let staging_tree = BPlusTreeUpdate::::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( + staging: &Path, + published: &Path, + acquire_final_lock: impl FnOnce(&Path) -> io::Result, + 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::( + 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(staging: &Path, published: &Path) -> io::Result<()> +where + K: Ord + Serialize + DeserializeOwned + Clone, + V: Serialize + DeserializeOwned, +{ + publish_staged_database_with_lock_acquirer::( + staging, + published, + ExclusiveSidecarGuard::acquire, + sync_parent_directory, + ) +} + +#[cfg(test)] +fn publish_staged_database_after_observed_final_lock_contention( + 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::( + staging, + published, + |database| ExclusiveSidecarGuard::acquire_after_observed_contention(database, on_contention), + sync_parent_directory, + ) +} + +#[cfg(test)] +fn publish_staged_database_with_directory_sync( + 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::( + 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); + + impl KillChildOnDrop { + fn spawn(command: &mut Command) -> io::Result { command.spawn().map(|child| Self(Some(child))) } + + fn try_wait(&mut self) -> io::Result> { + self.0.as_mut().map_or(Ok(None), Child::try_wait) + } + + fn wait_until(&mut self, deadline: Instant) -> io::Result { + 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::( + 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::::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::::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::(&staging, &published)?; + + assert!(!wal_path(&published).exists(), "active final WAL must be recovered before publish"); + let mut query = BPlusTreeQuery::::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::(&first_staging, &published)?; + + let mut first_query = BPlusTreeQuery::::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::(&second_staging, &published)?; + + let mut second_query = BPlusTreeQuery::::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::(&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::( + &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::::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(()) + } +} diff --git a/backend/src/repository/bplustree/v3/tree.rs b/backend/src/repository/bplustree/v3/tree.rs index b80cab06b..acc3050de 100644 --- a/backend/src/repository/bplustree/v3/tree.rs +++ b/backend/src/repository/bplustree/v3/tree.rs @@ -684,27 +684,90 @@ fn temporary_path(filepath: &Path) -> io::Result { 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 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 BPlusTree where K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, @@ -732,6 +795,14 @@ where } fn store_exclusive(&mut self, filepath: &Path) -> io::Result { + 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 { 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::>::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::::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::::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(()) } diff --git a/backend/src/repository/bplustree/v3/wal.rs b/backend/src/repository/bplustree/v3/wal.rs index 399840f64..d51ea9a27 100644 --- a/backend/src/repository/bplustree/v3/wal.rs +++ b/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 { + 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(database: &Path, operation: impl FnOnce() -> io::Result) -> io::Result { diff --git a/backend/src/repository/storage.rs b/backend/src/repository/storage.rs index b31ac118b..06297ffdb 100644 --- a/backend/src/repository/storage.rs +++ b/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 { - 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-.` 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-.`). +/// +/// 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 { + // `.refresh-.`. 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"); + } +} diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 86236e1a7..c8d973297 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -4,7 +4,11 @@ use crate::model::{AppConfig, ProxyUserCredentials}; use crate::model::{Config, ConfigTarget}; use crate::model::{ConfigInput, PlaylistXtreamCategory}; use crate::processing::parser::xtream; -use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery, BPlusTreeUpdate, FlushPolicy}; +use crate::repository::bplustree::{ + common::{ensure_distinct_sidecar_lock_domains, remove_file_if_exists}, + v3::{publish_staged_database, BPlusTreeStagingArtifacts}, + BPlusTree, BPlusTreeError, BPlusTreeQuery, BPlusTreeUpdate, FlushPolicy, +}; use crate::repository::open_playlist_reader; use crate::repository::playlist_scratch::PlaylistScratch; use crate::repository::storage::{ensure_input_storage_path, ensure_target_storage_subpath, get_file_path_for_db_index, get_input_storage_path, get_target_id_mapping_file, get_target_storage_path}; @@ -28,12 +32,14 @@ use shared::model::{LiveStreamProperties, PlaylistGroup, PlaylistItem, PlaylistI use shared::utils::{arc_str_serde, get_u32_from_serde_value, Internable}; use shared::concat_string; use std::collections::HashMap; +use std::ffi::OsString; use std::fs::File; -use std::io::{Error, ErrorKind}; +use std::io::{self, Error, ErrorKind}; use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; +use uuid::Uuid; macro_rules! cant_write_result { ($path:expr, $err:expr) => { @@ -779,31 +785,230 @@ pub(crate) async fn xtream_get_playlist_categories(config: &Config, target_name: const BATCH_SIZE: usize = 1000; +#[derive(Debug, Clone, Eq, PartialEq)] +struct XtreamRefreshPaths { + generation: Uuid, + published_database: PathBuf, + staging_database: PathBuf, + published_categories: PathBuf, + staging_categories: PathBuf, +} + +impl XtreamRefreshPaths { + fn new(storage_path: &Path, cluster: XtreamCluster) -> Result { + Self::for_generation(storage_path, cluster, Uuid::new_v4()) + } + + fn for_generation( + storage_path: &Path, + cluster: XtreamCluster, + generation: Uuid, + ) -> Result { + let published_database = xtream_get_file_path(storage_path, cluster); + let published_categories = get_collection_path(storage_path, xtream_cluster_category_collection(cluster)); + let staging_database = refresh_staging_path(&published_database, generation)?; + // The lock-domain check is repeated by `XtreamRefreshLease::new` with a stricter + // aliasing scan; doing it here too would canonicalize the same paths twice. + Ok(Self { + generation, + staging_database, + staging_categories: refresh_staging_path(&published_categories, generation)?, + published_database, + published_categories, + }) + } +} + +fn refresh_staging_path(path: &Path, generation: Uuid) -> Result { + let stem = path.file_stem().ok_or_else(|| { + TuliproxError::RepositoryXtream(format!("Refresh path has no file stem: {}", path.display())) + })?; + let extension = path.extension().ok_or_else(|| { + TuliproxError::RepositoryXtream(format!("Refresh path has no extension: {}", path.display())) + })?; + let mut filename = OsString::from(stem); + filename.push(".refresh-"); + filename.push(generation.simple().to_string()); + filename.push("."); + filename.push(extension); + Ok(path.with_file_name(filename)) +} + +#[derive(Debug, Clone)] +struct XtreamRefreshLease(Arc); + +#[derive(Debug)] +struct XtreamRefreshLeaseInner { + paths: XtreamRefreshPaths, + database_artifacts: BPlusTreeStagingArtifacts, +} + +impl XtreamRefreshLease { + fn new(paths: XtreamRefreshPaths) -> Result { + let database_artifacts = BPlusTreeStagingArtifacts::new( + &paths.published_database, + &paths.staging_database, + ) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Invalid Xtream staging artifacts for generation {}: {error}", + paths.generation + )) + })?; + Ok(Self(Arc::new(XtreamRefreshLeaseInner { paths, database_artifacts }))) + } + + fn paths(&self) -> &XtreamRefreshPaths { &self.0.paths } + + fn cleanup_staging_artifacts(&self) -> io::Result<()> { + let database_result = self.0.database_artifacts.remove_owned_staging_artifacts(); + let categories_result = remove_file_if_exists(&self.0.paths.staging_categories); + match (database_result, categories_result) { + (Ok(()), Ok(())) => Ok(()), + (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error), + (Err(database_error), Err(categories_error)) => Err(io::Error::new( + database_error.kind(), + format!("{database_error}; category staging cleanup also failed: {categories_error}"), + )), + } + } +} + +impl Drop for XtreamRefreshLeaseInner { + fn drop(&mut self) { + let database_result = self.database_artifacts.remove_owned_staging_artifacts(); + let categories_result = remove_file_if_exists(&self.paths.staging_categories); + if let Err(error) = database_result { + log::warn!( + "Failed to clean Xtream staging database artifacts for generation {}: {error}", + self.paths.generation + ); + } + if let Err(error) = categories_result { + log::warn!( + "Failed to clean Xtream staging categories for generation {} at {}: {error}", + self.paths.generation, + self.paths.staging_categories.display() + ); + } + } +} + +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +enum PreserveDetailsOutcome { + SourceMissing, + Merged { scanned: usize, updated: usize }, +} + +#[derive(Debug, Clone, Copy, Eq, PartialEq)] +enum DetailPreservationOperation { + Query, + BatchWrite, + Commit, +} + +fn write_preserved_detail_batch( + staging_tree: &mut BPlusTreeUpdate, + staging_path: &Path, + updates: &mut Vec<(u32, XtreamPlaylistItem)>, + before_operation: &mut F, +) -> Result +where + F: FnMut(DetailPreservationOperation) -> io::Result<()>, +{ + if updates.is_empty() { + return Ok(0); + } + let batch_len = updates.len(); + let refs: Vec<(&u32, &XtreamPlaylistItem)> = updates.iter().map(|(id, item)| (id, item)).collect(); + before_operation(DetailPreservationOperation::BatchWrite) + .and_then(|()| staging_tree.update_batch(&refs).map(|_| ()).map_err(BPlusTreeError::to_io)) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to update staging Xtream tree {} during detail preservation: {error}", + staging_path.display() + )) + })?; + updates.clear(); + Ok(batch_len) +} + fn preserve_details_input_xtream_playlist_cluster_to_disk( - old_path: &Path, - tmp_path: &Path, -) -> Result<(), TuliproxError> { - let Ok(mut old_tree) = BPlusTreeQuery::::try_new(old_path) else { - return Ok(()) + published_path: &Path, + staging_path: &Path, +) -> Result { + preserve_details_input_xtream_playlist_cluster_to_disk_with_hook( + published_path, + staging_path, + |_| Ok(()), + ) +} + +fn preserve_details_input_xtream_playlist_cluster_to_disk_with_hook( + published_path: &Path, + staging_path: &Path, + mut before_operation: F, +) -> Result +where + F: FnMut(DetailPreservationOperation) -> io::Result<()>, +{ + ensure_distinct_sidecar_lock_domains(published_path, staging_path) + .map_err(|error| TuliproxError::RepositoryXtream(error.to_string()))?; + + let mut published_tree = match BPlusTreeQuery::::try_new(published_path) { + Ok(tree) => tree, + Err(error) if error.kind() == io::ErrorKind::NotFound => { + return Ok(PreserveDetailsOutcome::SourceMissing); + } + Err(error) => { + return Err(TuliproxError::RepositoryXtream(format!( + "Failed to open published Xtream tree {} for detail preservation: {error}", + published_path.display() + ))); + } }; - let Ok(mut new_tree) = BPlusTreeUpdate::::try_new_with_backoff(tmp_path) else { - return Ok(()) - }; + let mut staging_tree = BPlusTreeUpdate::::try_new_with_backoff(staging_path) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to open staging Xtream tree {} for detail preservation: {error}", + staging_path.display() + )) + })?; - let mut updates: Vec<(u32, XtreamPlaylistItem)> = Vec::with_capacity(BATCH_SIZE); - for entry in old_tree.iter() { - let (_, old_item) = entry.map_err(|error| TuliproxError::RepositoryXtream(error.to_string()))?; + let mut pending_updates: Vec<(u32, XtreamPlaylistItem)> = Vec::with_capacity(BATCH_SIZE); + let mut scanned_count = 0usize; + let mut updated_count = 0usize; + for entry in published_tree.iter() { + let (_, old_item) = entry.map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to read published Xtream tree {} during detail preservation: {error}", + published_path.display() + )) + })?; + scanned_count = scanned_count.saturating_add(1); if let Some(old_props) = old_item.additional_properties.as_ref() { if old_props.has_details() { - if let Ok(Some(mut new_item)) = new_tree.query(&old_item.provider_id) { + let staging_item = before_operation(DetailPreservationOperation::Query) + .and_then(|()| staging_tree.query(&old_item.provider_id).map_err(BPlusTreeError::to_io)) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to query staging Xtream tree {} for provider {}: {error}", + staging_path.display(), + old_item.provider_id + )) + })?; + if let Some(mut new_item) = staging_item { if let Some(new_props) = new_item.additional_properties.as_mut() { if merge_preserved_stream_properties(new_props, old_props) { - updates.push((new_item.provider_id, new_item)); - if updates.len() >= BATCH_SIZE { - let refs: Vec<(&u32, &XtreamPlaylistItem)> = updates.iter().map(|(id, pli)| (id, pli)).collect(); - new_tree.update_batch(&refs).map_err(|e| TuliproxError::RepositoryXtream(format!("Failed to update tmp tree during merge: {e}")))?; - updates.clear(); + pending_updates.push((new_item.provider_id, new_item)); + if pending_updates.len() >= BATCH_SIZE { + updated_count = updated_count.saturating_add(write_preserved_detail_batch( + &mut staging_tree, + staging_path, + &mut pending_updates, + &mut before_operation, + )?); } } } @@ -812,17 +1017,44 @@ fn preserve_details_input_xtream_playlist_cluster_to_disk( } } - if !updates.is_empty() { - let refs: Vec<(&u32, &XtreamPlaylistItem)> = - updates.iter().map(|(id, pli)| (id, pli)).collect(); - new_tree.update_batch(&refs) - .map_err(|e| TuliproxError::RepositoryXtream(format!("Failed to update tmp tree during merge: {e}")))?; - } + updated_count = updated_count.saturating_add(write_preserved_detail_batch( + &mut staging_tree, + staging_path, + &mut pending_updates, + &mut before_operation, + )?); + before_operation(DetailPreservationOperation::Commit) + .and_then(|()| staging_tree.commit()) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to commit staging Xtream tree {} after detail preservation: {error}", + staging_path.display() + )) + })?; - new_tree.commit() - .map_err(|e| TuliproxError::RepositoryXtream(format!("Failed to commit tmp tree merge: {e}")))?; + Ok(PreserveDetailsOutcome::Merged { + scanned: scanned_count, + updated: updated_count, + }) +} - Ok(()) +#[cfg(test)] +fn preserve_details_with_injected_operation_failure( + published_path: &Path, + staging_path: &Path, + failure: DetailPreservationOperation, +) -> Result { + preserve_details_input_xtream_playlist_cluster_to_disk_with_hook( + published_path, + staging_path, + move |operation| { + if operation == failure { + Err(io::Error::other(format!("injected {failure:?} failure"))) + } else { + Ok(()) + } + }, + ) } #[allow(clippy::too_many_lines)] @@ -835,7 +1067,8 @@ pub async fn persist_input_xtream_playlist_cluster_to_disk( ) -> Result<(), TuliproxError> { let cfg = app_config.config.load(); let storage_path = ensure_input_storage_path(&cfg, &input.name).await?; - let xtream_path = xtream_get_file_path(&storage_path, cluster); + drop(cfg); + let refresh_lease = XtreamRefreshLease::new(XtreamRefreshPaths::new(&storage_path, cluster)?)?; // Channel for transferring items from Parser (Async Task) to Consumer (Blocking Task) let (tx, mut rx) = tokio::sync::mpsc::channel::(BATCH_SIZE * 2); @@ -872,24 +1105,25 @@ pub async fn persist_input_xtream_playlist_cluster_to_disk( }); // 2. Consumer Task: Handles heavy Disk I/O (BPlusTree updates) - let xtream_path_for_consumer = xtream_path.clone(); + let consumer_lease = refresh_lease.clone(); let consumer_task = tokio::task::spawn_blocking(move || { - let tmp_xtream_path = xtream_path_for_consumer.with_extension("tmp"); + let staging_path = &consumer_lease.paths().staging_database; BPlusTree::::new() - .store(&tmp_xtream_path) - .map_err(|e| { - error!( - "Failed to initialize ghost BPlusTree at {}: {e}", - tmp_xtream_path.display() - ); - TuliproxError::RepositoryXtream(format!("Init tree error {e}")) + .store(staging_path) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to initialize staging Xtream tree {} for {cluster}: {error}", + staging_path.display() + )) })?; let mut tree: BPlusTreeUpdate = - BPlusTreeUpdate::try_new_with_backoff(&tmp_xtream_path).map_err(|e| { - error!("Failed to open ghost tree at {}: {e}", tmp_xtream_path.display()); - TuliproxError::RepositoryXtream(format!("Failed to open tree {e}")) + BPlusTreeUpdate::try_new_with_backoff(staging_path).map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to open staging Xtream tree {} for {cluster}: {error}", + staging_path.display() + )) })?; tree.set_flush_policy(FlushPolicy::Batch); @@ -902,21 +1136,28 @@ pub async fn persist_input_xtream_playlist_cluster_to_disk( let batch: Vec<(&u32, &XtreamPlaylistItem)> = buffer.iter().map(|i| (&i.provider_id, i)).collect(); let prepared = BPlusTreeUpdate::::prepare_upsert_batch(&batch) - .map_err(|e| { - error!("Batch prepare failed for cluster {cluster}: {e}"); - TuliproxError::RepositoryXtream(format!("Prepare failed {e}")) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to prepare staging batch for {cluster} at {}: {error}", + staging_path.display() + )) })?; - tree.upsert_batch_encoded(prepared).map_err(|e| { - error!("Batch upsert failed for cluster {cluster}: {e}"); - TuliproxError::RepositoryXtream(format!("Upsert failed {e}")) + tree.upsert_batch_encoded(prepared).map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to write staging batch for {cluster} at {}: {error}", + staging_path.display() + )) })?; // Commit per batch so the write transaction's dirty-page map stays bounded. // Holding it open for the whole cluster buffered ~48k pages (196 MB); the // import writes into a .tmp file that is renamed on success, so atomicity // comes from the rename, not from a single transaction. tree.commit().map_err(|e| { - error!("Batch commit failed for cluster {cluster}: {e}"); - TuliproxError::RepositoryXtream(format!("Commit failed {e}")) + error!("Batch commit failed for cluster {cluster} at {}: {e}", staging_path.display()); + TuliproxError::RepositoryXtream(format!( + "Failed to commit staging batch for {cluster} at {}: {e}", + staging_path.display() + )) })?; buffer.clear(); } @@ -927,19 +1168,25 @@ pub async fn persist_input_xtream_playlist_cluster_to_disk( let batch: Vec<(&u32, &XtreamPlaylistItem)> = buffer.iter().map(|i| (&i.provider_id, i)).collect(); let prepared = BPlusTreeUpdate::::prepare_upsert_batch(&batch) - .map_err(|e| { - error!("Final batch prepare failed for cluster {cluster}: {e}"); - TuliproxError::RepositoryXtream(format!("Prepare failed {e}")) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to prepare final staging batch for {cluster} at {}: {error}", + staging_path.display() + )) })?; - tree.upsert_batch_encoded(prepared).map_err(|e| { - error!("Final batch upsert failed for cluster {cluster}: {e}"); - TuliproxError::RepositoryXtream(format!("Upsert failed {e}")) + tree.upsert_batch_encoded(prepared).map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to write final staging batch for {cluster} at {}: {error}", + staging_path.display() + )) })?; } - tree.commit().map_err(|e| { - error!("Commit failed for cluster {cluster}: {e}"); - TuliproxError::RepositoryXtream(format!("Commit failed {e}")) + tree.commit().map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to commit staging Xtream tree for {cluster} at {}: {error}", + staging_path.display() + )) })?; Ok::<(), TuliproxError>(()) }); @@ -953,94 +1200,197 @@ pub async fn persist_input_xtream_playlist_cluster_to_disk( let parsed_categories = parse_res?; consumer_res?; - // --- Post-Processing & Atomic Swap --- + save_xtream_categories_to_file(refresh_lease.clone(), &parsed_categories).await?; - let col_path = match cluster { - XtreamCluster::Live => get_live_cat_collection_path(&storage_path), - XtreamCluster::Video => get_vod_cat_collection_path(&storage_path), - XtreamCluster::Series => get_series_cat_collection_path(&storage_path), - }; + // Lock order for the publish phase is always FileLockManager(final) followed by B+Tree sidecars. No B+Tree + // handle escapes its blocking closure, so none is held while this async lock is acquired. + let publish_lock = Arc::new( + app_config + .file_locks + .write_lock(&refresh_lease.paths().published_database) + .await, + ); - let tmp_col_path = col_path.with_extension("tmp"); + let merge_lease = refresh_lease.clone(); + let merge_lock = Arc::clone(&publish_lock); + let merge_outcome = tokio::task::spawn_blocking(move || { + let _publish_guard = merge_lock; + preserve_details_input_xtream_playlist_cluster_to_disk( + &merge_lease.paths().published_database, + &merge_lease.paths().staging_database, + ) + }) + .await + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Detail-preservation task failed to join during {cluster} refresh: {error}" + )) + })??; + log::debug!( + "Xtream cluster detail preservation completed: cluster={cluster} generation={} outcome={merge_outcome:?}", + refresh_lease.paths().generation + ); - // Use the parsed_categories returned from the task, not the 'categories' reader - save_xtream_categories_to_file(&tmp_col_path, &parsed_categories).await?; + let compact_lease = refresh_lease.clone(); + let compact_lock = Arc::clone(&publish_lock); + tokio::task::spawn_blocking(move || { + let _publish_guard = compact_lock; + let staging_path = &compact_lease.paths().staging_database; + let mut tree = BPlusTreeUpdate::::try_new_with_backoff(staging_path) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to open staging Xtream tree {} for compaction: {error}", + staging_path.display() + )) + })?; + tree.compact().map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to compact staging Xtream tree {}: {error}", + staging_path.display() + )) + }) + }) + .await + .map_err(|error| { + TuliproxError::RepositoryXtream(format!("Compaction task failed to join during {cluster} refresh: {error}")) + })??; - let tmp_xtream_path = xtream_path.with_extension("tmp"); + let database_publish_lease = refresh_lease.clone(); + let database_publish_lock = Arc::clone(&publish_lock); + tokio::task::spawn_blocking(move || { + let _publish_guard = database_publish_lock; + publish_staged_database::( + &database_publish_lease.paths().staging_database, + &database_publish_lease.paths().published_database, + ) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to publish staging Xtream database for {cluster}: {error}" + )) + }) + }) + .await + .map_err(|error| { + TuliproxError::RepositoryXtream(format!("Database publish task failed to join for {cluster}: {error}")) + })??; - // Acquire file lock to ensure atomic operations - let swap_lock = app_config.file_locks.write_lock(&xtream_path).await; + let category_publish_lease = refresh_lease.clone(); + let category_publish_lock = Arc::clone(&publish_lock); + tokio::task::spawn_blocking(move || { + let _publish_guard = category_publish_lock; + publish_staged_file_same_directory( + &category_publish_lease.paths().staging_categories, + &category_publish_lease.paths().published_categories, + ) + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Xtream database for {cluster} was published, but category publication failed: {error}" + )) + }) + }) + .await + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Xtream database for {cluster} was published, but the category publish task failed to join: {error}" + )) + })??; - { - let old_path = xtream_path.clone(); - let new_tmp_path = tmp_xtream_path.clone(); - tokio::task::spawn_blocking(move || preserve_details_input_xtream_playlist_cluster_to_disk(&old_path, &new_tmp_path)) - .await - .map_err(|e| TuliproxError::RepositoryXtream(format!("Merge task join error during cluster {cluster} update: {e}")))??; - } - - // Optional compaction to optimize the newly created database file. - // Run on blocking pool because B+Tree lock backoff uses std::thread::sleep. - let compact_path = tmp_xtream_path.clone(); - match tokio::task::spawn_blocking(move || -> Result<(), std::io::Error> { - if let Ok(mut tree_update) = BPlusTreeUpdate::::try_new_with_backoff(&compact_path) { - tree_update.compact()?; - } - Ok(()) + let cleanup_lease = refresh_lease.clone(); + let cleanup_lock = Arc::clone(&publish_lock); + tokio::task::spawn_blocking(move || { + let _publish_guard = cleanup_lock; + cleanup_lease.cleanup_staging_artifacts() }) .await - { - Ok(Ok(())) => {} - Ok(Err(e)) => { - error!( - "Failed to compact temporary database for {cluster} at {:?}: {e}", - tmp_xtream_path.display() - ); - } - Err(e) => { - error!( - "Compaction task join failed for {cluster} at {:?}: {e}", - tmp_xtream_path.display() - ); - } - } + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Xtream refresh for {cluster} was published, but cleanup task failed to join: {error}" + )) + })? + .map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Xtream refresh for {cluster} was published, but staging cleanup failed: {error}" + )) + })?; - // Atomic Swap: Replace old database with new one - if let Err(e) = crate::utils::rename_or_copy(&tmp_xtream_path, &xtream_path, false) { - error!( - "Failed to swap xtream database for {cluster} (from {:?} to {:?}): {e}", - tmp_xtream_path.display(), - xtream_path.display() - ); - return Err(TuliproxError::RepositoryXtream(format!("Failed to swap database: {e}"))); - } - - // Atomic Swap: Replace old categories with new ones - if let Err(e) = crate::utils::rename_or_copy(&tmp_col_path, &col_path, false) { - error!( - "Failed to swap xtream categories for {cluster} (from {:?} to {:?}): {e}", - tmp_col_path.display(), - col_path.display() - ); - return Err(TuliproxError::RepositoryXtream(format!("Failed to swap categories: {e}"))); - } - - // Cleanup: Remove temporary files if they still exist (defensive) - let _ = tokio::fs::remove_file(&tmp_xtream_path).await; - let _ = tokio::fs::remove_file(&tmp_col_path).await; - - // Explicitly release the lock (though RAII would handle it too) - drop(swap_lock); - - log::debug!("Cluster {cluster} updated successfully."); + drop(publish_lock); + log::debug!( + "Xtream cluster updated successfully: cluster={cluster} generation={}", + refresh_lease.paths().generation + ); Ok(()) } +fn publish_staged_file_same_directory(staging: &Path, published: &Path) -> io::Result<()> { + use crate::repository::bplustree::common::{parent_or_dot, require_same_parent_directory}; + require_same_parent_directory(staging, published)?; + let staging_path = tempfile::TempPath::try_from_path(staging).map_err(|error| { + io::Error::new( + error.kind(), + format!("failed to prepare staging file {} for publication: {error}", staging.display()), + ) + })?; + staging_path.persist(published).map_err(io::Error::from)?; + sync_published_file_parent(published).map_err(|error| { + io::Error::new( + error.kind(), + format!( + "file {} was published, but its parent directory {} could not be synchronized: {error}", + published.display(), + parent_or_dot(published).display() + ), + ) + }) +} + +#[cfg(unix)] +fn sync_published_file_parent(path: &Path) -> io::Result<()> { + File::open(crate::repository::bplustree::common::parent_or_dot(path))?.sync_all() +} + +/// Windows (NTFS) and other non-Unix targets open directories as files for +/// `sync_all`, just as Unix does. The durability barrier is required after +/// every rename so a power loss cannot leave the directory entry pointing at +/// the pre-rename file. +#[cfg(not(unix))] +fn sync_published_file_parent(path: &Path) -> io::Result<()> { + File::open(crate::repository::bplustree::common::parent_or_dot(path))?.sync_all() +} + +/// Owns the staging category file plus its `flock`, so the lock is released +/// even when a later step (`serde_json::to_writer`, `sync_all`) returns an +/// error. The unlock runs in `Drop` and is logged on failure; closing the +/// underlying `File` releases the OS-level lock either way. +struct LockedCategoryFile { + file: File, + path: PathBuf, +} + +impl LockedCategoryFile { + fn create(path: &Path) -> io::Result { + let file = File::create(path)?; + file.lock_exclusive()?; + Ok(Self { file, path: path.to_path_buf() }) + } + + fn sync_all(&self) -> io::Result<()> { self.file.sync_all() } +} + +impl Drop for LockedCategoryFile { + fn drop(&mut self) { + if let Err(error) = fs2::FileExt::unlock(&self.file) { + log::warn!( + "Failed to unlock staging category file {}: {error}; the OS will release it on close", + self.path.display() + ); + } + } +} + async fn save_xtream_categories_to_file( - col_path: &Path, + refresh_lease: XtreamRefreshLease, categories: &[XtreamCategory], ) -> Result<(), TuliproxError> { - let col_path_buf = col_path.to_path_buf(); let cat_entries: Vec = categories .iter() .map(|c| CategoryEntry { @@ -1051,21 +1401,25 @@ async fn save_xtream_categories_to_file( .collect(); tokio::task::spawn_blocking(move || { - if let Ok(file) = File::create(&col_path_buf) { - if let Err(e) = file.lock_exclusive() { - log::warn!( - "Could not acquire exclusive lock for {}: {e}, proceeding without lock", - col_path_buf.display() - ); - } - serde_json::to_writer(&file, &cat_entries).map_err(|e| { - error!("Failed to write categories to file {}: {e}", col_path_buf.display()); - TuliproxError::RepositoryXtream(format!("Write failed: {e}")) - })?; - let _ = file.unlock(); - } else { - return Err(TuliproxError::RepositoryXtream(format!("Failed to create category file {}", col_path_buf.display()))); - } + let staging_path = &refresh_lease.paths().staging_categories; + let locked = LockedCategoryFile::create(staging_path).map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to create or lock staging category file {}: {error}", + staging_path.display() + )) + })?; + serde_json::to_writer(&locked.file, &cat_entries).map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to write staging category file {}: {error}", + staging_path.display() + )) + })?; + locked.sync_all().map_err(|error| { + TuliproxError::RepositoryXtream(format!( + "Failed to synchronize staging category file {}: {error}", + staging_path.display() + )) + })?; Ok(()) }) .await @@ -1484,15 +1838,98 @@ pub async fn load_input_xtream_playlist(app_config: &Arc, storage_pat #[cfg(test)] mod tests { - use super::{merge_preserved_stream_properties, needs_update_info_details, preserve_details_input_xtream_playlist_cluster_to_disk}; - use crate::repository::{BPlusTreeQuery, BPlusTreeUpdate}; + use super::{ + merge_preserved_stream_properties, needs_update_info_details, + persist_input_xtream_playlist_cluster_to_disk, preserve_details_input_xtream_playlist_cluster_to_disk, + preserve_details_with_injected_operation_failure, publish_staged_file_same_directory, + refresh_staging_path, DetailPreservationOperation, PreserveDetailsOutcome, XtreamRefreshLease, + XtreamRefreshPaths, + }; + use crate::model::{ + ApiProxyConfig, AppConfig, Config, ConfigInput, CustomStreamResponse, HdHomeRunConfig, + MediaToolCapabilities, SourcesConfig, + }; + use crate::repository::{ + bplustree::common::{ensure_distinct_sidecar_lock_domains, sidecar_lock_path}, + build_input_storage_path, get_file_path_for_db_index, BPlusTreeQuery, BPlusTreeUpdate, + }; + use crate::utils::{request::DynReader, FileLockManager}; + use arc_swap::{ArcSwap, ArcSwapOption}; use shared::model::{ - CatchupProperties, LiveStreamProperties, SeriesStreamProperties, StreamProperties, VideoStreamProperties, - XtreamCluster, XtreamPlaylistItem, + CatchupProperties, ConfigPaths, InputType, LiveStreamProperties, SeriesStreamProperties, StreamProperties, + VideoStreamProperties, XtreamCluster, XtreamPlaylistItem, }; use shared::utils::Internable; - use std::path::Path; + use std::{ + env, fs, io, + path::Path, + process::{Child, Command, ExitStatus, Stdio}, + sync::Arc, + thread, + time::{Duration, Instant}, + }; use tempfile::tempdir; + use tokio::io::AsyncWriteExt; + use uuid::Uuid; + + fn test_app_config(storage_dir: &Path) -> Arc { + Arc::new(AppConfig { + config: Arc::new(ArcSwap::from_pointee(Config { + storage_dir: storage_dir.to_string_lossy().into_owned(), + ..Config::default() + })), + sources: Arc::new(ArcSwap::from_pointee(SourcesConfig::default())), + hdhomerun: Arc::new(ArcSwapOption::::default()), + api_proxy: Arc::new(ArcSwapOption::::default()), + file_locks: Arc::new(FileLockManager::default()), + paths: Arc::new(ArcSwap::from_pointee(ConfigPaths { + home_path: String::new(), + config_path: String::new(), + storage_path: String::new(), + config_file_path: String::new(), + sources_file_path: String::new(), + mapping_file_path: None, + mapping_files_used: None, + template_file_path: None, + template_files_used: None, + api_proxy_file_path: String::new(), + custom_stream_response_path: None, + })), + custom_stream_response: Arc::new(ArcSwapOption::::default()), + access_token_secret: [0; 32], + encrypt_secret: [0; 16], + media_tools: Arc::new(MediaToolCapabilities::new()), + }) + } + + fn json_reader(content: &'static str) -> DynReader { + let (mut writer, reader) = tokio::io::duplex(4096); + tokio::spawn(async move { + writer.write_all(content.as_bytes()).await.expect("fixture should fit into duplex reader"); + writer.shutdown().await.expect("fixture writer should shut down"); + }); + Box::pin(reader) + } + + fn wait_for_child(mut child: Child, timeout: Duration) -> io::Result { + let deadline = Instant::now() + timeout; + loop { + match child.try_wait() { + Ok(Some(status)) => return Ok(status), + Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(10)), + Ok(None) => { + let _ = child.kill(); + let _ = child.wait(); + return Err(io::Error::new(io::ErrorKind::TimedOut, "Xtream refresh child timed out")); + } + Err(error) => { + let _ = child.kill(); + let _ = child.wait(); + return Err(error); + } + } + } + } #[test] fn keeps_existing_details_when_new_timestamp_is_missing() { @@ -1745,15 +2182,14 @@ mod tests { } } - #[test] - fn preserve_details_for_disk_cluster_copies_missing_live_probe_fields() { - let dir = tempdir().expect("temp dir should be created"); - let old_path = dir.path().join("old_live.db"); - let tmp_path = dir.path().join("tmp_live.db"); - let provider_id = 100_u32; + fn fixed_refresh_paths(path: &Path, generation: u128) -> XtreamRefreshPaths { + XtreamRefreshPaths::for_generation(path, XtreamCluster::Live, Uuid::from_u128(generation)) + .expect("fixed refresh paths should be valid") + } + fn write_detail_preservation_fixture(paths: &XtreamRefreshPaths, provider_id: u32) { write_single_item( - &old_path, + &paths.published_database, &make_live_item( provider_id, Some("{\"codec_name\":\"h264\"}"), @@ -1763,11 +2199,71 @@ mod tests { 2_500_000, ), ); - write_single_item(&tmp_path, &make_live_item(provider_id, None, None, None, None, 0)); + write_single_item( + &paths.staging_database, + &make_live_item(provider_id, None, None, None, None, 0), + ); + } - preserve_details_input_xtream_playlist_cluster_to_disk(&old_path, &tmp_path).expect("merge should succeed"); + #[test] + fn refresh_staging_database_uses_distinct_published_lock_domain() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 1); - let merged = read_live_props(&tmp_path, provider_id); + assert_eq!(paths.published_database.file_name().and_then(|name| name.to_str()), Some("live.db")); + assert_ne!(sidecar_lock_path(&paths.published_database), sidecar_lock_path(&paths.staging_database)); + } + + #[test] + fn refresh_staging_database_and_index_share_one_generation_lock_domain() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 2); + let staging_index = get_file_path_for_db_index(&paths.staging_database); + + assert_eq!(sidecar_lock_path(&paths.staging_database), sidecar_lock_path(&staging_index)); + } + + #[test] + fn colliding_staging_path_is_rejected_before_lock_acquisition() { + let dir = tempdir().expect("temp dir should be created"); + let published = dir.path().join("live.db"); + let colliding = dir.path().join("live.tmp"); + + let error = ensure_distinct_sidecar_lock_domains(&published, &colliding) + .expect_err("colliding sidecar domains must be rejected"); + assert_eq!(error.kind(), io::ErrorKind::InvalidInput); + } + + #[test] + fn preserve_details_for_disk_cluster_copies_missing_live_probe_fields() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 3); + let provider_id = 100_u32; + + write_single_item( + &paths.published_database, + &make_live_item( + provider_id, + Some("{\"codec_name\":\"h264\"}"), + Some("{\"codec_name\":\"aac\"}"), + Some(1_700_000_000), + Some(1_700_000_100), + 2_500_000, + ), + ); + write_single_item( + &paths.staging_database, + &make_live_item(provider_id, None, None, None, None, 0), + ); + + let outcome = preserve_details_input_xtream_playlist_cluster_to_disk( + &paths.published_database, + &paths.staging_database, + ) + .expect("merge should succeed"); + assert_eq!(outcome, PreserveDetailsOutcome::Merged { scanned: 1, updated: 1 }); + + let merged = read_live_props(&paths.staging_database, provider_id); assert_eq!(merged.video, Some("{\"codec_name\":\"h264\"}".intern())); assert_eq!(merged.audio, Some("{\"codec_name\":\"aac\"}".intern())); assert_eq!(merged.last_probed_timestamp, Some(1_700_000_000)); @@ -1775,6 +2271,309 @@ mod tests { assert_eq!(merged.bitrate, 2_500_000); } + #[test] + fn preserve_details_reports_missing_published_database_explicitly() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 4); + write_single_item(&paths.staging_database, &make_live_item(101, None, None, None, None, 0)); + + let outcome = preserve_details_input_xtream_playlist_cluster_to_disk( + &paths.published_database, + &paths.staging_database, + ) + .expect("a missing published database should not fail the refresh"); + + assert_eq!(outcome, PreserveDetailsOutcome::SourceMissing); + } + + #[test] + fn preserve_details_propagates_corrupt_published_database() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 5); + fs::write(&paths.published_database, b"corrupt").expect("corrupt fixture should be written"); + write_single_item(&paths.staging_database, &make_live_item(102, None, None, None, None, 0)); + + let error = preserve_details_input_xtream_playlist_cluster_to_disk( + &paths.published_database, + &paths.staging_database, + ) + .expect_err("corrupt published data must fail"); + + assert!(error.to_string().contains(&paths.published_database.display().to_string())); + } + + #[test] + fn preserve_details_propagates_corrupt_staging_database() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 6); + write_single_item(&paths.published_database, &make_live_item(103, None, None, None, None, 0)); + fs::write(&paths.staging_database, b"corrupt").expect("corrupt fixture should be written"); + + let error = preserve_details_input_xtream_playlist_cluster_to_disk( + &paths.published_database, + &paths.staging_database, + ) + .expect_err("corrupt staging data must fail"); + + assert!(error.to_string().contains(&paths.staging_database.display().to_string())); + } + + #[test] + fn preserve_details_propagates_missing_staging_database() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 14); + write_single_item(&paths.published_database, &make_live_item(105, None, None, None, None, 0)); + + let error = preserve_details_input_xtream_playlist_cluster_to_disk( + &paths.published_database, + &paths.staging_database, + ) + .expect_err("a missing staging database must fail"); + let message = error.to_string(); + + assert!(message.contains("Failed to open staging Xtream tree")); + assert!(message.contains(&paths.staging_database.display().to_string())); + } + + #[test] + fn preserve_details_propagates_staging_query_failure() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 15); + write_detail_preservation_fixture(&paths, 106); + + let error = preserve_details_with_injected_operation_failure( + &paths.published_database, + &paths.staging_database, + DetailPreservationOperation::Query, + ) + .expect_err("a staging query failure must fail the merge"); + let message = error.to_string(); + + assert!(message.contains("Failed to query staging Xtream tree")); + assert!(message.contains(&paths.staging_database.display().to_string())); + assert!(message.contains("injected Query failure")); + } + + #[test] + fn preserve_details_propagates_staging_batch_write_failure() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 16); + write_detail_preservation_fixture(&paths, 107); + + let error = preserve_details_with_injected_operation_failure( + &paths.published_database, + &paths.staging_database, + DetailPreservationOperation::BatchWrite, + ) + .expect_err("a staging batch write failure must fail the merge"); + let message = error.to_string(); + + assert!(message.contains("Failed to update staging Xtream tree")); + assert!(message.contains(&paths.staging_database.display().to_string())); + assert!(message.contains("injected BatchWrite failure")); + } + + #[test] + fn preserve_details_propagates_staging_commit_failure() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 17); + write_detail_preservation_fixture(&paths, 108); + + let error = preserve_details_with_injected_operation_failure( + &paths.published_database, + &paths.staging_database, + DetailPreservationOperation::Commit, + ) + .expect_err("a staging commit failure must fail the merge"); + let message = error.to_string(); + + assert!(message.contains("Failed to commit staging Xtream tree")); + assert!(message.contains(&paths.staging_database.display().to_string())); + assert!(message.contains("injected Commit failure")); + } + + #[test] + fn preserve_details_empty_merge_reports_zero_updates() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 7); + let item = make_live_item(104, None, None, None, None, 0); + write_single_item(&paths.published_database, &item); + write_single_item(&paths.staging_database, &item); + + let outcome = preserve_details_input_xtream_playlist_cluster_to_disk( + &paths.published_database, + &paths.staging_database, + ) + .expect("empty merge should succeed"); + + assert_eq!(outcome, PreserveDetailsOutcome::Merged { scanned: 1, updated: 0 }); + } + + #[test] + fn refresh_lease_cleanup_is_idempotent_and_generation_local() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 8); + let other_paths = fixed_refresh_paths(dir.path(), 9); + let lease = XtreamRefreshLease::new(paths.clone()).expect("refresh lease should be valid"); + + fs::write(&paths.published_database, b"published").expect("published fixture should be written"); + let published_lock = sidecar_lock_path(&paths.published_database); + fs::write(&published_lock, b"").expect("published lock fixture should be written"); + for artifact in lease.0.database_artifacts.owned_paths() { + fs::write(artifact, b"staging").expect("staging artifact should be written"); + } + fs::write(&paths.staging_categories, b"staging").expect("staging category should be written"); + fs::write(&other_paths.staging_database, b"other generation") + .expect("other generation fixture should be written"); + + lease.cleanup_staging_artifacts().expect("first cleanup should succeed"); + lease.cleanup_staging_artifacts().expect("second cleanup should be idempotent"); + + assert!(paths.published_database.exists()); + assert!(published_lock.exists()); + assert!(other_paths.staging_database.exists()); + assert!(!paths.staging_database.exists()); + assert!(!paths.staging_categories.exists()); + } + + #[test] + fn refresh_lease_defers_cleanup_until_last_worker_clone_drops() { + let dir = tempdir().expect("temp dir should be created"); + let paths = fixed_refresh_paths(dir.path(), 13); + let parent_lease = XtreamRefreshLease::new(paths.clone()).expect("refresh lease should be valid"); + let worker_lease = parent_lease.clone(); + fs::write(&paths.staging_database, b"staging").expect("staging fixture should be written"); + fs::write(&paths.staging_categories, b"categories").expect("category fixture should be written"); + + drop(parent_lease); + assert!(paths.staging_database.exists()); + assert!(paths.staging_categories.exists()); + + drop(worker_lease); + assert!(!paths.staging_database.exists()); + assert!(!paths.staging_categories.exists()); + } + + #[test] + fn sequential_refresh_generations_use_distinct_stems() { + let dir = tempdir().expect("temp dir should be created"); + let first = fixed_refresh_paths(dir.path(), 10); + let second = fixed_refresh_paths(dir.path(), 11); + + assert_ne!(first.staging_database.file_stem(), second.staging_database.file_stem()); + assert_ne!(sidecar_lock_path(&first.staging_database), sidecar_lock_path(&second.staging_database)); + } + + #[test] + fn category_publish_atomically_replaces_same_directory_file() { + let dir = tempdir().expect("temp dir should be created"); + let published = dir.path().join("cat_live.json"); + let staging = dir.path().join("cat_live.refresh-fixed.json"); + fs::write(&published, b"old").expect("published fixture should be written"); + fs::write(&staging, b"new").expect("staging fixture should be written"); + + publish_staged_file_same_directory(&staging, &published).expect("category publish should succeed"); + + assert_eq!(fs::read(&published).expect("published categories should be readable"), b"new"); + assert!(!staging.exists()); + } + + #[cfg(unix)] + #[test] + fn refresh_staging_path_preserves_non_utf8_stem() { + use std::os::unix::ffi::{OsStrExt, OsStringExt}; + + let mut input_name = std::ffi::OsString::from_vec(vec![b'l', b'i', b'v', b'e', 0xff]); + input_name.push(".db"); + let published = Path::new("/tmp").join(input_name); + let generation = Uuid::from_u128(12); + + let staging = refresh_staging_path(&published, generation).expect("non-UTF-8 path should be supported"); + let bytes = staging.file_name().expect("staging file name").as_bytes(); + assert!(bytes.starts_with(&[b'l', b'i', b'v', b'e', 0xff])); + assert!(bytes.ends_with(b".db")); + assert!(bytes.windows(b".refresh-".len()).any(|window| window == b".refresh-")); + } + + #[test] + fn xtream_refresh_end_to_end_child() -> io::Result<()> { + let Some(storage_root) = env::var_os("TULIPROX_XTREAM_REFRESH_TEST_ROOT") else { + return Ok(()); + }; + let storage_root = Path::new(&storage_root); + let app_config = test_app_config(storage_root); + let input = ConfigInput { + name: "deadlock-test".intern(), + input_type: InputType::Xtream, + ..ConfigInput::default() + }; + let runtime = tokio::runtime::Builder::new_multi_thread().enable_all().build()?; + runtime.block_on(async { + for _ in 0..2 { + persist_input_xtream_playlist_cluster_to_disk( + &app_config, + &input, + XtreamCluster::Live, + json_reader(r#"[{"category_id":"1","category_name":"Sports"}]"#), + json_reader(r#"[{"name":"Live","stream_id":700,"category_id":"1","added":"0"}]"#), + ) + .await + .map_err(|error| io::Error::other(error.to_string()))?; + } + Ok::<(), io::Error>(()) + })?; + + let input_storage = build_input_storage_path(&input.name, storage_root.to_string_lossy().as_ref()); + let published = super::xtream_get_file_path(&input_storage, XtreamCluster::Live); + let learned = read_live_props(&published, 700); + assert_eq!(learned.video, Some("{\"codec_name\":\"h264\"}".intern())); + assert_eq!(learned.bitrate, 2_500_000); + Ok(()) + } + + #[test] + fn two_sequential_cluster_refreshes_complete_without_generation_artifacts() -> io::Result<()> { + let directory = tempfile::tempdir()?; + let input_name = "deadlock-test".intern(); + let input_storage = build_input_storage_path(&input_name, directory.path().to_string_lossy().as_ref()); + fs::create_dir_all(&input_storage)?; + let published = super::xtream_get_file_path(&input_storage, XtreamCluster::Live); + write_single_item( + &published, + &make_live_item( + 700, + Some("{\"codec_name\":\"h264\"}"), + Some("{\"codec_name\":\"aac\"}"), + Some(1_700_000_000), + Some(1_700_000_100), + 2_500_000, + ), + ); + + let child = Command::new(env::current_exe()?) + .arg("--exact") + .arg("repository::xtream_repository::tests::xtream_refresh_end_to_end_child") + .arg("--nocapture") + .env("TULIPROX_XTREAM_REFRESH_TEST_ROOT", directory.path()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn()?; + let status = wait_for_child(child, Duration::from_secs(20))?; + assert!(status.success(), "Xtream refresh child failed with {status}"); + + let learned = read_live_props(&published, 700); + assert_eq!(learned.video, Some("{\"codec_name\":\"h264\"}".intern())); + assert_eq!(learned.bitrate, 2_500_000); + let entries = fs::read_dir(&input_storage)?.collect::>>()?; + let generation_artifacts = entries + .into_iter() + .filter(|entry| entry.file_name().to_string_lossy().contains(".refresh-")) + .collect::>(); + assert!(generation_artifacts.is_empty(), "generation artifacts survived successful refreshes"); + assert!(sidecar_lock_path(&published).exists()); + Ok(()) + } + #[test] fn preserve_details_for_disk_cluster_does_not_override_existing_live_probe_fields() { let dir = tempdir().expect("temp dir should be created");