diff --git a/crates/emulebb-daemon/src/bin/emulebb-rust.rs b/crates/emulebb-daemon/src/bin/emulebb-rust.rs index ef00010c..d2ac0fcb 100644 --- a/crates/emulebb-daemon/src/bin/emulebb-rust.rs +++ b/crates/emulebb-daemon/src/bin/emulebb-rust.rs @@ -88,15 +88,15 @@ async fn main() -> Result<()> { Some(CliCommand::ReconcileSharedFiles | CliCommand::ImportSharedDirectories { .. }) ) { if let Some(CliCommand::ImportSharedDirectories { ref file }) = cli.command { - let store = emulebb_metadata::MetadataStore::open(profile.metadata_path())?; + let store = profile.open_metadata_store()?; emulebb_daemon::shared_directory_import::import_shared_directories(&store, file)?; } let core = emulebb_core::EmulebbCore::new( BUILD_VERSION, - emulebb_index::FileIndex::open(profile.metadata_path())?, + emulebb_index::FileIndex::from_metadata_store(profile.open_metadata_store()?), profile.transfer_root(), )? - .with_profile_dir(profile.profile_dir.clone()) + .with_profile_dir(profile.profile_io_dir().to_path_buf()) .with_incoming_dir(profile.incoming_dir()); if matches!( cli.command, @@ -119,11 +119,10 @@ async fn main() -> Result<()> { } let work = core.shared_directory_reload_progress(); let reconciliation = core.reconcile_shared_files().await?; - let database = profile.metadata_path(); - let missing_recovery = tokio::task::spawn_blocking(move || { - emulebb_metadata::MetadataStore::open(database)?.missing_live_aich_recovery_count() - }) - .await??; + let metadata = profile.open_metadata_store()?; + let missing_recovery = + tokio::task::spawn_blocking(move || metadata.missing_live_aich_recovery_count()) + .await??; println!( "{}", serde_json::to_string_pretty( @@ -136,11 +135,14 @@ async fn main() -> Result<()> { println!("{}", serde_json::to_string_pretty(&summary)?); return Ok(()); } - let logging_guard = logging::init(&profile.profile_dir)?; + // Keep profile ownership until the final diagnostic writer has drained too. + let profile_lease = profile.profile_lease.clone(); + let logging_guard = logging::init(profile.profile_io_dir())?; let result = run(profile).await; if let Err(error) = result.as_ref() { tracing::error!(error = ?error, "daemon terminated with an error"); } logging_guard.shutdown().await; + drop(profile_lease); result } diff --git a/crates/emulebb-daemon/src/lib.rs b/crates/emulebb-daemon/src/lib.rs index df02f6d6..088e0cfd 100644 --- a/crates/emulebb-daemon/src/lib.rs +++ b/crates/emulebb-daemon/src/lib.rs @@ -2,7 +2,7 @@ use std::{ env, fs, io::Write, net::{IpAddr, Ipv4Addr, SocketAddr}, - path::PathBuf, + path::{Path, PathBuf}, sync::{Arc, atomic::AtomicBool}, time::Duration, }; @@ -34,12 +34,14 @@ mod bind_config; mod discovery_bootstrap; pub mod logging; mod profile_import; +mod profile_ownership; mod rest_listener; pub mod shared_directory_import; mod shutdown; mod vpn_guard_monitor; pub use profile_import::{ProfileImportApplySummary, apply_profile_import}; +pub use profile_ownership::ProfileLease; pub use shutdown::{ShutdownRequest, ShutdownSignal}; pub const PROFILE_SETTINGS_FILE: &str = "emulebb-rust-settings.toml"; @@ -89,6 +91,10 @@ pub struct DaemonProfile { pub rest: RestBootstrapSettings, pub vpn_guard: VpnGuardSettings, pub ip_filter: IpFilterSettings, + /// Runtime ownership shared with metadata handles; never a configuration key. + #[serde(skip)] + #[doc(hidden)] + pub profile_lease: Option, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -122,6 +128,7 @@ impl Default for DaemonProfile { rest: RestBootstrapSettings::default(), vpn_guard: VpnGuardSettings::default(), ip_filter: IpFilterSettings::default(), + profile_lease: None, } } } @@ -141,7 +148,24 @@ impl DaemonProfile { } fn load_from_dir(profile_dir: PathBuf, create_if_missing: bool) -> Result { - let settings_path = profile_dir.join(PROFILE_SETTINGS_FILE); + if !create_if_missing && !profile_dir.exists() { + bail!( + "profile settings file does not exist: {}", + profile_dir.join(PROFILE_SETTINGS_FILE).display() + ); + } + // WHY: bootstrap creation, schema initialization and reads of mutable + // state must all belong to the same exclusive writable-profile session. + let profile_lease = ProfileLease::acquire(&profile_dir, create_if_missing)?; + Self::load_owned(profile_dir, profile_lease, create_if_missing) + } + + fn load_owned( + profile_dir: PathBuf, + profile_lease: ProfileLease, + create_if_missing: bool, + ) -> Result { + let settings_path = profile_lease.directory().join(PROFILE_SETTINGS_FILE); let created_default_profile = if create_if_missing && !settings_path.exists() { create_default_profile_bootstrap(&profile_dir, &settings_path)?; true @@ -159,13 +183,16 @@ impl DaemonProfile { .with_context(|| format!("failed to read settings {}", settings_path.display()))?; let bootstrap: DaemonBootstrapSettings = toml::from_str(&text) .with_context(|| format!("failed to parse settings {}", settings_path.display()))?; - let metadata = - MetadataStore::open(profile_dir.join(PROFILE_METADATA_FILE)).with_context(|| { - format!( - "failed to open metadata store under {}", - profile_dir.display() - ) - })?; + let metadata = MetadataStore::open_with_profile_lease( + profile_lease.directory().join(PROFILE_METADATA_FILE), + profile_lease.file(), + ) + .with_context(|| { + format!( + "failed to open metadata store under {}", + profile_dir.display() + ) + })?; if created_default_profile { seed_default_profile_network_settings(&metadata)?; } @@ -184,6 +211,7 @@ impl DaemonProfile { rest: bootstrap.rest, vpn_guard: settings.vpn_guard, ip_filter: settings.ip_filter, + profile_lease: Some(profile_lease), }; profile .nat @@ -193,11 +221,33 @@ impl DaemonProfile { } pub fn metadata_path(&self) -> PathBuf { - self.profile_dir.join(PROFILE_METADATA_FILE) + self.profile_io_dir().join(PROFILE_METADATA_FILE) + } + + /// Use the acquired directory for profile I/O after loading or starting. + pub fn profile_io_dir(&self) -> &Path { + self.profile_lease + .as_ref() + .map_or(self.profile_dir.as_path(), ProfileLease::directory) + } + + /// Open an independent metadata connection retaining this profile's lease. + /// + /// Load the profile first so all connections share the same OS lock rather + /// than independently contending against their own process. + pub fn open_metadata_store(&self) -> Result { + let lease = self + .profile_lease + .as_ref() + .context("load the profile before opening writable metadata")?; + MetadataStore::open_with_profile_lease( + lease.directory().join(PROFILE_METADATA_FILE), + lease.file(), + ) } pub fn transfer_root(&self) -> PathBuf { - self.profile_dir.join("transfers") + self.profile_io_dir().join("transfers") } /// Resolve the finished-file delivery directory: the configured @@ -205,7 +255,7 @@ impl DaemonProfile { pub fn incoming_dir(&self) -> PathBuf { self.incoming_dir .clone() - .unwrap_or_else(|| self.profile_dir.join("incoming")) + .unwrap_or_else(|| self.profile_io_dir().join("incoming")) } pub fn ed2k_network_config( @@ -250,8 +300,8 @@ impl DaemonProfile { kad_local_store: kad_local_store_config(&self.kad), kad_snoop_queue: SnoopQueueConfig::default(), kad_passive_replay_enabled: self.kad.passive_replay_enabled, - kad_nodes_dat: discovery_bootstrap::load_valid_nodes_dat(&self.profile_dir)?, - kad_nodes_dat_path: Some(self.profile_dir.join("nodes.dat")), + kad_nodes_dat: discovery_bootstrap::load_valid_nodes_dat(self.profile_io_dir())?, + kad_nodes_dat_path: Some(self.profile_io_dir().join("nodes.dat")), kad_bootstrap_endpoints: self.kad_bootstrap_endpoints.clone(), kad_bootstrap_min_routing_contacts: self.kad.bootstrap_min_routing_contacts.max(1), kad_publish_shared_files: self.kad.publish_shared_files_enabled, @@ -361,7 +411,7 @@ impl DaemonProfile { let resolved = if configured.is_absolute() { configured.clone() } else { - self.profile_dir.join(configured) + self.profile_io_dir().join(configured) }; validate_web_root_dir(&resolved, "rest.webRootDir")?; return Ok(Some(resolved)); @@ -422,7 +472,6 @@ fn create_default_profile_bootstrap( profile_dir: &std::path::Path, settings_path: &std::path::Path, ) -> Result<()> { - create_private_profile_dir(profile_dir)?; let api_key = uuid::Uuid::new_v4().simple().to_string(); let text = format!("[rest]\nbindAddr = \"127.0.0.1:4711\"\napiKey = \"{api_key}\"\n"); let mut options = fs::OpenOptions::new(); @@ -700,7 +749,7 @@ struct HostedLifecycle { ready: oneshot::Sender, } -async fn run_inner(profile: DaemonProfile, host: Option) -> Result<()> { +async fn run_inner(mut profile: DaemonProfile, host: Option) -> Result<()> { if let Some(host) = &host { if host.shutdown.is_requested() { return Ok(()); @@ -724,11 +773,14 @@ async fn run_inner(profile: DaemonProfile, host: Option) -> Res "REST is listening beyond loopback over plaintext HTTP; restrict network exposure or terminate TLS at a trusted reverse proxy" ); } - fs::create_dir_all(&profile.profile_dir) - .with_context(|| format!("failed to create {}", profile.profile_dir.display()))?; - let index = FileIndex::open(profile.metadata_path())?; + // Embedded callers can construct a profile without loading it from disk. + // They acquire the same lease before any discovery, DB or runtime mutation. + if profile.profile_lease.is_none() { + profile.profile_lease = Some(ProfileLease::acquire(&profile.profile_dir, true)?); + } + let index = FileIndex::from_metadata_store(profile.open_metadata_store()?); let metadata_store = index.metadata_store(); - discovery_bootstrap::seed_empty_discovery(&metadata_store, &profile.profile_dir).await?; + discovery_bootstrap::seed_empty_discovery(&metadata_store, profile.profile_io_dir()).await?; let ed2k_network = profile.ed2k_network_config(&metadata_store)?; let ed2k_network_configured = ed2k_network.is_some(); let vpn_guard_monitor = ed2k_network @@ -744,7 +796,7 @@ async fn run_inner(profile: DaemonProfile, host: Option) -> Res profile.transfer_root(), ed2k_network, )? - .with_profile_dir(profile.profile_dir.clone()) + .with_profile_dir(profile.profile_io_dir().to_path_buf()) .with_incoming_dir(incoming_dir), ); let deletion_recovery = core.recover_pending_transfer_deletions().await?; diff --git a/crates/emulebb-daemon/src/profile_import.rs b/crates/emulebb-daemon/src/profile_import.rs index 3045a4ac..445a453d 100644 --- a/crates/emulebb-daemon/src/profile_import.rs +++ b/crates/emulebb-daemon/src/profile_import.rs @@ -112,7 +112,7 @@ pub fn apply_profile_import( header["targetPlatform"] = Value::from(NativePath::platform()); header["sharedRoots"] = serde_json::json!([]); } - let metadata = MetadataStore::open(profile.metadata_path())?; + let metadata = profile.open_metadata_store()?; let mut total = ImportCounts { imported_known_files: 0, skipped_native_known_files: 0, diff --git a/crates/emulebb-daemon/src/profile_ownership.rs b/crates/emulebb-daemon/src/profile_ownership.rs new file mode 100644 index 00000000..047ad910 --- /dev/null +++ b/crates/emulebb-daemon/src/profile_ownership.rs @@ -0,0 +1,123 @@ +//! OS-backed ownership of a writable profile, retained by all metadata handles. + +use std::{ + fs::{self, File, OpenOptions, TryLockError}, + path::{Path, PathBuf}, + sync::Arc, +}; + +use anyhow::{Context, Result, bail, ensure}; + +const PROFILE_LEASE_FILE: &str = ".emulebb-profile.lock"; + +/// An opaque, cloneable lease for one physical profile directory. +/// +/// The persistent empty file is a lock target, not an ownership marker. Never +/// remove it on release: the operating system releases ownership when the last +/// handle closes, including after process death. Locks must be supported by the +/// selected filesystem; an unsupported/failed lock is an error, never a bypass. +#[derive(Debug, Clone)] +pub struct ProfileLease { + directory: PathBuf, + file: Arc, +} + +impl ProfileLease { + pub(crate) fn acquire(profile: &Path, create_if_missing: bool) -> Result { + if create_if_missing { + super::create_private_profile_dir(profile)?; + } + let canonical_profile = fs::canonicalize(profile) + .with_context(|| format!("failed to resolve writable profile {}", profile.display()))?; + ensure!( + canonical_profile.is_dir(), + "profile is not a directory: {}", + profile.display() + ); + #[cfg(windows)] + let canonical_profile = short_profile_directory(&canonical_profile).with_context(|| { + format!( + "failed to resolve short-path profile directory {}", + profile.display() + ) + })?; + let lock_path = canonical_profile.join(PROFILE_LEASE_FILE); + let mut options = OpenOptions::new(); + options.read(true).write(true).create(true).truncate(false); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + #[cfg(windows)] + { + use std::os::windows::fs::OpenOptionsExt; + // WHY: sharing reads/writes lets contenders reach the OS lock; + // withholding FILE_SHARE_DELETE prevents replacing the live target. + const FILE_SHARE_READ_WRITE: u32 = 0x0000_0001 | 0x0000_0002; + options.share_mode(FILE_SHARE_READ_WRITE); + } + let file = options.open(&lock_path).with_context(|| { + format!( + "failed to open ownership lock for profile {}; check profile permissions and filesystem locking support", + profile.display() + ) + })?; + match file.try_lock() { + Ok(()) => Ok(Self { directory: canonical_profile, file: Arc::new(file) }), + Err(TryLockError::WouldBlock) => bail!( + "profile is already in use: {}; stop its daemon or offline command before retrying", + profile.display() + ), + Err(TryLockError::Error(error)) => Err(error).with_context(|| { + format!( + "failed to acquire exclusive ownership of profile {}; check permissions and use a filesystem supporting OS file locks", + profile.display() + ) + }), + } + } + + pub(crate) fn directory(&self) -> &Path { + &self.directory + } + + pub(crate) fn file(&self) -> Arc { + Arc::clone(&self.file) + } +} + +#[cfg(windows)] +fn short_profile_directory(canonical: &Path) -> Result { + use std::{ + ffi::OsString, + path::{Component, Prefix}, + }; + + // WHY: canonicalize uses a verbatim prefix on Windows. Keep internal I/O + // on ordinary paths and verify the spelling still names the same directory. + let mut components = canonical.components(); + let Some(Component::Prefix(prefix)) = components.next() else { + bail!("writable profile directory needs an absolute drive or UNC path"); + }; + let mut directory = match prefix.kind() { + Prefix::VerbatimDisk(drive) => PathBuf::from(format!("{}:", char::from(drive))), + Prefix::VerbatimUNC(server, share) => { + let mut unc = OsString::from(r"\\"); + unc.push(server); + unc.push(r"\"); + unc.push(share); + PathBuf::from(unc) + } + Prefix::Disk(_) | Prefix::UNC(_, _) => return Ok(canonical.to_path_buf()), + _ => bail!("writable profile directory needs an ordinary drive or UNC path"), + }; + for component in components { + directory.push(component.as_os_str()); + } + ensure!( + fs::canonicalize(&directory)? == canonical, + "ordinary profile path resolves to a different directory" + ); + Ok(directory) +} diff --git a/crates/emulebb-daemon/src/tests.rs b/crates/emulebb-daemon/src/tests.rs index e273bc42..42704d76 100644 --- a/crates/emulebb-daemon/src/tests.rs +++ b/crates/emulebb-daemon/src/tests.rs @@ -217,12 +217,53 @@ fn put_setting(metadata: &MetadataStore, section: &str, key: &str, value: serde_ .unwrap(); } +#[cfg(unix)] +#[test] +fn retargeted_profile_alias_keeps_io_in_the_leased_directory() { + use std::os::unix::fs::symlink; + + let target = PathBuf::from( + env::var_os("CARGO_TARGET_DIR").expect("profile ownership tests require CARGO_TARGET_DIR"), + ); + assert!(target.is_absolute(), "CARGO_TARGET_DIR must be absolute"); + let scratch = target.join("tmp/profile-ownership-tests"); + fs::create_dir_all(&scratch).unwrap(); + let temporary = tempfile::tempdir_in(scratch).unwrap(); + let leased = temporary.path().join("leased"); + let other = temporary.path().join("other"); + let alias = temporary.path().join("alias"); + fs::create_dir(&leased).unwrap(); + fs::create_dir(&other).unwrap(); + symlink(&leased, &alias).unwrap(); + + let lease = ProfileLease::acquire(&alias, false).unwrap(); + fs::remove_file(&alias).unwrap(); + symlink(&other, &alias).unwrap(); + assert_eq!( + fs::canonicalize(&alias).unwrap(), + fs::canonicalize(&other).unwrap() + ); + + let profile = DaemonProfile::load_owned(alias.clone(), lease, true).unwrap(); + let metadata = profile.open_metadata_store().unwrap(); + put_setting(&metadata, "fixture", "leased", serde_json::json!(true)); + let leased = fs::canonicalize(leased).unwrap(); + assert_eq!(profile.profile_dir, alias); + assert!(leased.join(PROFILE_SETTINGS_FILE).is_file()); + assert_eq!(profile.metadata_path(), leased.join(PROFILE_METADATA_FILE)); + assert!(profile.metadata_path().is_file()); + assert_eq!(profile.transfer_root(), leased.join("transfers")); + assert_eq!(profile.incoming_dir(), leased.join("incoming")); + assert!(fs::read_dir(other).unwrap().next().is_none()); +} + #[test] fn default_profile_bootstrap_is_loopback_authenticated_and_seeds_network_ports() { let temp = tempfile::tempdir().unwrap(); let profile_dir = temp.path().join("profile"); let settings_path = profile_dir.join(PROFILE_SETTINGS_FILE); + create_private_profile_dir(&profile_dir).unwrap(); create_default_profile_bootstrap(&profile_dir, &settings_path).unwrap(); let text = fs::read_to_string(&settings_path).unwrap(); let bootstrap: DaemonBootstrapSettings = toml::from_str(&text).unwrap(); diff --git a/crates/emulebb-daemon/tests/profile_ownership.rs b/crates/emulebb-daemon/tests/profile_ownership.rs new file mode 100644 index 00000000..e5694ca0 --- /dev/null +++ b/crates/emulebb-daemon/tests/profile_ownership.rs @@ -0,0 +1,271 @@ +//! Non-protocol lifecycle regressions using real, isolated OS processes. + +use std::{ + env, fs, + io::{Read, Write}, + path::{Path, PathBuf}, + process::{Child, Command, Stdio}, + thread, + time::{Duration, Instant}, +}; + +use emulebb_daemon::DaemonProfile; + +const DEADLINE: Duration = Duration::from_secs(15); +const CHILD_PROFILE: &str = "PROFILE_OWNERSHIP_TEST_PROFILE"; +const CHILD_MODE: &str = "PROFILE_OWNERSHIP_TEST_MODE"; +const CHILD_READY: &str = "PROFILE_OWNERSHIP_TEST_READY"; + +struct OwnedChild(Child); + +impl OwnedChild { + fn start(profile: &Path, mode: &str, ready: &Path) -> Self { + Self( + Command::new(env::current_exe().unwrap()) + .args([ + "--ignored", + "--exact", + "profile_ownership_child", + "--nocapture", + ]) + .env(CHILD_PROFILE, profile) + .env(CHILD_MODE, mode) + .env(CHILD_READY, ready) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(), + ) + } + + fn finish(&mut self) { + if let Some(mut input) = self.0.stdin.take() { + let _ = input.write_all(b"stop\n"); + } + let deadline = Instant::now() + DEADLINE; + let status = loop { + if let Some(status) = self.0.try_wait().unwrap() { + break status; + } + assert!(Instant::now() < deadline, "profile fixture did not exit"); + thread::sleep(Duration::from_millis(10)); + }; + let mut stderr = String::new(); + if let Some(mut pipe) = self.0.stderr.take() { + pipe.read_to_string(&mut stderr).unwrap(); + } + assert!(status.success(), "profile fixture failed: {stderr}"); + } +} + +impl Drop for OwnedChild { + fn drop(&mut self) { + if self.0.try_wait().ok().flatten().is_none() { + let _ = self.0.kill(); + } + let _ = self.0.wait(); + } +} + +fn temporary() -> tempfile::TempDir { + let target = PathBuf::from( + env::var_os("CARGO_TARGET_DIR").expect("profile ownership tests require CARGO_TARGET_DIR"), + ); + assert!(target.is_absolute(), "CARGO_TARGET_DIR must be absolute"); + let root = target.join("tmp/profile-ownership-tests"); + fs::create_dir_all(&root).unwrap(); + tempfile::Builder::new() + .prefix("lease-") + .tempdir_in(root) + .unwrap() +} + +fn ready(path: &Path) -> String { + let deadline = Instant::now() + DEADLINE; + loop { + if let Ok(value) = fs::read_to_string(path) + && matches!(value.as_str(), "owned" | "busy") + { + return value; + } + assert!( + Instant::now() < deadline, + "profile fixture did not become ready" + ); + thread::sleep(Duration::from_millis(10)); + } +} + +fn probe(profile: &Path, mode: &str, ready: &Path) { + let mut child = OwnedChild::start(profile, mode, ready); + child.finish(); +} + +fn profile_snapshot(profile: &Path) -> [Option>; 4] { + [ + "emulebb-rust-settings.toml", + "emulebb-rust-metadata.db", + "emulebb-rust-metadata.db-wal", + "emulebb-rust-metadata.db-shm", + ] + .map(|name| match fs::read(profile.join(name)) { + Ok(bytes) => Some(bytes), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => None, + Err(error) => panic!("failed to snapshot synthetic profile {name}: {error}"), + }) +} + +#[test] +fn metadata_handles_retain_ownership_after_profile_drop_and_process_death_releases_it() { + let temporary = temporary(); + let profile = temporary.path().join("profile"); + let signal = temporary.path().join("ready"); + let mut holder = OwnedChild::start(&profile, "holder", &signal); + assert_eq!(ready(&signal), "owned"); + // The holder is idle after READY. A denied process must not mutate even + // uncheckpointed WAL state or the shared-memory sidecar, not just the DB. + let before = profile_snapshot(&profile); + probe(&profile, "busy", &temporary.path().join("probe")); + assert!( + before == profile_snapshot(&profile), + "denied profile open changed bootstrap, database, WAL or SHM state" + ); + holder.0.kill().unwrap(); + holder.0.wait().unwrap(); + assert!(profile.join(".emulebb-profile.lock").is_file()); + probe(&profile, "available", &temporary.path().join("after-death")); +} + +#[test] +fn aliases_share_ownership_and_clean_exit_releases_it() { + let temporary = temporary(); + let profile = temporary.path().join("profile"); + let signal = temporary.path().join("ready"); + let mut holder = OwnedChild::start(&profile, "holder", &signal); + assert_eq!(ready(&signal), "owned"); + probe(&profile.join("."), "busy", &temporary.path().join("dot")); + #[cfg(unix)] + { + let alias = temporary.path().join("alias"); + std::os::unix::fs::symlink(&profile, &alias).unwrap(); + probe(&alias, "busy", &temporary.path().join("symlink")); + } + #[cfg(windows)] + { + let alias = PathBuf::from(profile.to_string_lossy().to_uppercase()); + assert!( + alias.is_dir(), + "Windows case-alias fixture requires a case-insensitive volume" + ); + probe(&alias, "busy", &temporary.path().join("case")); + + // Junctions do not require the symlink privilege. Keep both command + // arguments fixed and relative to this test's unique scratch directory. + let windows = PathBuf::from(env::var_os("SystemRoot").expect("Windows system root")); + // WHY: cmd.exe rejects a mixed-separator executable path after launch. + // Join native components to keep the fixture command valid. + let creation = Command::new(windows.join("System32").join("cmd.exe")) + .current_dir(temporary.path()) + .args(["/d", "/v:off", "/c", "mklink", "/J", "junction", "profile"]) + .output() + .expect("create synthetic profile junction"); + assert!( + creation.status.success(), + "could not create synthetic profile junction: {}", + String::from_utf8_lossy(&creation.stderr) + ); + let junction = temporary.path().join("junction"); + assert_eq!( + fs::canonicalize(&junction).unwrap(), + fs::canonicalize(&profile).unwrap() + ); + let before = profile_snapshot(&profile); + probe(&junction, "busy", &temporary.path().join("junction-busy")); + assert!( + before == profile_snapshot(&profile), + "denied junction open changed the owned profile" + ); + } + holder.finish(); + probe(&profile, "available", &temporary.path().join("after-exit")); + #[cfg(windows)] + { + let junction = temporary.path().join("junction"); + probe( + &junction, + "available", + &temporary.path().join("junction-after-exit"), + ); + fs::remove_dir(&junction).unwrap(); + assert!(profile.is_dir(), "junction cleanup removed its target"); + } +} + +#[test] +fn fresh_profile_creation_race_has_exactly_one_owner() { + let temporary = temporary(); + let profile = temporary.path().join("fresh"); + let first_ready = temporary.path().join("first"); + let second_ready = temporary.path().join("second"); + let mut first = OwnedChild::start(&profile, "race", &first_ready); + let mut second = OwnedChild::start(&profile, "race", &second_ready); + let mut outcomes = [ready(&first_ready), ready(&second_ready)]; + outcomes.sort(); + assert_eq!(outcomes, ["busy", "owned"]); + first.finish(); + second.finish(); + probe(&profile, "available", &temporary.path().join("after-race")); +} + +#[test] +fn unrelated_profiles_can_be_owned_concurrently() { + let temporary = temporary(); + let first_ready = temporary.path().join("first-ready"); + let second_ready = temporary.path().join("second-ready"); + let mut first = OwnedChild::start(&temporary.path().join("first"), "holder", &first_ready); + let mut second = OwnedChild::start(&temporary.path().join("second"), "holder", &second_ready); + assert_eq!(ready(&first_ready), "owned"); + assert_eq!(ready(&second_ready), "owned"); + first.finish(); + second.finish(); +} + +#[test] +#[ignore = "isolated subprocess fixture invoked by the ownership regression tests"] +fn profile_ownership_child() { + let profile_path = PathBuf::from(env::var_os(CHILD_PROFILE).unwrap()); + let mode = env::var(CHILD_MODE).unwrap(); + let result = DaemonProfile::load_for_import(profile_path); + if mode == "busy" { + assert!( + result + .unwrap_err() + .to_string() + .contains("profile is already in use") + ); + return; + } + if mode == "race" + && let Err(error) = &result + { + assert!(error.to_string().contains("profile is already in use")); + fs::write(PathBuf::from(env::var_os(CHILD_READY).unwrap()), "busy").unwrap(); + return; + } + let profile = result.unwrap(); + if mode == "available" { + return; + } + let first = profile.open_metadata_store().unwrap(); + let cloned = first.clone(); + let independent = profile.open_metadata_store().unwrap(); + drop(profile); + drop(first); + // The original profile is gone: only separate/cloned metadata handles keep + // ownership alive while this process waits, rather than an early loader. + fs::write(PathBuf::from(env::var_os(CHILD_READY).unwrap()), "owned").unwrap(); + let mut stop = String::new(); + std::io::stdin().read_line(&mut stop).unwrap(); + drop((cloned, independent)); +} diff --git a/crates/emulebb-metadata/src/store.rs b/crates/emulebb-metadata/src/store.rs index 26f96a11..2a7bb613 100644 --- a/crates/emulebb-metadata/src/store.rs +++ b/crates/emulebb-metadata/src/store.rs @@ -1,4 +1,5 @@ use std::{ + fs::File, path::Path, sync::{Arc, Mutex, MutexGuard, TryLockError}, time::{SystemTime, UNIX_EPOCH}, @@ -23,6 +24,9 @@ const TRANSFER_SOURCE_TTL_MS: i64 = 30 * 24 * 60 * 60 * 1000; #[derive(Debug, Clone)] pub struct MetadataStore { conn: Arc>, + // WHY: close the connection before releasing its process profile lease. + // Clones retain both handles; independent profile connections share the lease. + _profile_lease: Option>, } impl MetadataStore { @@ -31,29 +35,51 @@ impl MetadataStore { Self::from_connection(conn) } + /// Open under an already acquired OS-backed writable-profile lease. + /// + /// The caller acquires the lock before bootstrap/metadata initialization. + /// Every cloned store retains it until its connection has finished closing. + pub fn open_with_profile_lease(path: impl AsRef, lease: Arc) -> Result { + let conn = Connection::open(path)?; + Self::from_connection_with_profile_lease(conn, Some(lease)) + } + pub fn in_memory() -> Result { Self::from_connection(Connection::open_in_memory()?) } pub fn from_connection(conn: Connection) -> Result { - conn.pragma_update(None, "journal_mode", "WAL")?; - conn.pragma_update(None, "foreign_keys", "ON")?; - // Crash-consistency: WAL's default `synchronous = NORMAL` does NOT fsync - // the WAL on every commit, so a committed transfer-manifest transaction - // (e.g. a piece marked Verified) can reach SQLite's cache while the - // matching payload bytes are still in the OS cache. On an OS crash / - // power loss this lets the manifest outrace the on-disk bytes and would - // resurrect a "Verified" range whose bytes are stale/zero. FULL fsyncs - // the WAL on commit so a committed manifest state is durable. Paired - // with `sync_all()` on the payload before the piece-complete checkpoint - // commit, neither side can outrace the other. - conn.pragma_update(None, "synchronous", "FULL")?; - // Defensive: bound contention waits instead of failing immediately with - // SQLITE_BUSY (the persistence audit flagged the absent busy_timeout). - conn.pragma_update(None, "busy_timeout", 5000)?; + Self::from_connection_with_profile_lease(conn, None) + } + + fn from_connection_with_profile_lease( + conn: Connection, + profile_lease: Option>, + ) -> Result { + // The guarded constructor retains ownership before any pragma/schema + // initialization, including all early-error exits below. let mut store = Self { conn: Arc::new(Mutex::new(conn)), + _profile_lease: profile_lease, }; + { + let conn = store.connection()?; + conn.pragma_update(None, "journal_mode", "WAL")?; + conn.pragma_update(None, "foreign_keys", "ON")?; + // Crash-consistency: WAL's default `synchronous = NORMAL` does NOT fsync + // the WAL on every commit, so a committed transfer-manifest transaction + // (e.g. a piece marked Verified) can reach SQLite's cache while the + // matching payload bytes are still in the OS cache. On an OS crash / + // power loss this lets the manifest outrace the on-disk bytes and would + // resurrect a "Verified" range whose bytes are stale/zero. FULL fsyncs + // the WAL on commit so a committed manifest state is durable. Paired + // with `sync_all()` on the payload before the piece-complete checkpoint + // commit, neither side can outrace the other. + conn.pragma_update(None, "synchronous", "FULL")?; + // Defensive: bound contention waits instead of failing immediately with + // SQLITE_BUSY (the persistence audit flagged the absent busy_timeout). + conn.pragma_update(None, "busy_timeout", 5000)?; + } store.ensure_schema()?; store.connection()?.execute_batch( "CREATE TEMP TABLE session_verified_local_paths (