Supervise boot services with uniform five-second upgrade grace

cli/src/upgrades.rs

@@ -1,9 +1,15 @@
 //! Grace periods, service restarts, and retirement of replaced versions (PROTOCOL.md section 3.7).
-use crate::{Host, services::Service};
+use crate::Host;
 use serde_json::json;
 use std::{collections::BTreeMap, sync::atomic::Ordering, time::Duration};
 use tokio::sync::watch;
 
+/// Uniform host policy (PROTOCOL 3.7), independent of discovery metadata.
+const UPGRADE_GRACE: Duration = Duration::from_secs(5);
+
+#[cfg(test)]
+mod tests;
+
 pub(crate) struct OldProcess {
     pid: String,
     version: String,
@@ -13,9 +19,7 @@ pub(crate) struct Upgrade {
     plugin: String,
     previous: String,
     current: String,
-    grace: Duration,
     processes: Vec<OldProcess>,
-    services: Vec<(Service, Option<watch::Receiver<bool>>)>,
 }
 impl Host {
     /// Snapshots old admissions while activation holds the admission write guard.
@@ -24,54 +28,48 @@ impl Host {
         plugin: &str,
         previous: String,
         current: &str,
-        grace: f64,
     ) -> Upgrade {
         let mut state = self.0.state.lock().unwrap();
         let processes = state
             .processes
-            .iter()
+            .iter_mut()
             .filter(|(_, process)| {
                 process.info.plugin == plugin
                     && process.info.version != current
                     && process.info.method != "rpc.discover"
             })
-            .map(|(pid, process)| OldProcess {
-                pid: pid.clone(),
-                version: process.info.version.clone(),
-                ended: process.ended.subscribe(),
+            .map(|(pid, process)| {
+                // The service supervisor owns the restart, including an exit on
+                // the superseded notice before the grace timer is installed.
+                process.superseded = !process.stopped;
+                OldProcess {
+                    pid: pid.clone(),
+                    version: process.info.version.clone(),
+                    ended: process.ended.subscribe(),
+                }
             })
             .collect();
-        let mut services = Vec::new();
-        let mut retained = Vec::new();
-        for service in std::mem::take(&mut state.services) {
-            if service
-                .method
-                .split_once('.')
-                .is_some_and(|(name, _)| name == plugin)
+        // A cleanly returned service is dormant until an upgrade. A pending
+        // admission will choose the current version itself. Disabled services
+        // (explicit stop or crash-limit give-up) must never be re-armed here.
+        for service in &mut state.services {
+            if service.enabled
+                && !service.pending
+                && service.owner.is_none()
+                && service.version.as_deref() != Some(current)
+                && service
+                    .method
+                    .split_once('.')
+                    .is_some_and(|(name, _)| name == plugin)
             {
-                let process = state
-                    .processes
-                    .values()
-                    .find(|process| process.id.as_ref() == Some(&service.request));
-                if process.is_some_and(|process| process.info.version == current)
-                    || (process.is_none() && service.starting.load(Ordering::SeqCst))
-                {
-                    retained.push(service);
-                } else {
-                    services.push((service, process.map(|process| process.ended.subscribe())));
-                }
-            } else {
-                retained.push(service);
+                self.queue_service(service);
             }
         }
-        state.services = retained;
         Upgrade {
             plugin: plugin.into(),
             previous,
             current: current.into(),
-            grace: Duration::from_secs_f64(grace),
             processes,
-            services,
         }
     }
     /// Lets old calls finish, stops them after grace, and starts each lifecycle hook once.
@@ -92,7 +90,7 @@ impl Host {
                 "core",
                 &upgrade.plugin,
                 "core",
-                &format!("upgrade grace {} seconds", upgrade.grace.as_secs_f64()),
+                &format!("upgrade grace {} seconds", UPGRADE_GRACE.as_secs()),
             );
             let mut versions: BTreeMap<String, Vec<watch::Receiver<bool>>> = BTreeMap::new();
             versions.entry(upgrade.previous).or_default();
@@ -102,23 +100,11 @@ impl Host {
                     .or_default()
                     .push(process.ended.clone());
                 let host = host.clone();
-                let grace = upgrade.grace;
                 tokio::spawn(async move {
                     let mut ended = process.ended;
                     tokio::select! {
                         _ = ended.wait_for(|ended|*ended) => (),
-                        _ = tokio::time::sleep(grace) => { host.begin_stop(&process.pid); }
-                    }
-                });
-            }
-            for (service, ended) in upgrade.services {
-                let host = host.clone();
-                tokio::spawn(async move {
-                    if let Some(mut ended) = ended {
-                        let _ = ended.wait_for(|ended| *ended).await;
-                    }
-                    if !host.0.shutdown.load(Ordering::SeqCst) {
-                        host.start_service(service.method, service.params).await;
+                        _ = tokio::time::sleep(UPGRADE_GRACE) => { host.begin_retire(&process.pid); }
                     }
                 });
             }