fix(service): keep health probes and sidecar fallback consistent

This commit is contained in:
wonfen
2026-07-29 12:39:24 +08:00
parent 7fb5001bea
commit 98d70978e4
5 changed files with 267 additions and 18 deletions
+23 -2
View File
@@ -288,10 +288,11 @@ impl CoreManager {
if !can_allow_sidecar_for_session(&mode, &status) {
anyhow::bail!("Sidecar continuation is not allowed from {mode:?} / {status:?}");
}
proxy_control::stop_guard().await;
Config::suppress_tun_for_session().await;
Config::generate().await?;
SERVICE_MANAGER.allow_sidecar_for_session()?;
// Drop the guard last so an earlier failure cannot leave a running Sidecar unguarded.
proxy_control::stop_guard().await;
let result = async {
self.start_core_inner().await?;
if !matches!(*self.get_running_mode(), RunningMode::Sidecar) {
@@ -301,7 +302,8 @@ impl CoreManager {
}
.await;
if let Err(error) = &result {
SERVICE_MANAGER.mark_unavailable(format!("Sidecar startup failed: {error:#}"));
// Revoke only the failed Sidecar allowance; its startup says nothing about Service health.
SERVICE_MANAGER.withdraw_sidecar_allowance();
if let Err(cleanup_error) = self.rollback_failed_sidecar_transition().await {
return Err(anyhow::anyhow!(
"Sidecar startup failed: {error:#}; failed to clear proxy before rollback: {cleanup_error:#}"
@@ -1256,6 +1258,25 @@ mod tests {
));
}
#[test]
fn a_sidecar_that_failed_to_start_can_be_tried_again() {
// Revoking a failed attempt's allowance keeps retries reachable without changing Service health.
for health in [
ServiceStatus::NotInstalled,
ServiceStatus::NeedsReinstall,
ServiceStatus::Unavailable("boom".into()),
] {
assert!(
can_allow_sidecar_for_session(&RunningMode::NotRunning, &health),
"{health:?} must still admit a second attempt"
);
}
assert!(!can_allow_sidecar_for_session(
&RunningMode::NotRunning,
&ServiceStatus::SidecarAllowed,
));
}
#[test]
fn proxy_restore_requires_the_same_ready_core_generation() {
let sidecar =
+9 -5
View File
@@ -5,7 +5,7 @@
//! this trait rather than reaching for globals, the state machine can be exercised without
//! a running app, an installed Service, or elevation.
use anyhow::Result;
use anyhow::{Context as _, Result};
use super::health::{PendingAction, RunState};
use super::probe::ServiceVersionReply;
@@ -19,7 +19,9 @@ pub trait RunStateEnv: Send + Sync + 'static {
///
/// Fails when the registry itself could not be inspected, which is different from
/// "no installation" and is treated as unavailable rather than not-installed.
fn trusted_install_evidence(&self) -> Result<bool>;
///
/// Async so blocking systemd and Windows SCM checks can run off the runtime workers.
fn trusted_install_evidence(&self) -> impl Future<Output = Result<bool>> + Send;
/// Whether this app process is running elevated.
fn is_elevated(&self) -> bool;
@@ -58,8 +60,10 @@ impl RunStateEnv for RealEnv {
})
}
fn trusted_install_evidence(&self) -> Result<bool> {
crate::core::service::trusted_service_evidence()
async fn trusted_install_evidence(&self) -> Result<bool> {
tokio::task::spawn_blocking(crate::core::service::trusted_service_evidence)
.await
.context("service registration probe did not finish")?
}
fn is_elevated(&self) -> bool {
@@ -259,7 +263,7 @@ mod fake {
reply.map_err(|error| anyhow!(error))
}
fn trusted_install_evidence(&self) -> Result<bool> {
async fn trusted_install_evidence(&self) -> Result<bool> {
self.evidence.clone().map_err(|error| anyhow!(error))
}
+228 -5
View File
@@ -66,18 +66,29 @@ impl std::fmt::Display for ReadyWaitError {
}
}
/// The service-owned state together with the counter that versions it.
/// The service-owned state together with the counters that version it.
///
/// They share one lock so a reader can never see a change without its version bump.
/// `generation` tracks visible changes; `observation` also tracks assertions that invalidate
/// older health probes without changing visible state.
#[derive(Debug, Default)]
struct VersionedService {
service: StoredService,
generation: u64,
observation: u64,
}
impl VersionedService {
/// Record a visible change and invalidate older health probes.
const fn bump(&mut self) {
self.generation = self.generation.wrapping_add(1);
self.observation = self.observation.wrapping_add(1);
}
/// Reserve a revision that invalidates older health probes.
const fn next_observation(&mut self) -> u64 {
self.observation = self.observation.wrapping_add(1);
self.observation
}
}
@@ -194,7 +205,7 @@ impl<E: RunStateEnv> RunStateStore<E> {
/// Platform evidence that cannot be inspected is *unavailable*, not *absent* — assuming
/// "not installed" there would offer the user an install that is already there.
pub async fn detect_service_health(&self) -> ServiceHealth {
let has_marker = match self.env.trusted_install_evidence() {
let has_marker = match self.env.trusted_install_evidence().await {
Ok(exists) => exists,
Err(error) => {
logging!(
@@ -225,10 +236,52 @@ impl<E: RunStateEnv> RunStateStore<E> {
// ─────────────────────────── writes ───────────────────────────
/// Probe and record Service health unless a newer observation arrived first.
///
/// Reserving before I/O prevents slower probes from overwriting later observations, including
/// assertions that leave visible state unchanged. Returns the health that remains recorded.
pub async fn observe_current_health(&self) -> ServiceHealth {
let reservation = self.reserve_health_observation();
let health = self.detect_service_health().await;
self.commit_reserved_observation(health, reservation)
}
/// Reserve the revision for a health probe.
fn reserve_health_observation(&self) -> u64 {
self.service.lock().next_observation()
}
/// Commit `health` only if `reservation` is still latest; otherwise return the newer health.
///
/// The comparison and write share one lock. A successful commit reuses its reserved revision.
fn commit_reserved_observation(&self, health: ServiceHealth, reservation: u64) -> ServiceHealth {
let mut state = self.service.lock();
if state.observation != reservation {
let newer = state.service.health.clone();
drop(state);
logging!(
debug,
Type::Service,
"discarding a stale service reading of {health:?}; {newer:?} arrived while it ran"
);
return newer;
}
if state.service.health == health && state.service.pending.is_none() && !state.service.sidecar_allowed {
return health;
}
state.service.observe(health.clone());
state.generation = state.generation.wrapping_add(1);
drop(state);
self.announce();
health
}
/// Record an observation about the Service, clearing any request it answers.
pub fn observe(&self, health: ServiceHealth) {
let mut state = self.service.lock();
if state.service.health == health && state.service.pending.is_none() && !state.service.sidecar_allowed {
// Reassertion is invisible but must invalidate an older health probe.
state.next_observation();
return;
}
state.service.observe(health);
@@ -245,6 +298,11 @@ impl<E: RunStateEnv> RunStateStore<E> {
pub fn request_action(&self, action: PendingAction) -> bool {
let mut state = self.service.lock();
let displaced = state.service.sidecar_allowed;
if state.service.pending == Some(action) && !displaced {
// Reassertion must prevent an older probe from retiring the request.
state.next_observation();
return displaced;
}
state.service.request(action);
state.bump();
drop(state);
@@ -258,6 +316,8 @@ impl<E: RunStateEnv> RunStateStore<E> {
pub fn accept_sidecar(&self) {
let mut state = self.service.lock();
if state.service.sidecar_allowed && state.service.pending.is_none() {
// Reassertion must prevent an older probe from clearing the allowance.
state.next_observation();
return;
}
state.service.allow_sidecar();
@@ -272,7 +332,13 @@ impl<E: RunStateEnv> RunStateStore<E> {
/// Service. Returns whether the allowance was restored.
pub fn restore_sidecar_allowance(&self) -> bool {
let mut state = self.service.lock();
if state.service.sidecar_allowed || matches!(state.service.health, ServiceHealth::Ready) {
if state.service.sidecar_allowed {
// Reassertion must prevent an older probe from clearing the allowance.
state.next_observation();
return false;
}
if matches!(state.service.health, ServiceHealth::Ready) {
// Cached Ready health does not invalidate an in-flight probe.
return false;
}
state.service.allow_sidecar();
@@ -282,6 +348,21 @@ impl<E: RunStateEnv> RunStateStore<E> {
true
}
/// Revoke an allowance after Sidecar startup fails without changing Service health.
///
/// Returns whether the allowance was revoked.
pub fn withdraw_sidecar_allowance(&self) -> bool {
let mut state = self.service.lock();
if !state.service.sidecar_allowed {
return false;
}
state.service.sidecar_allowed = false;
state.bump();
drop(state);
self.announce();
true
}
/// Accept Sidecar for the rest of this app session.
///
/// Rejected while an operation is in flight, or from a state where the Service is
@@ -324,6 +405,9 @@ impl<E: RunStateEnv> RunStateStore<E> {
{
state.service.request(PendingAction::Install);
state.bump();
} else if state.service.pending == Some(PendingAction::Install) {
// Reassertion must prevent an older probe from clearing the install request.
state.next_observation();
}
drop(state);
self.announce();
@@ -341,13 +425,12 @@ impl<E: RunStateEnv> RunStateStore<E> {
Ok(()) if matches!(action, PendingAction::Uninstall) => self.observe(ServiceHealth::NotInstalled),
Ok(()) => {}
Err(error) => {
let health = self.detect_service_health().await;
let health = self.observe_current_health().await;
logging!(
warn,
Type::Service,
"privileged service action {action:?} did not complete ({error:#}); service is {health:?}"
);
self.observe(health);
}
}
outcome
@@ -504,6 +587,146 @@ mod tests {
}
}
#[tokio::test]
async fn a_reading_nothing_overtook_is_recorded() {
let store = with_env(FakeEnv::service_ready(FakeEnv::new()));
let health = store.observe_current_health().await;
assert_eq!(health, ServiceHealth::Ready);
assert_eq!(store.state().health, ServiceHealth::Ready);
}
#[test]
fn a_reading_overtaken_by_a_changed_observation_is_discarded() {
// A slower earlier probe must not overwrite a newer Ready observation.
let store = with_env(FakeEnv::new());
let reservation = store.reserve_health_observation();
store.observe(ServiceHealth::Ready);
let recorded = store.commit_reserved_observation(ServiceHealth::NotInstalled, reservation);
assert_eq!(recorded, ServiceHealth::Ready, "the newer observation is reported back");
assert_eq!(store.state().health, ServiceHealth::Ready, "and it is what stands");
}
#[test]
fn a_reading_overtaken_by_an_unchanged_observation_is_also_discarded() {
// Even an unchanged "still Ready" observation must invalidate an older probe.
let store = with_env(FakeEnv::new());
store.observe(ServiceHealth::Ready);
let reservation = store.reserve_health_observation();
let generation = store.generation_count();
store.observe(ServiceHealth::Ready);
assert_eq!(store.generation_count(), generation, "nothing visible changed");
let recorded = store.commit_reserved_observation(ServiceHealth::NotInstalled, reservation);
assert_eq!(recorded, ServiceHealth::Ready);
assert_eq!(store.state().health, ServiceHealth::Ready);
}
#[test]
fn a_reading_overtaken_by_an_already_satisfied_sidecar_restore_is_discarded() {
// Reasserting an existing allowance must stop an older probe from clearing it.
let store = with_env(FakeEnv::new());
store.observe(ServiceHealth::VersionMismatch);
store.accept_sidecar();
let reservation = store.reserve_health_observation();
assert!(!store.restore_sidecar_allowance(), "the allowance is already there");
store.commit_reserved_observation(ServiceHealth::VersionMismatch, reservation);
assert!(store.state().sidecar_allowed, "the confirmed allowance survives");
assert!(!store.state().service_needs_attention());
}
#[test]
fn re_requesting_the_same_action_changes_nothing_visible_but_still_orders() {
// The same request leaves `generation` unchanged but must invalidate an older probe.
let store = with_env(FakeEnv::new());
store.observe(ServiceHealth::NotInstalled);
store.require_install_for_session().expect("absent service");
let generation = store.generation_count();
let reservation = store.reserve_health_observation();
assert!(
!store.request_action(PendingAction::Install),
"no Sidecar allowance was displaced"
);
assert_eq!(store.generation_count(), generation, "nothing visible changed");
store.commit_reserved_observation(ServiceHealth::NotInstalled, reservation);
assert_eq!(store.state().pending, Some(PendingAction::Install));
}
#[test]
fn a_reading_overtaken_by_an_already_pending_install_is_discarded() {
// Requiring an existing install request must stop an older probe from clearing it.
let store = with_env(FakeEnv::new());
store.observe(ServiceHealth::NotInstalled);
store.require_install_for_session().expect("absent service");
let reservation = store.reserve_health_observation();
store.require_install_for_session().expect("already requested");
store.commit_reserved_observation(ServiceHealth::NotInstalled, reservation);
assert_eq!(store.state().pending, Some(PendingAction::Install));
assert!(store.state().service_needs_attention());
}
#[test]
fn a_later_reading_supersedes_one_that_is_still_running() {
// The last probe started wins regardless of completion order.
let store = with_env(FakeEnv::new());
store.observe(ServiceHealth::Ready);
let first = store.reserve_health_observation();
let second = store.reserve_health_observation();
assert_eq!(
store.commit_reserved_observation(ServiceHealth::NotInstalled, first),
ServiceHealth::Ready,
"the earlier reading is already superseded"
);
assert_eq!(
store.commit_reserved_observation(ServiceHealth::VersionMismatch, second),
ServiceHealth::VersionMismatch
);
assert_eq!(store.state().health, ServiceHealth::VersionMismatch);
}
#[test]
fn withdrawing_a_sidecar_allowance_says_nothing_about_the_service() {
// A failed Sidecar startup must not change Service health.
let store = with_env(FakeEnv::new());
store.observe(ServiceHealth::NotInstalled);
store.accept_sidecar();
assert!(store.withdraw_sidecar_allowance());
assert!(!store.state().sidecar_allowed);
assert_eq!(store.state().health, ServiceHealth::NotInstalled, "health is untouched");
assert!(
!store.state().service_needs_attention(),
"an absent Service still asks nothing"
);
}
#[test]
fn withdrawing_an_allowance_that_was_never_granted_changes_nothing() {
let store = with_env(FakeEnv::new());
store.observe(ServiceHealth::NotInstalled);
let generation = store.generation_count();
assert!(!store.withdraw_sidecar_allowance());
assert_eq!(store.generation_count(), generation);
}
#[test]
fn a_fresh_store_knows_nothing() {
let store = with_env(FakeEnv::new());
+3 -3
View File
@@ -1299,8 +1299,8 @@ impl ServiceManager {
RUN_STATE.require_install_for_session()
}
pub(crate) fn mark_unavailable(&self, reason: impl Into<String>) {
self.set_status(ServiceStatus::Unavailable(reason.into()));
pub(crate) fn withdraw_sidecar_allowance(&self) -> bool {
RUN_STATE.withdraw_sidecar_allowance()
}
pub async fn detect_startup_status(&self) {
@@ -1308,7 +1308,7 @@ impl ServiceManager {
RUN_STATE.accept_sidecar();
return;
}
RUN_STATE.observe(RUN_STATE.detect_service_health().await);
RUN_STATE.observe_current_health().await;
}
fn set_status(&self, status: ServiceStatus) {
@@ -34,14 +34,15 @@ export const ServiceMigrationDialog = () => {
// snapshot; a failed refresh is treated as needing one, since we cannot tell otherwise.
const needsDecision =
stateRefreshFailed || Boolean(runState?.serviceNeedsAttention)
// Which of the three remedies the dialog offers. A refresh we could not complete is
// treated as an unreachable service, which is what 'repair' is for.
// Treat refresh failures as unreachable; an absent Service still needs install after a failed Sidecar attempt.
const remedy: 'install' | 'repair' | 'reinstall' =
runState?.pendingAction === 'install'
? 'install'
: stateRefreshFailed || runState?.service === 'unavailable'
? 'repair'
: 'reinstall'
: runState?.service === 'notInstalled'
? 'install'
: 'reinstall'
const open = loading || workflowIncomplete || needsDecision
const showCheckingMessage = loading || !needsDecision