fix(lifecycle): share session end with local core supervision

backend/fake-core/src/main.rs

@@ -137,10 +137,11 @@ fn run_start() -> ExitCode {
     // Successful long-running start requires barrier mode.
     let Some(ready_addr) = ready_addr else {
         eprintln!(
-            "fake-core: long-running start requires {} control barrier \
-             (or {} for immediate exit)",
+            "Error: fake-core: long-running start requires {} control barrier \
+             (or {} for immediate exit); pid={}",
             env_keys::READY_ADDR,
-            env_keys::START_EXIT
+            env_keys::START_EXIT,
+            std::process::id()
         );
         return ExitCode::from(2);
     };

backend/nyanpasu-core/src/actor_v2/local_host.rs

@@ -11,6 +11,8 @@ use snafu::{ResultExt, Snafu};
 
 use nyanpasu_paths::PathResolver;
 
+use crate::client::session_end::SessionEnding;
+
 #[derive(Debug, Snafu)]
 pub enum Error {
     #[snafu(display("failed to inspect {}", path.display()))]
@@ -32,6 +34,7 @@ pub enum Error {
 pub async fn build(
     paths: &PathResolver,
     shutdown_helper: &std::path::Path,
+    session_ending: SessionEnding,
 ) -> Result<CoreControl, Error> {
     let runtime_root = paths.app_config_dir().join("runtime");
     let options = ManagerOptions {
@@ -56,7 +59,9 @@ pub async fn build(
             path: config_path.as_std_path(),
         })?,
     ));
-    let manager = CoreManager::builder(options).native_store(native_store);
+    let manager = CoreManager::builder(options)
+        .native_store(native_store)
+        .process_activity(session_ending.process_activity());
 
     #[cfg(target_os = "macos")]
     let manager = manager.dns_controller(Arc::new(
@@ -119,19 +124,120 @@ fn core_spec_with(
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
+    use nyanpasu_core_manager::{
+        ConfigInput, CoreCommand, CoreCommandEnvelope, CoreState, InstanceOptions, OperationId,
+        ReconcileRequest,
+    };
+    use nyanpasu_utils::process::{Backoff, BackoffRange, RestartPolicy, read_epoch_pid_file};
+
     use super::*;
 
     #[tokio::test]
-    async fn the_local_host_spawns_under_a_temp_root() {
+    async fn revoking_session_restores_normal_supervision() {
         let root = tempfile::TempDir::new().unwrap();
         let paths = crate::utils::test_support::test_paths(
             root.path().join("config"),
             root.path().join("data"),
         );
-
-        let control = build(&paths, &root.path().join("helper.exe"))
+        let session = crate::client::session_end::SessionEndSource::new();
+        std::fs::create_dir_all(paths.app_data_dir()).unwrap();
+        let control = build(&paths, &root.path().join("helper.exe"), session.subscribe())
             .await
             .unwrap();
+        let mut starts = control.subscribe_logs();
+        let executable = std::env::current_exe()
+            .unwrap()
+            .ancestors()
+            .skip(1)
+            .map(|directory| directory.join(format!("fake-core{}", std::env::consts::EXE_SUFFIX)))
+            .find(|path| path.is_file())
+            .expect(
+                "build the fake with cargo build --manifest-path backend/Cargo.toml -p fake-core",
+            );
+        session.set_pending(true);
+        let operation = control
+            .submit(CoreCommandEnvelope {
+                operation_id: OperationId::generate(),
+                command: CoreCommand::Reconcile(Box::new(ReconcileRequest {
+                    core: CoreSpec {
+                        kind: CoreKind::Mihomo,
+                        binary_path: Utf8PathBuf::from_path_buf(executable).unwrap(),
+                        version: Some("1.18.9".into()),
+                        features: vec![],
+                    },
+                    config: ConfigInput::inline(b"external-controller: 127.0.0.1:0\n".to_vec()),
+                    options: InstanceOptions {
+                        restart_policy: RestartPolicy::OnFailure { max_restarts: 2 },
+                        backoff: Backoff::exponential(BackoffRange {
+                            initial: Duration::ZERO,
+                            max: Duration::ZERO,
+                        }),
+                        ..InstanceOptions::default()
+                    },
+                    expected_applied: None,
+                })),
+            })
+            .unwrap();
+        let first_start = tokio::time::timeout(Duration::from_secs(30), starts.recv())
+            .await
+            .expect("the first failed start must publish its termination record")
+            .unwrap();
+        let first_pid: u32 = first_start
+            .message
+            .rsplit_once("; pid=")
+            .unwrap()
+            .1
+            .parse()
+            .unwrap();
+        let active = read_epoch_pid_file(paths.app_data_dir().join("native-store/active"))
+            .await
+            .expect("pending session end must prevent preparing another attempt")
+            .expect("the first attempt must remain recorded while paused");
+        assert_eq!(
+            active.pid, first_pid,
+            "the local supervisor started another attempt while session end was pending"
+        );
+        assert!(matches!(
+            starts.try_recv(),
+            Err(tokio::sync::broadcast::error::TryRecvError::Empty)
+        ));
+
+        session.set_pending(false);
+        let error = tokio::time::timeout(Duration::from_secs(30), operation.wait())
+            .await
+            .expect("revocation must resume the deferred retries")
+            .expect_err("the fake core exits without its required start barrier");
+        assert!(error.message.contains("FAKE_CORE_READY_ADDR"), "{error}");
+        let mut resumed_starts = 0;
+        while let Ok(start) = starts.try_recv() {
+            assert!(start.message.contains("; pid="), "{start:?}");
+            resumed_starts += 1;
+        }
+        assert_eq!(
+            resumed_starts, 2,
+            "revocation must resume exactly two retries"
+        );
+        control.shutdown().await.unwrap();
+        assert!(matches!(control.status().state, CoreState::Stopped { .. }));
+    }
+
+    #[tokio::test]
+    async fn the_local_host_spawns_under_a_temp_root() {
+        let root = tempfile::TempDir::new().unwrap();
+        let paths = crate::utils::test_support::test_paths(
+            root.path().join("config"),
+            root.path().join("data"),
+        );
+
+        let control = build(
+            &paths,
+            &root.path().join("helper.exe"),
+            crate::client::session_end::SessionEndSource::new().subscribe(),
+        )
+        .await
+        .unwrap();
 
         let _ = control.status();
         assert!(!control.executor_is_closed());
@@ -144,16 +250,24 @@ mod tests {
         std::fs::write(paths.application_config_path(), "core: mihomo\n").unwrap();
         std::fs::write(paths.nyanpasu_config_path(), "invalid: [").unwrap();
 
-        let control = build(&paths, &root.path().join("helper.exe"))
-            .await
-            .unwrap();
+        let control = build(
+            &paths,
+            &root.path().join("helper.exe"),
+            crate::client::session_end::SessionEndSource::new().subscribe(),
+        )
+        .await
+        .unwrap();
         assert!(!control.executor_is_closed());
 
         std::fs::remove_file(paths.application_config_path()).unwrap();
         assert!(
-            build(&paths, &root.path().join("helper.exe"))
-                .await
-                .is_err()
+            build(
+                &paths,
+                &root.path().join("helper.exe"),
+                crate::client::session_end::SessionEndSource::new().subscribe(),
+            )
+            .await
+            .is_err()
         );
     }
 

backend/nyanpasu-core/src/client/session_end.rs

@@ -1,14 +1,21 @@
 //! Revocable session-end input supplied by the application composition root.
 
+use nyanpasu_utils::process::{ProcessActivity, ProcessActivitySource, ProcessActivityState};
 use tokio::sync::watch;
 
 /// Publishes whether the operating system is preparing to end the session.
 #[derive(Clone, Debug)]
-pub struct SessionEndSource(watch::Sender<bool>);
+pub struct SessionEndSource {
+    activity: ProcessActivitySource,
+    lifetime: watch::Sender<()>,
+}
 
 /// Reads session-end intent without cancelling work already owned by an actor.
 #[derive(Clone, Debug)]
-pub struct SessionEnding(watch::Receiver<bool>);
+pub struct SessionEnding {
+    activity: ProcessActivity,
+    lifetime: watch::Receiver<()>,
+}
 
 impl Default for SessionEndSource {
     fn default() -> Self {
@@ -19,36 +26,47 @@ impl Default for SessionEndSource {
 impl SessionEndSource {
     /// Creates an independent session signal, initially inactive.
     pub fn new() -> Self {
-        Self(watch::channel(false).0)
+        Self {
+            activity: ProcessActivitySource::new(),
+            lifetime: watch::channel(()).0,
+        }
     }
 
     /// Changes the session intent, retaining it even before a reader subscribes.
     pub fn set_pending(&self, pending: bool) {
-        self.0.send_if_modified(|current| {
-            if *current == pending {
-                false
-            } else {
-                *current = pending;
-                true
-            }
+        self.activity.set(if pending {
+            ProcessActivityState::Quiescing
+        } else {
+            ProcessActivityState::Active
         });
     }
 
     /// Creates a reader of this source's current and future intent.
     pub fn subscribe(&self) -> SessionEnding {
-        SessionEnding(self.0.subscribe())
+        SessionEnding {
+            activity: self.activity.subscribe(),
+            lifetime: self.lifetime.subscribe(),
+        }
     }
 }
 
 impl SessionEnding {
     /// Whether a session-end query is pending or has been confirmed.
     pub fn is_pending(&self) -> bool {
-        *self.0.borrow()
+        self.activity.state() == ProcessActivityState::Quiescing
     }
 
     /// Waits for a change; false means the source has been dropped.
     pub async fn changed(&mut self) -> bool {
-        self.0.changed().await.is_ok()
+        tokio::select! {
+            biased;
+            _ = self.activity.changed() => true,
+            _ = self.lifetime.changed() => false,
+        }
+    }
+
+    pub(crate) fn process_activity(&self) -> ProcessActivity {
+        self.activity.clone()
     }
 }
 
@@ -58,6 +76,8 @@ mod tests {
 
     #[tokio::test]
     async fn signal_is_revocable_and_instance_local() {
+        use futures_util::FutureExt;
+
         let source = SessionEndSource::new();
         let other = SessionEndSource::new();
         let mut reader = source.subscribe();
@@ -69,12 +89,21 @@ mod tests {
         source.set_pending(false);
         assert!(reader.changed().await);
         assert!(!reader.is_pending());
+        assert!(reader.changed().now_or_never().is_none());
+        drop(source);
+        assert!(!reader.changed().await);
     }
 
     #[test]
     fn publishing_before_subscription_is_retained() {
         let source = SessionEndSource::new();
         source.set_pending(true);
-        assert!(source.subscribe().is_pending());
+        let reader = source.subscribe();
+        let activity = reader.process_activity();
+        assert!(reader.is_pending());
+        assert_eq!(activity.state(), ProcessActivityState::Quiescing);
+        source.set_pending(false);
+        assert!(!reader.is_pending());
+        assert_eq!(activity.state(), ProcessActivityState::Active);
     }
 }

backend/tauri/src/setup.rs

@@ -261,9 +261,13 @@ pub fn setup<M: tauri::Manager<tauri::Wry>>(
     );
     let span = tracing::info_span!("spawn_core_actors").entered();
     let (core_v2, service, jobs) = tauri::async_runtime::block_on(async {
-        let control = nyanpasu_core::actor_v2::local_host::build(&paths, &service_binary)
-            .await
-            .context(BuildCoreHostSnafu)?;
+        let control = nyanpasu_core::actor_v2::local_host::build(
+            &paths,
+            &service_binary,
+            owners.session_end.subscribe(),
+        )
+        .await
+        .context(BuildCoreHostSnafu)?;
         let local: nyanpasu_core::actor_v2::endpoint::EndpointHandle = Arc::new(
             nyanpasu_core::actor_v2::endpoint::LocalEndpoint::new(control),
         );