Merge pull request #478 from HashemKhalifa/codex/fix-worktree-store-resolution

fix: stabilize worktree storage and daemon lifecycle

Author: ScriptedAlchemy

src/daemon.rs

@@ -120,6 +120,13 @@ impl StoreAdministration {
         ensure_no_external_branch_store_holders(database_paths)
     }
 
+    /// Reports whether writer administration is already held. Work that would
+    /// queue behind an unrelated writer can be delayed for that writer's whole
+    /// operation, which callers may want to answer with a retry hint instead.
+    pub(super) fn writer_is_busy(&self) -> bool {
+        self.gate.try_lock().is_err()
+    }
+
     /// Acquires writer administration before constructing the supplied future
     /// and holds it until that future completes.
     pub(super) async fn with_writer<Operation, OperationFuture, Output>(

src/daemon/branch_admin.rs

@@ -9,7 +9,7 @@ use crate::errors::{Result, TraceDecayError};
 
 use super::{
     DAEMON_TASK_ABORT_DEADLINE, DaemonEngine, DaemonHandshake, ProjectServerKey, log_daemon_event,
-    open_existing_project_with_options,
+    open_existing_project_with_options, spawn_lifecycle_automation_scheduler_activation,
 };
 
 pub(super) fn scheduler_task_log_fields(
@@ -158,12 +158,73 @@ impl DaemonEngine {
         key: ProjectServerKey,
         project_path: PathBuf,
         handshake: DaemonHandshake,
+        cg: Arc<crate::tracedecay::TraceDecay>,
     ) {
+        if !self.lifecycle.accepting() {
+            return;
+        }
+        {
+            let schedulers = self
+                .store_administration
+                .automation_schedulers()
+                .lock()
+                .await;
+            if schedulers.contains_key(&key) {
+                return;
+            }
+        }
+
+        // Discover configuration from the project server that is already
+        // registered for this owner. This keeps writable project opening and
+        // identity repair under the project-server coordination path while
+        // avoiding the daemon-wide writer gate for read-only discovery.
+        let configured = match async {
+            let config =
+                effective_automation_config_for_project(&cg, &handshake.client_identity).await?;
+            automation_scheduler_has_work(&cg, &config).await
+        }
+        .await
+        {
+            Ok(configured) => configured,
+            Err(e) => {
+                log_daemon_event(
+                    "scheduler_config",
+                    &[
+                        ("project", project_path.display().to_string()),
+                        ("outcome", "error".to_string()),
+                        ("error", e.to_string()),
+                    ],
+                );
+                false
+            }
+        };
+        if !configured {
+            log_daemon_event(
+                "scheduler_config",
+                &[
+                    ("project", project_path.display().to_string()),
+                    ("outcome", "skipped".to_string()),
+                    ("reason", "not_configured".to_string()),
+                ],
+            );
+            return;
+        }
+
         self.store_administration
             .with_writer(|| async move {
                 if !self.lifecycle.accepting() {
                     return;
                 }
+                let owner_is_current = self
+                    .store_administration
+                    .project_servers()
+                    .lock()
+                    .await
+                    .get(&key)
+                    .is_some();
+                if !owner_is_current {
+                    return;
+                }
                 {
                     let schedulers = self
                         .store_administration
@@ -174,38 +235,6 @@ impl DaemonEngine {
                         return;
                     }
                 }
-
-                let configured = match Box::pin(automation_scheduler_has_work_for_project(
-                    &project_path,
-                    &handshake,
-                ))
-                .await
-                {
-                    Ok(configured) => configured,
-                    Err(e) => {
-                        log_daemon_event(
-                            "scheduler_config",
-                            &[
-                                ("project", project_path.display().to_string()),
-                                ("outcome", "error".to_string()),
-                                ("error", e.to_string()),
-                            ],
-                        );
-                        false
-                    }
-                };
-                if !configured {
-                    log_daemon_event(
-                        "scheduler_config",
-                        &[
-                            ("project", project_path.display().to_string()),
-                            ("outcome", "skipped".to_string()),
-                            ("reason", "not_configured".to_string()),
-                        ],
-                    );
-                    return;
-                }
-
                 self.start_automation_scheduler(key, project_path, handshake)
                     .await;
             })
@@ -224,10 +253,24 @@ impl DaemonEngine {
             let current_key = Arc::clone(&current_key);
             let project_path = project_path.clone();
             let handshake = handshake.clone();
-            tokio::spawn(async move {
+            let lifecycle = engine.lifecycle.clone();
+            spawn_lifecycle_automation_scheduler_activation(lifecycle, async move {
                 let key = current_key.lock().await.clone();
+                let server = {
+                    engine
+                        .store_administration
+                        .project_servers()
+                        .lock()
+                        .await
+                        .get(&key)
+                        .cloned()
+                };
+                let Some(server) = server else {
+                    return;
+                };
+                let cg = server.cg().await;
                 engine
-                    .ensure_automation_scheduler(key.clone(), project_path, handshake)
+                    .ensure_automation_scheduler(key.clone(), project_path, handshake, cg)
                     .await;
                 if let Some(handle) = engine
                     .store_administration

src/daemon/scheduler.rs

@@ -12,6 +12,9 @@ use super::SOCKET_ENV;
 
 const LAUNCHD_LABEL: &str = "com.tracedecay.daemon";
 const LAUNCHD_PLIST_NAME: &str = "com.tracedecay.daemon.plist";
+// Cached project owners retain SQLite families and coordination locks. The
+// platform default of 256 descriptors is too small for multi-worktree use.
+const DAEMON_OPEN_FILE_LIMIT: u32 = 8_192;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct DaemonServiceSpec {
@@ -54,12 +57,14 @@ impl DaemonServiceSpec {
              ExecStart={} daemon run --socket {}\n\
              Restart=on-failure\n\
              RestartSec=2\n\
+             LimitNOFILE={}\n\
              \n\
              [Install]\n\
              WantedBy=default.target\n",
             systemd_escape_env_value(&service_path),
             self.tracedecay_bin.display(),
-            self.socket_path.display()
+            self.socket_path.display(),
+            DAEMON_OPEN_FILE_LIMIT,
         )
     }
 
@@ -136,6 +141,12 @@ impl DaemonServiceSpec {
              \n\
                <key>ThrottleInterval</key>\n\
                <integer>2</integer>\n\
+             \n\
+               <key>SoftResourceLimits</key>\n\
+               <dict>\n\
+                 <key>NumberOfFiles</key>\n\
+                 <integer>{open_file_limit}</integer>\n\
+               </dict>\n\
              \n\
                <key>StandardOutPath</key>\n\
                <string>{stdout}</string>\n\
@@ -147,6 +158,7 @@ impl DaemonServiceSpec {
             label = plist_xml_escape(LAUNCHD_LABEL),
             bin = plist_xml_escape(&self.tracedecay_bin.display().to_string()),
             socket = plist_xml_escape(&self.socket_path.display().to_string()),
+            open_file_limit = DAEMON_OPEN_FILE_LIMIT,
             stdout = plist_xml_escape(&data_dir.join("daemon.out.log").display().to_string()),
             stderr = plist_xml_escape(&data_dir.join("daemon.err.log").display().to_string()),
         ))
@@ -373,8 +385,41 @@ fn refresh_installed_service_with_state(
     refresh_service_with_runner(&runner, &refreshed_spec, previous_state).map(Some)
 }
 
+/// Stops a managed daemon before `daemon restart` acquires exclusive lifecycle
+/// ownership. The daemon itself holds a shared lease while it is running.
+#[doc(hidden)]
+pub fn quiesce_installed_service_for_restart() -> Result<DaemonServiceState> {
+    quiesce_installed_service()
+}
+
 #[doc(hidden)]
 pub fn quiesce_installed_service_under_lease() -> Result<DaemonServiceState> {
+    quiesce_installed_service()
+}
+
+/// Restores an installed service that was running before restart quiesced it.
+/// This deliberately reuses the existing unit instead of rewriting it because
+/// the caller invokes it only after exclusive lifecycle acquisition failed.
+#[doc(hidden)]
+pub fn restore_quiesced_installed_service(previous_state: DaemonServiceState) -> Result<()> {
+    if !cfg!(any(target_os = "linux", target_os = "macos")) || !previous_state.is_running() {
+        return Ok(());
+    }
+    let service_path = service_unit_path()?;
+    if !service_path.exists() {
+        return Err(TraceDecayError::Config {
+            message: format!(
+                "cannot restore missing TraceDecay daemon service '{}'",
+                service_path.display()
+            ),
+        });
+    }
+    let unit = read_service_unit(&service_path)?;
+    let socket_path = socket_path_from_unit_text(&unit).unwrap_or(default_socket_path()?);
+    ServiceRunner::current()?.restore_after_quiesce(&service_path, &socket_path, previous_state)
+}
+
+fn quiesce_installed_service() -> Result<DaemonServiceState> {
     if !cfg!(any(target_os = "linux", target_os = "macos")) {
         return Ok(DaemonServiceState::Missing);
     }
@@ -758,6 +803,27 @@ impl ServiceRunner {
         }
     }
 
+    fn restore_after_quiesce(
+        &self,
+        service_path: &Path,
+        socket_path: &Path,
+        previous_state: DaemonServiceState,
+    ) -> Result<()> {
+        if !previous_state.is_running() {
+            return Ok(());
+        }
+        match self {
+            Self::Systemd => run_systemctl(&["start", super::SERVICE_NAME]),
+            Self::Launchd => {
+                launchd_refresh(service_path, socket_path)?;
+                if !previous_state.is_enabled() {
+                    run_launchctl(&["disable", &launchd_service_target()?])?;
+                }
+                Ok(())
+            }
+        }
+    }
+
     fn after_uninstall(&self, stop: bool) {
         match self {
             Self::Systemd => {
@@ -1224,6 +1290,7 @@ mod tests {
         ));
         assert!(unit.contains("Environment=\"PATH="));
         assert!(unit.contains("Restart=on-failure"));
+        assert!(unit.contains("LimitNOFILE=8192"));
     }
 
     // The launchd render tests use Unix-style absolute binary paths, which
@@ -1265,6 +1332,9 @@ mod tests {
         assert!(plist.contains("<key>TRACEDECAY_DATA_DIR</key>"));
         assert!(plist.contains("<key>RunAtLoad</key>"));
         assert!(plist.contains("<key>KeepAlive</key>"));
+        assert!(plist.contains("<key>SoftResourceLimits</key>"));
+        assert!(plist.contains("<key>NumberOfFiles</key>"));
+        assert!(plist.contains("<integer>8192</integer>"));
     }
 
     #[cfg(unix)]
@@ -1619,6 +1689,55 @@ mod tests {
         );
     }
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn restore_quiesced_service_starts_existing_unit_without_rewriting_it() {
+        let _env_lock = lock_user_data_dir_test_env();
+        let dir = TempDir::new().expect("temp dir");
+        let config_home = dir.path().join("config");
+        let fake_bin = dir.path().join("bin");
+        let home = dir.path().join("home");
+        std::fs::create_dir_all(&fake_bin).expect("fake bin dir");
+        std::fs::create_dir_all(&home).expect("home dir");
+
+        let systemctl = fake_bin.join("systemctl");
+        let log = dir.path().join("systemctl.log");
+        std::fs::write(
+            &systemctl,
+            "#!/bin/sh\nprintf '%s\\n' \"$*\" >> \"$TRACEDECAY_SYSTEMCTL_LOG\"\nexit 0\n",
+        )
+        .expect("fake systemctl");
+        std::fs::set_permissions(&systemctl, std::fs::Permissions::from_mode(0o755))
+            .expect("systemctl permissions");
+
+        let _config_guard = EnvVarGuard::set("XDG_CONFIG_HOME", &config_home);
+        let _home_guard = EnvVarGuard::set("HOME", &home);
+        let _data_guard =
+            EnvVarGuard::set(crate::config::USER_DATA_DIR_ENV, dir.path().join("profile"));
+        let _path_guard = EnvVarGuard::set("PATH", &fake_bin);
+        let _log_guard = EnvVarGuard::set("TRACEDECAY_SYSTEMCTL_LOG", &log);
+        let service_path = config_home
+            .join("systemd/user")
+            .join(crate::daemon::SERVICE_NAME);
+        std::fs::create_dir_all(service_path.parent().expect("service parent"))
+            .expect("service dir");
+        let original_unit =
+            "[Service]\nExecStart=/old/tracedecay daemon run --socket /custom/tracedecay.sock\n";
+        std::fs::write(&service_path, original_unit).expect("existing service unit");
+
+        super::restore_quiesced_installed_service(DaemonServiceState::RunningEnabled)
+            .expect("restore service");
+
+        assert_eq!(
+            std::fs::read_to_string(service_path).expect("service unit"),
+            original_unit
+        );
+        assert_eq!(
+            std::fs::read_to_string(log).expect("systemctl log"),
+            "--user start tracedecay.service\n"
+        );
+    }
+
     #[cfg(target_os = "linux")]
     #[test]
     fn refresh_installed_service_preserves_stopped_state() {