Merge pull request #5074 from loopx-project/codex/desktop-parallel-service-startup

Author: huangruiteng

apps/desktop/loopx-control-plane/src-tauri/src/lib.rs

@@ -229,6 +229,11 @@ mod tests {
         assert!(style.contains("--warning: #f5a623"));
         assert!(script.contains("desktop_update_status"));
         assert!(script.contains("window.loopxBootRetrying"));
+        // Services connect concurrently, so the phase names the loopback set
+        // until one connection outlives its peer and can be named on its own.
+        assert!(script.contains("正在连接本地服务"));
+        assert!(script.contains("正在连接状态服务"));
+        assert!(script.contains("正在连接管家对话服务"));
         // The first screen must offer both operator choices, not a repair path
         // that silently replaces the CLI runtime.
         assert!(html.contains("id=\"pairing-align\""));

apps/desktop/loopx-control-plane/src-tauri/src/maintenance.rs

@@ -749,9 +749,10 @@ pub fn start_services(app: &AppHandle) -> Result<Option<crate::services::Service
             }
             return Err(error);
         }
-        crate::services::ServiceSet::start(|kind| {
+        crate::services::ServiceSet::start(|pending| {
+            let service = crate::services::ServiceKind::pending_label(pending);
             app.state::<Maintenance>()
-                .publish("connecting", json!({"service":kind.label()}));
+                .publish("connecting", json!({"service":service}));
         })
         .map_err(|e| e.to_string())
     })

apps/desktop/loopx-control-plane/src-tauri/src/services.rs

@@ -7,10 +7,14 @@ use std::{
     net::{SocketAddr, TcpStream},
     path::{Path, PathBuf},
     process::{Command, Stdio},
+    sync::Mutex,
     thread,
     time::{Duration, Instant},
 };
 
+/// Every loopback service the App must reach before it opens the workspace.
+pub const SERVICE_KINDS: [ServiceKind; 2] = [ServiceKind::Status, ServiceKind::Chat];
+
 const STARTUP_TIMEOUT: Duration = Duration::from_secs(15);
 const PROBE_TIMEOUT: Duration = Duration::from_millis(500);
 const MAX_PROBE_RESPONSE_BYTES: u64 = 1024 * 1024;
@@ -29,6 +33,17 @@ impl ServiceKind {
         }
     }
 
