fix(profile): retain exclusive ownership through physical profile I/O

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
 }

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<ProfileLease>,
 }
 
 #[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<Self> {
-        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<Self> {
+        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,19 +221,41 @@ 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<MetadataStore> {
+        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
     /// `incomingDir`, or `<profile>/incoming` by default.
     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<SocketAddr>,
 }
 
-async fn run_inner(profile: DaemonProfile, host: Option<HostedLifecycle>) -> Result<()> {
+async fn run_inner(mut profile: DaemonProfile, host: Option<HostedLifecycle>) -> 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<HostedLifecycle>) -> 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<HostedLifecycle>) -> 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?;

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,

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<File>,
+}
+
+impl ProfileLease {
+    pub(crate) fn acquire(profile: &Path, create_if_missing: bool) -> Result<Self> {
+        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<File> {
+        Arc::clone(&self.file)
+    }
+}
+
+#[cfg(windows)]
+fn short_profile_directory(canonical: &Path) -> Result<PathBuf> {
+    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)
+}