Identity activation follow-ups from #341 review

CHANGELOG.md

@@ -18,6 +18,9 @@ Deprecated, Fixed or Security.
 
 ### Fixed
 
+- Identity activation recovery waits for an already-stopping daemon's lifetime
+  lease even after its control socket closes. Interrupt, terminate, and hangup
+  signals cannot interrupt the stop/swap/restart recovery window.
 - Identity preparation removes failed scratch-key writes and sweeps abandoned
   scratch keys under its lock. Only malformed pending keys are replaced;
   pending-key I/O failures are reported without replacing the key.

README.md

@@ -150,6 +150,9 @@ returns the same `next`; the running daemon is unchanged. Over the still-working
 old connection, arrange for every peer to trust `next` with the exact existing
 grants and set the old runtime record's immutable expiry (at most ten minutes)
 in one [atomic runtime-peer batch](docs/runtime-peer-api.md).
+Preparation removes failed scratch-key writes and sweeps abandoned scratch keys
+under the identity lock. A malformed pending key can be replaced; other pending
+key read failures are errors and leave the key unchanged.
 Only after peers have installed that batch:
 
 ```sh
@@ -173,6 +176,10 @@ return `already_active`; they never swap again, and `previous` echoes the caller
 `identity_mismatch` without swapping the identity. The identity lock remains held
 through restart: concurrent `status`, `prepare`, or `activate` returns
 `identity_busy` rather than waiting.
+Interrupt, terminate, and hangup signals cannot interrupt activation's
+stop/swap/restart recovery window, including when the invoking SSH session drops.
+If teardown has closed the control socket, recovery still waits for its lifetime
+lease before restarting the current identity.
 
 After a crash, `status` reports either `prepared` (old key plus pending new key)
 or `active` (the rename completed). `restart_failed` is non-zero and includes

src/main.rs

@@ -3126,7 +3126,59 @@ async fn identity_stop_lease(home: &FabricHome) -> Result<std::fs::File> {
     }).await.map_err(|_| FabricError::new(ErrorCode::DaemonStopTimeout, "daemon did not stop within 15 seconds"))?
 }
 
+/// Activation must finish its rename/restart or old-key recovery after stopping
+/// the daemon. Discard cancellation signals for this scope, then restore the
+/// caller's dispositions. A caught no-op (not SIG_IGN) resets on exec, so a
+/// replacement daemon does not inherit ignored termination signals.
+struct IdentityActivationSignals {
+    previous: [Option<libc::sigaction>; 3],
+}
+
+impl IdentityActivationSignals {
+    const SIGNALS: [libc::c_int; 3] = [libc::SIGINT, libc::SIGTERM, libc::SIGHUP];
+
+    fn install() -> Result<Self> {
+        extern "C" fn finish_activation(_: libc::c_int) {}
+
+        let mut guard = Self { previous: [None, None, None] };
+        // SAFETY: zero initializes the platform sigaction structure; its mask is
+        // initialized below, and the handler neither accesses memory nor calls
+        // anything that could violate async-signal safety.
+        let mut action: libc::sigaction = unsafe { std::mem::zeroed() };
+        action.sa_sigaction = finish_activation as *const () as libc::sighandler_t;
+        action.sa_flags = libc::SA_RESTART;
+        // SAFETY: action owns a valid, writable signal set.
+        if unsafe { libc::sigemptyset(&mut action.sa_mask) } != 0 {
+            return Err(std::io::Error::last_os_error().into());
+        }
+        for (index, signal) in Self::SIGNALS.into_iter().enumerate() {
+            let mut previous = std::mem::MaybeUninit::<libc::sigaction>::uninit();
+            // SAFETY: both pointers refer to live correctly sized structures.
+            // sigaction initializes previous on success and retains no pointers.
+            if unsafe { libc::sigaction(signal, &action, previous.as_mut_ptr()) } != 0 {
+                return Err(std::io::Error::last_os_error().into());
+            }
+            // SAFETY: the successful sigaction call initialized previous.
+            guard.previous[index] = Some(unsafe { previous.assume_init() });
+        }
+        Ok(guard)
+    }
+}
+
+impl Drop for IdentityActivationSignals {
+    fn drop(&mut self) {
+        for (signal, previous) in Self::SIGNALS.into_iter().zip(&self.previous) {
+            if let Some(previous) = previous {
+                // SAFETY: restore exactly the disposition returned by sigaction;
+                // the pointer is valid and no old-action output is requested.
+                unsafe { libc::sigaction(signal, previous, std::ptr::null_mut()); }
+            }
+        }
+    }
+}
+
 async fn activate_identity(home: &FabricHome, current: iroh::EndpointId, next: iroh::EndpointId) -> Result<fabric::identity::Activated> {
+    let _signals = IdentityActivationSignals::install().context("protect identity activation from cancellation signals")?;
     let snapshot_attempted = std::cell::Cell::new(false);
     let stop_attempted = std::cell::Cell::new(false);
     fabric::identity::activate_with_restart(home, current, next, || async {
@@ -3196,7 +3248,14 @@ async fn reconcile_identity_daemon(home: &FabricHome, expected: iroh::EndpointId
         }
     }
     if live.is_some() || !fabric::daemon::daemon_lock_available(home)? {
-        identity_control(home, ControlRequest::Shutdown).await?;
+        match identity_control(home, ControlRequest::Shutdown).await {
+            Ok(_) => {}
+            // A previous Shutdown can have closed the control socket while
+            // teardown still owns the lease. Wait for that teardown, then
+            // restart instead of abandoning pre-rename recovery.
+            Err(error) if fabric::error::code_of(&error) == ErrorCode::DaemonNotRunning => {}
+            Err(error) => return Err(error),
+        }
         drop(identity_stop_lease(home).await?);
     }
     if service::restart_identity_service(home).await? {
@@ -3788,6 +3847,130 @@ async fn spawn_daemon_with_output(home: &FabricHome, options: DaemonOptions, ann
 mod identity_review_tests {
     use super::*;
 
+    fn activation_signal_subprocess(signal: libc::c_int) -> Result<()> {
+        for recovery in [false, true] {
+            let output = ProcessCommand::new(std::env::current_exe()?)
+                .args(["--exact", "identity_review_tests::identity_activation_signal_child", "--nocapture"])
+                .env("FABRIC_ACTIVATION_TEST_SIGNAL", signal.to_string())
+                .env("FABRIC_ACTIVATION_TEST_RECOVERY", recovery.to_string())
+                .output()?;
+            assert!(output.status.success(),
+                "activation signal {signal}, recovery={recovery}: {}\nstdout: {}\nstderr: {}",
+                output.status, String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr));
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn identity_activation_survives_sigint_through_restart_and_recovery() -> Result<()> {
+        activation_signal_subprocess(libc::SIGINT)
+    }
+
+    #[test]
+    fn identity_activation_survives_sigterm_through_restart_and_recovery() -> Result<()> {
+        activation_signal_subprocess(libc::SIGTERM)
+    }
+
+    #[test]
+    fn identity_activation_survives_sighup_through_restart_and_recovery() -> Result<()> {
+        activation_signal_subprocess(libc::SIGHUP)
+    }
+
+    #[tokio::test]
+    async fn identity_activation_signal_child() -> Result<()> {
+        use std::os::unix::process::ExitStatusExt;
+        use tokio::io::{AsyncBufReadExt, BufReader};
+
+        let Ok(signal) = std::env::var("FABRIC_ACTIVATION_TEST_SIGNAL") else { return Ok(()); };
+        let signal: libc::c_int = signal.parse()?;
+        let recovery = std::env::var("FABRIC_ACTIVATION_TEST_RECOVERY")? == "true";
+        assert!(IdentityActivationSignals::SIGNALS.contains(&signal));
+        // SAFETY: only this disposable subprocess is affected. Establish a
+        // terminating disposition so each regression fails without the guard.
+        unsafe { libc::signal(signal, libc::SIG_DFL); }
+        let dir = tempfile::tempdir()?;
+        let home = FabricHome::new(dir.path());
+        let prepared = fabric::identity::prepare(&home)?;
+        let lease = fabric::daemon::DaemonLease::acquire(&home)?;
+        let listener = tokio::net::UnixListener::bind(home.control_socket_path())?;
+        let server_home = home.clone();
+        // Model the control boundary without starting network endpoints. Signals
+        // are delivered synchronously at Shutdown, strictly before the rename,
+        // and again during post-stop reconciliation / pre-rename recovery.
+        let server = tokio::spawn(async move {
+            let mut lease = Some(lease);
+            let mut stopped = false;
+            let mut post_stop_signals = 0;
+            loop {
+                let (stream, _) = listener.accept().await?;
+                let mut stream = BufReader::new(stream);
+                let mut line = String::new();
+                stream.read_line(&mut line).await?;
+                let request: ControlRequest = serde_json::from_str(&line)?;
+                let response = match request {
+                    ControlRequest::Status => {
+                        if stopped {
+                            // SAFETY: raise targets only this subprocess and
+                            // uses one of the validated catchable signals.
+                            assert_eq!(unsafe { libc::raise(signal) }, 0);
+                            post_stop_signals += 1;
+                        }
+                        ControlResponse::Status {
+                            node_id: load_or_create_identity(&server_home)?.public().to_string(),
+                            endpoint_addr: serde_json::Value::Null,
+                            exposed_protocols: vec![], dial_sockets: vec![],
+                            allow_shell: false, allow_exec: false, audit: None,
+                        }
+                    }
+                    ControlRequest::IdentitySnapshot { .. } => {
+                        ControlResponse::IdentitySnapshot { runtime_peers: false }
+                    }
+                    ControlRequest::Shutdown => {
+                        assert_eq!(load_or_create_identity(&server_home)?.public(), prepared.current);
+                        drop(lease.take());
+                        stopped = true;
+                        if recovery {
+                            // Force the actual pre-rename error/recovery path.
+                            fs::remove_file(server_home.root().join("identity.next.toml"))?;
+                        }
+                        // SAFETY: same subprocess-local signal delivery as above.
+                        assert_eq!(unsafe { libc::raise(signal) }, 0);
+                        // Exec'd replacement/service children must still obey
+                        // termination; SIG_IGN would incorrectly survive exec.
+                        let child = ProcessCommand::new("sh")
+                            .args(["-c", &format!("kill -{signal} $$; exit 0")])
+                            .status()?;
+                        assert_eq!(child.signal(), Some(signal));
+                        ControlResponse::Ok
+                    }
+                    ControlRequest::IdentityAbort { .. } => ControlResponse::Ok,
+                    request => bail!("unexpected activation request: {request:?}"),
+                };
+                stream.get_mut().write_all(&serde_json::to_vec(&response)?).await?;
+                stream.get_mut().shutdown().await?;
+                if stopped && ((!recovery && post_stop_signals == 1) || (recovery && post_stop_signals == 2)) {
+                    return Ok::<_, anyhow::Error>(());
+                }
+            }
+        });
+        let result = activate_identity(&home, prepared.current, prepared.next).await;
+        if recovery {
+            let error = result.err().context("activation should fail before rename")?;
+            assert!(format!("{error:#}").contains("current identity daemon restarted successfully"));
+            assert_eq!(load_or_create_identity(&home)?.public(), prepared.current);
+        } else {
+            assert_eq!(result?.active, prepared.next);
+            assert_eq!(load_or_create_identity(&home)?.public(), prepared.next);
+        }
+        server.await??;
+        let mut restored = std::mem::MaybeUninit::<libc::sigaction>::uninit();
+        // SAFETY: querying with a null action only writes the live output.
+        assert_eq!(unsafe { libc::sigaction(signal, std::ptr::null(), restored.as_mut_ptr()) }, 0);
+        // SAFETY: the successful query initialized restored.
+        assert_eq!(unsafe { restored.assume_init() }.sa_sigaction, libc::SIG_DFL);
+        Ok(())
+    }
+
     #[tokio::test]
     async fn identity_activation_waits_for_an_already_stopping_foreground_node() -> Result<()> {
         let dir = tempfile::tempdir()?;

tests/json_output.rs

@@ -289,6 +289,48 @@ async fn identity_foreground_activation_failed_swap_and_stale_socket_recovery_ke
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn identity_stop_timeout_recovery_waits_for_teardown_and_restarts_current() -> Result<()> {
+    use fabric::{control::{ControlRequest, ControlResponse}, daemon::{DaemonLease, send_control}};
+    struct Stop(FabricHome);
+    impl Drop for Stop {
+        fn drop(&mut self) {
+            let _ = Command::new(fabric_bin()).arg("--home").arg(self.0.root()).arg("down").output();
+        }
+    }
+    let dir = TempDir::new()?;
+    let home = FabricHome::new(dir.path());
+    let prepared = fabric::identity::prepare(&home)?;
+    let lease = DaemonLease::acquire(&home)?;
+    let node = FabricNode::start(home.clone()).await?;
+    let _stop = Stop(home.clone());
+    // Model a daemon which acknowledged Shutdown and stopped listening, but
+    // retains its lifetime lease beyond the activation's 15-second deadline.
+    let teardown = tokio::spawn(async move {
+        node.wait().await?;
+        tokio::time::sleep(Duration::from_secs(18)).await;
+        drop(lease);
+        Ok::<_, anyhow::Error>(())
+    });
+    let output = tokio::time::timeout(Duration::from_secs(45),
+        tokio::process::Command::new(fabric_bin())
+            .arg("--home").arg(home.root())
+            .args(["identity", "activate", "--expect-current", &prepared.current.to_string(),
+                "--expect-next", &prepared.next.to_string(), "--json"])
+            .kill_on_drop(true).output()
+    ).await.context("activation recovery deadline")??;
+    let result: Value = serde_json::from_slice(&output.stdout)?;
+    assert!(!output.status.success());
+    assert_eq!(result["error"]["code"], "daemon_stop_timeout");
+    assert!(String::from_utf8_lossy(&output.stdout).contains("current identity daemon restarted successfully"),
+        "{}", String::from_utf8_lossy(&output.stdout));
+    teardown.await??;
+    assert_eq!(fabric::identity::status(&home)?.next, Some(prepared.next));
+    assert!(matches!(send_control(&home, ControlRequest::Status).await?,
+        ControlResponse::Status { node_id, .. } if node_id == prepared.current.to_string()));
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn identity_snapshot_abort_removes_handoff_and_unfences_exposures() -> Result<()> {
     use fabric::{control::{ControlRequest, ControlResponse}, daemon::send_control};