+    /// Name the services a `connecting` phase is still waiting for. One
+    /// pending service keeps its own name so a stalled connection stays
+    /// diagnosable on the boot page; a concurrent connect reports the loopback
+    /// set, which the boot page renders as "local services".
+    pub fn pending_label(pending: &[Self]) -> &'static str {
+        match pending {
+            [kind] => kind.label(),
+            _ => "local",
+        }
+    }
+
     fn port(self) -> u16 {
         match self {
             Self::Status => 8766,
@@ -118,115 +133,157 @@ pub struct ServiceSet {
 }
 
 impl ServiceSet {
-    pub fn start(mut progress: impl FnMut(ServiceKind)) -> Result<Self, ServiceError> {
+    pub fn start(progress: impl Fn(&[ServiceKind]) + Sync) -> Result<Self, ServiceError> {
+        Self::collect(connect_all(SERVICE_KINDS, connect, progress))
+    }
+
+    /// Fold finished connection attempts into one owned set. Every outcome
+    /// surrenders its child here, so a set that fails still stops the
+    /// processes its successful peers started.
+    fn collect(outcomes: [ServiceOutcome; SERVICE_KINDS.len()]) -> Result<Self, ServiceError> {
         let mut services = Self {
             owned: Vec::new(),
             healed: false,
         };
-        for kind in [ServiceKind::Status, ServiceKind::Chat] {
-            progress(kind);
-            if let Err(error) = services.ensure(kind) {
+        let mut failure = None;
+        for outcome in outcomes {
+            services.owned.extend(outcome.owned);
+            services.healed |= outcome.healed;
+            if let Err(error) = outcome.result {
+                failure.get_or_insert(error);
+            }
+        }
+        match failure {
+            Some(error) => {
                 services.stop();
-                return Err(error);
+                Err(error)
             }
+            None => Ok(services),
         }
-        Ok(services)
     }
 
-    fn ensure(&mut self, kind: ServiceKind) -> Result<(), ServiceError> {
-        let executable = loopx_executable();
-        let expected_runtime_identity = runtime_identity_for_executable(&executable);
-        let stale_deadline = Instant::now() + STARTUP_TIMEOUT;
-        loop {
-            match probe(kind, expected_runtime_identity.as_ref()) {
-                Probe::Matching => return Ok(()),
-                Probe::NotReady => return Err(status_readiness_error(kind)),
-                Probe::Foreign => {
+    pub fn stop(&mut self) {
+        for service in self.owned.iter_mut().rev() {
+            service.stop();
+        }
+        self.owned.clear();
+    }
+}
+
+/// One service's connection attempt. The child this App spawned travels with
+/// the outcome even when the attempt failed, so `ServiceSet` can stop it
+/// instead of leaking a process that no longer has an owner.
+struct ServiceOutcome {
+    owned: Option<OwnedService>,
+    healed: bool,
+    result: Result<(), ServiceError>,
+}
+
+/// Connect every loopback service at once.
+///
+/// The services own separate ports, commands and processes, and neither reads
+/// the other's readiness, so the window should wait for the slowest one rather
+/// than their sum. A start that follows a runtime update pays that difference
+/// twice over: each stale listener is replaced and then warms a fresh
+/// interpreter before it answers a readiness probe.
+///
+/// `progress` names the services still being waited on: the whole set while
+/// they run together, then whichever connection outlives its peer, so a
+/// stalled service is still named on the boot page.
+fn connect_all<const N: usize>(
+    kinds: [ServiceKind; N],
+    connect: impl Fn(ServiceKind) -> ServiceOutcome + Sync,
+    progress: impl Fn(&[ServiceKind]) + Sync,
+) -> [ServiceOutcome; N] {
+    let pending = Mutex::new(kinds.to_vec());
+    progress(&kinds);
+    thread::scope(|scope| {
+        kinds
+            .map(|kind| {
+                let (connect, progress, pending) = (&connect, &progress, &pending);
+                scope.spawn(move || {
+                    let outcome = connect(kind);
+                    let remaining = {
+                        let mut pending = pending.lock().expect("pending service lock");
+                        pending.retain(|entry| *entry != kind);
+                        pending.clone()
+                    };
+                    if !remaining.is_empty() {
+                        progress(&remaining);
+                    }
+                    outcome
+                })
+            })
+            .map(|handle| handle.join().expect("service connection thread"))
+    })
+}
+
+fn connect(kind: ServiceKind) -> ServiceOutcome {
+    let mut owned = None;
+    let mut healed = false;
+    let result = connect_service(kind, &mut owned, &mut healed);
+    ServiceOutcome {
+        owned,
+        healed,
+        result,
+    }
+}
+
+fn connect_service(
+    kind: ServiceKind,
+    owned: &mut Option<OwnedService>,
+    healed: &mut bool,
+) -> Result<(), ServiceError> {
+    let executable = loopx_executable();
+    let expected_runtime_identity = runtime_identity_for_executable(&executable);
+    let stale_deadline = Instant::now() + STARTUP_TIMEOUT;
+    loop {
+        match probe(kind, expected_runtime_identity.as_ref()) {
+            Probe::Matching => return Ok(()),
+            Probe::NotReady => return Err(status_readiness_error(kind)),
+            Probe::Foreign => {
+                return Err(ServiceError(format!(
+                    "port {} is occupied by a service that is not LoopX {}",
+                    kind.port(),
+                    kind.label()
+                )));
+            }
+            Probe::Stale => {
+                // Self-heal: the port is owned by a LoopX service from a
+                // different installed release (for example after a
+                // `loopx update`). Terminate that stale listener and keep
+                // waiting up to the startup timeout so a LaunchAgent-managed
+                // service (KeepAlive + throttle) has time to restart on the
+                // current release; unknown (Foreign) processes keep the
+                // hard error.
+                terminate_verified_listener(kind, &executable, kind.port())?;
+                *healed = true;
+                if Instant::now() >= stale_deadline {
                     return Err(ServiceError(format!(
-                        "port {} is occupied by a service that is not LoopX {}",
-                        kind.port(),
-                        kind.label()
-                    )));
-                }
-                Probe::Stale => {
-                    // Self-heal: the port is owned by a LoopX service from a
-                    // different installed release (for example after a
-                    // `loopx update`). Terminate that stale listener and keep
-                    // waiting up to the startup timeout so a LaunchAgent-managed
-                    // service (KeepAlive + throttle) has time to restart on the
-                    // current release; unknown (Foreign) processes keep the
-                    // hard error.
-                    terminate_verified_listener(kind, &executable, kind.port())?;
-                    self.healed = true;
-                    if Instant::now() >= stale_deadline {
-                        return Err(ServiceError(format!(
                             "port {} is serving LoopX {} from a different installed runtime and could not be restarted",
                             kind.port(),
                             kind.label()
                         )));
-                    }
-                    thread::sleep(Duration::from_millis(200));
-                }
-                Probe::Unresponsive => {
-                    // A bound socket is not HTTP readiness. Give slow startup
-                    // a full grace period, then replace only a verified LoopX
-                    // listener; unknown processes still fail closed.
-                    if Instant::now() < stale_deadline {
-                        thread::sleep(Duration::from_millis(100));
-                        continue;
-                    }
-                    terminate_verified_listener(kind, &executable, kind.port())?;
-                    self.healed = true;
-                    break;
                 }
-                Probe::Unavailable => break,
+                thread::sleep(Duration::from_millis(200));
             }
-        }
-
-        if request_platform_managed_start(kind) {
-            let deadline = Instant::now() + STARTUP_TIMEOUT;
-            while Instant::now() < deadline {
-                match probe(kind, expected_runtime_identity.as_ref()) {
-                    Probe::Matching => return Ok(()),
-                    Probe::NotReady => return Err(status_readiness_error(kind)),
-                    Probe::Foreign => {
-                        return Err(ServiceError(format!(
-                            "LoopX {} startup reached an unexpected service on port {}",
-                            kind.label(),
-                            kind.port()
-                        )));
-                    }
-                    Probe::Stale => {
-                        terminate_verified_listener(kind, &executable, kind.port())?;
-                        self.healed = true;
-                        request_platform_managed_start(kind);
-                    }
-                    Probe::Unavailable | Probe::Unresponsive => {}
+            Probe::Unresponsive => {
+                // A bound socket is not HTTP readiness. Give slow startup
+                // a full grace period, then replace only a verified LoopX
+                // listener; unknown processes still fail closed.
+                if Instant::now() < stale_deadline {
+                    thread::sleep(Duration::from_millis(100));
+                    continue;
                 }
-                thread::sleep(Duration::from_millis(100));
+                terminate_verified_listener(kind, &executable, kind.port())?;
+                *healed = true;
+                break;
             }
-            return Err(ServiceError(format!(
-                "system-managed LoopX {} did not become ready on port {}",
-                kind.label(),
-                kind.port()
-            )));
+            Probe::Unavailable => break,
         }
+    }
 
-        let mut command = Command::new(&executable);
-        configure_runtime_environment(&mut command);
-        command
-            .args(kind.command_args())
-            .stdin(Stdio::null())
-            .stdout(Stdio::null())
-            .stderr(Stdio::null());
-        let child = command.group_spawn().map_err(|error| {
-            ServiceError(format!(
-                "could not start LoopX {} with `{executable}`: {error}",
-                kind.label()
-            ))
-        })?;
-        self.owned.push(OwnedService { child });
-
+    if request_platform_managed_start(kind) {
         let deadline = Instant::now() + STARTUP_TIMEOUT;
         while Instant::now() < deadline {
             match probe(kind, expected_runtime_identity.as_ref()) {
@@ -241,27 +298,60 @@ impl ServiceSet {
                 }
                 Probe::Stale => {
                     terminate_verified_listener(kind, &executable, kind.port())?;
-                    self.healed = true;
-                    thread::sleep(Duration::from_millis(200));
-                }
-                Probe::Unavailable | Probe::Unresponsive => {
-                    thread::sleep(Duration::from_millis(100))
+                    *healed = true;
+                    request_platform_managed_start(kind);
                 }
+                Probe::Unavailable | Probe::Unresponsive => {}
             }
+            thread::sleep(Duration::from_millis(100));
         }
-        Err(ServiceError(format!(
-            "LoopX {} did not become ready on port {}",
+        return Err(ServiceError(format!(
+            "system-managed LoopX {} did not become ready on port {}",
             kind.label(),
             kind.port()
-        )))
+        )));
     }
 
-    pub fn stop(&mut self) {
-        for service in self.owned.iter_mut().rev() {
-            service.stop();
+    let mut command = Command::new(&executable);
+    configure_runtime_environment(&mut command);
+    command
+        .args(kind.command_args())
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null());
+    let child = command.group_spawn().map_err(|error| {
+        ServiceError(format!(
+            "could not start LoopX {} with `{executable}`: {error}",
+            kind.label()
+        ))
+    })?;
+    *owned = Some(OwnedService { child });
+
+    let deadline = Instant::now() + STARTUP_TIMEOUT;
+    while Instant::now() < deadline {
+        match probe(kind, expected_runtime_identity.as_ref()) {
+            Probe::Matching => return Ok(()),
+            Probe::NotReady => return Err(status_readiness_error(kind)),
+            Probe::Foreign => {
+                return Err(ServiceError(format!(
+                    "LoopX {} startup reached an unexpected service on port {}",
+                    kind.label(),
+                    kind.port()
+                )));
+            }
+            Probe::Stale => {
+                terminate_verified_listener(kind, &executable, kind.port())?;
+                *healed = true;
+                thread::sleep(Duration::from_millis(200));
+            }
+            Probe::Unavailable | Probe::Unresponsive => thread::sleep(Duration::from_millis(100)),
         }
-        self.owned.clear();
     }
+    Err(ServiceError(format!(
+        "LoopX {} did not become ready on port {}",
+        kind.label(),
+        kind.port()
+    )))
 }
 
 #[cfg(target_os = "macos")]