diff --git a/crates/openshell-server/src/supervisor_pool.rs b/crates/openshell-server/src/supervisor_pool.rs index cd2dc478bc..d042736c7f 100644 --- a/crates/openshell-server/src/supervisor_pool.rs +++ b/crates/openshell-server/src/supervisor_pool.rs @@ -398,6 +398,8 @@ pub struct SupervisorPool { takeovers_from: tokio::time::Instant, /// When each desired route without an owner record was first seen. unowned_since: Mutex>, + /// When each route's owner term was first seen with its process gone. + owner_gone_since: Mutex>, /// Takeover credentials minted for terms of members this replica serves. minted: Mutex>, /// Launch material loaded for instances owned by members this replica @@ -693,6 +695,7 @@ impl SupervisorPool { unowned_grace: UNOWNED_GRACE, takeovers_from: tokio::time::Instant::now() + TAKEOVER_STARTUP_GRACE, unowned_since: Mutex::new(HashMap::new()), + owner_gone_since: Mutex::new(HashMap::new()), minted: Mutex::new(HashMap::new()), materials: Mutex::new(HashMap::new()), withdrawn_since: Mutex::new(HashMap::new()), @@ -2088,6 +2091,8 @@ impl SupervisorPool { // report describes a process that may be gone, so it renders no // verdict until a new term is granted. let mut unconfirmed = BTreeSet::new(); + // Routes whose granted owner's process is gone, not just leaving. + let mut gone = BTreeSet::new(); for (route, instance) in &desired { if stopping.contains(route) { continue; @@ -2108,6 +2113,9 @@ impl SupervisorPool { if !matches!(failure, Ok(None)) { unconfirmed.insert(route.clone()); } + if matches!(failure, Ok(Some("expired" | "restarted"))) { + gone.insert(route.clone()); + } match failure { Ok(Some(cause)) if takeovers_allowed => { self.take_over(record, cause, &ring, &members, &charges) @@ -2124,6 +2132,7 @@ impl SupervisorPool { match outcome { Ok(Reconciled::Changed(record)) => { unconfirmed.remove(route); + gone.remove(route); recharge( &mut charges, route, @@ -2135,6 +2144,7 @@ impl SupervisorPool { owners.insert(route.clone(), *record); } Ok(Reconciled::Removed) => { + gone.remove(route); let previous = owners.remove(route); recharge( &mut charges, @@ -2157,10 +2167,11 @@ impl SupervisorPool { self.remove_stale_reports(&owners, &report_rows).await; let reports = reports_of_owners(&owners, &report_rows, &unconfirmed); let materials = self.load_materials(&owners, &desired).await; + let expired = self.expired_routes(&owners, &gone); // Only granted terms are served; pending intents stay recorded for // the next pass. if let Some(view) = self - .view(&members, &owners, &desired, materials, reports) + .view(&members, &owners, &desired, materials, reports, expired) .await { self.controller.update_pool(view, listed_at); @@ -2168,6 +2179,33 @@ impl SupervisorPool { Ok(()) } + /// Routes whose owner term's process has been gone longer than the + /// runtime's recovery deadline plus [`MEMBER_TTL`], so the runtime has + /// ended the workload. A takeover starts a new term and clears this. + fn expired_routes( + &self, + owners: &BTreeMap, + gone: &BTreeSet, + ) -> BTreeSet { + let now = tokio::time::Instant::now(); + let limit = Duration::from_secs(u64::from(self.controller.pool_recovery_timeout_secs())) + + MEMBER_TTL; + let mut since = lock(&self.owner_gone_since); + since.retain(|route, (auth_epoch, _)| { + gone.contains(route) + && owners + .get(route) + .is_some_and(|record| record.payload.auth_epoch == *auth_epoch) + }); + gone.iter() + .filter_map(|route| { + let auth_epoch = owners.get(route)?.payload.auth_epoch; + let (_, first_seen) = since.entry(route.clone()).or_insert((auth_epoch, now)); + (now.duration_since(*first_seen) >= limit).then(|| route.clone()) + }) + .collect() + } + /// Remove the owner records of instances that are no longer desired: at /// once when their generation ended, else once their owner had time to be /// told to stop them. @@ -2698,6 +2736,7 @@ impl SupervisorPool { desired: &BTreeMap, assignments: BTreeMap, reports: BTreeMap, + expired: BTreeSet, ) -> Option { let served = self .controller @@ -2719,6 +2758,7 @@ impl SupervisorPool { desired: desired.keys().cloned().collect(), assignments, reports, + expired, }; let mut current_terms = std::collections::HashSet::new(); for record in owners.values() { @@ -4103,6 +4143,47 @@ mod tests { } } + /// An instance whose owner's process is gone reads as absent once the + /// runtime's recovery deadline has passed without a takeover; a leaving + /// owner's instance never does, and a takeover clears the deadline. + #[tokio::test] + async fn an_instance_whose_owner_is_gone_past_the_recovery_deadline_is_absent() { + use openshell_supervisor_control::ObservedInstance; + let fixture = Fixture::new().await; + fixture.member("m1", "p1", false).await; + let route = fixture.fill("m1", "p1", 1).await.remove(0); + let pool = fixture.replica("gw-a"); + pool.reconcile_once().await.unwrap(); + let limit = Duration::from_secs(u64::from(pool.controller.pool_recovery_timeout_secs())) + + MEMBER_TTL; + let backdate = || { + for (_, first_seen) in lock(&pool.owner_gone_since).values_mut() { + *first_seen = first_seen.checked_sub(limit).unwrap(); + } + }; + + // A leaving owner still runs its instance. + fixture.member("m1", "p1", true).await; + pool.reconcile_once().await.unwrap(); + backdate(); + pool.reconcile_once().await.unwrap(); + assert_eq!(pool.controller.observed(&route), ObservedInstance::NoHost); + + // Its process is gone: no verdict until the deadline has passed. + fixture.remove_member("m1").await; + pool.reconcile_once().await.unwrap(); + assert_eq!(pool.controller.observed(&route), ObservedInstance::NoHost); + backdate(); + pool.reconcile_once().await.unwrap(); + assert_eq!(pool.controller.observed(&route), ObservedInstance::Absent); + + // A takeover grants a new term, which is not expired. + fixture.member("m2", "p2", false).await; + pool.reconcile_once().await.unwrap(); + assert_eq!(fixture.owner(&route).await.unwrap().member_id, "m2"); + assert_eq!(pool.controller.observed(&route), ObservedInstance::NoHost); + } + /// One more instance than a listing page, the live one listed last and /// owned by `m1`; returns the routes and a replica that places nothing, /// whose first pass has loaded the view. diff --git a/crates/openshell-supervisor-control/src/controller.rs b/crates/openshell-supervisor-control/src/controller.rs index d05c5ba728..9bcabb6c42 100644 --- a/crates/openshell-supervisor-control/src/controller.rs +++ b/crates/openshell-supervisor-control/src/controller.rs @@ -114,7 +114,7 @@ pub enum ObservedInstance { /// assigned this route. Pending, /// The host reconciled to that snapshot (or the route is not assigned) - /// and does not host the route. + /// and does not host the route, or the pool reports the route expired. Absent, /// The host's latest report of the instance. Reported(SandboxInstanceStatus), @@ -187,6 +187,9 @@ pub struct PoolView { /// gateway replica that serves the owner, for judging instances whose /// owner streams to another replica. pub reports: BTreeMap, + /// Routes whose owner's process has been gone past the runtime's + /// recovery deadline, so the runtime has ended their workloads. + pub expired: BTreeSet, } impl std::fmt::Debug for PoolView { @@ -198,6 +201,7 @@ impl std::fmt::Debug for PoolView { .field("desired", &self.desired) .field("assignments", &self.assignments.keys().collect::>()) .field("reports", &self.reports.keys().collect::>()) + .field("expired", &self.expired) .finish() } } @@ -757,6 +761,9 @@ impl SupervisorHostController { let desired = lock(&self.inner.desired); let pool = lock(&self.inner.pool); let observed = lock(&self.inner.observed); + if pool.view.expired.contains(route) { + return ObservedInstance::Absent; + } let Some(owner) = pool.view.owners.get(route) else { return ObservedInstance::NoHost; }; @@ -1480,6 +1487,7 @@ mod tests { .collect(), owners, reports: BTreeMap::new(), + expired: BTreeSet::new(), } }