From 7f420257a6637701fcdcb1c134fb0a66fbb2fcc6 Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 20 Aug 2026 15:11:42 +0800 Subject: [PATCH] fix(api): keep provider degradation out of healthz Co-authored-by: Codex --- backend/crates/gateway-api/src/health.rs | 24 ++++++- backend/crates/gateway-api/tests/health.rs | 71 +++++++++++++++++++ .../crates/gateway-api/tests/openai/mod.rs | 22 +++++- 3 files changed, 113 insertions(+), 4 deletions(-) diff --git a/backend/crates/gateway-api/src/health.rs b/backend/crates/gateway-api/src/health.rs index ace024e0e..4a6c7fd54 100644 --- a/backend/crates/gateway-api/src/health.rs +++ b/backend/crates/gateway-api/src/health.rs @@ -1,4 +1,4 @@ -//! `/healthz` 对 Core、Store 与 Host worker 健康事实的聚合。 +//! `/healthz` 对 Core、Store 与 Host 关键 worker 健康事实的聚合。 use std::sync::Arc; use std::time::Duration; @@ -6,8 +6,10 @@ use std::time::Duration; use axum::{extract::State, http::StatusCode}; use futures::future::join_all; use gateway_core::health::{ - HealthProbe, HealthState, WorkerHealthSnapshot, WorkerHealthSource, WorkerRuntimeState, + HealthProbe, HealthState, WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource, + WorkerRuntimeState, }; +use gateway_core::task::WorkerKind; use crate::ApiState; @@ -32,7 +34,13 @@ impl HealthStatus { } pub(crate) async fn healthy(&self) -> bool { - if !self.workers.snapshot().iter().all(worker_is_healthy) { + if !self + .workers + .snapshot() + .iter() + .filter(|worker| worker_affects_healthz(worker)) + .all(worker_is_healthy) + { return false; } let checks = self.probes.iter().map(|probe| probe.check()); @@ -54,6 +62,16 @@ pub(crate) async fn healthz(State(state): State) -> StatusCode { } } +fn worker_affects_healthz(worker: &WorkerHealthSnapshot) -> bool { + let WorkerHealthKey::Task(id) = &worker.key else { + return false; + }; + !matches!( + id.kind(), + WorkerKind::OAuthRefresh | WorkerKind::QuotaCatalogHealth + ) +} + fn worker_is_healthy(worker: &WorkerHealthSnapshot) -> bool { match worker.state { WorkerRuntimeState::Disabled | WorkerRuntimeState::Standby => true, diff --git a/backend/crates/gateway-api/tests/health.rs b/backend/crates/gateway-api/tests/health.rs index f5cf2b0ab..bf82ed1b8 100644 --- a/backend/crates/gateway-api/tests/health.rs +++ b/backend/crates/gateway-api/tests/health.rs @@ -11,7 +11,9 @@ use gateway_core::{ StartedExecution, }, error::GatewayError, + health::{WorkerHealthKey, WorkerHealthSnapshot, WorkerHealthSource, WorkerRuntimeState}, routing::PublicModelId, + task::{WorkerId, WorkerKind}, }; use tower::ServiceExt; @@ -31,6 +33,75 @@ async fn healthz_should_return_no_content_when_all_inputs_are_healthy() { assert_eq!(response.status(), StatusCode::NO_CONTENT); } +#[tokio::test] +async fn healthz_should_ignore_provider_maintenance_worker_failure() { + let worker_health = StaticWorkerHealth(vec![worker_snapshot( + WorkerKind::QuotaCatalogHealth, + WorkerRuntimeState::BackingOff, + )]); + let response = crate::openai::api_router_with_worker_health( + Arc::new(UnusedExecution), + Arc::new(worker_health), + ) + .await + .oneshot( + Request::builder() + .uri("/healthz") + .body(Body::empty()) + .expect("health request"), + ) + .await + .expect("health response"); + + assert_eq!(response.status(), StatusCode::NO_CONTENT); +} + +#[tokio::test] +async fn healthz_should_reject_critical_worker_failure() { + let worker_health = StaticWorkerHealth(vec![worker_snapshot( + WorkerKind::RuntimeSnapshotReconciliation, + WorkerRuntimeState::BackingOff, + )]); + let response = crate::openai::api_router_with_worker_health( + Arc::new(UnusedExecution), + Arc::new(worker_health), + ) + .await + .oneshot( + Request::builder() + .uri("/healthz") + .body(Body::empty()) + .expect("health request"), + ) + .await + .expect("health response"); + + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); +} + +#[derive(Clone)] +struct StaticWorkerHealth(Vec); + +impl WorkerHealthSource for StaticWorkerHealth { + fn snapshot(&self) -> Vec { + self.0.clone() + } +} + +fn worker_snapshot(kind: WorkerKind, state: WorkerRuntimeState) -> WorkerHealthSnapshot { + let id = WorkerId::try_new(kind, "health-test").expect("worker ID"); + WorkerHealthSnapshot { + key: WorkerHealthKey::Task(id), + state, + consecutive_failures: 1, + completed_cycles: 0, + last_fencing_token: None, + last_success_at: None, + last_failure_at: None, + last_error: Some("test worker failure".to_owned()), + } +} + struct UnusedExecution; impl ExecutionService for UnusedExecution { diff --git a/backend/crates/gateway-api/tests/openai/mod.rs b/backend/crates/gateway-api/tests/openai/mod.rs index e43204e32..b825e71d1 100644 --- a/backend/crates/gateway-api/tests/openai/mod.rs +++ b/backend/crates/gateway-api/tests/openai/mod.rs @@ -44,9 +44,29 @@ pub(super) async fn api_router(execution: Arc) -> axum::Ro api_router_with_origins(execution, Vec::new()).await } +pub(super) async fn api_router_with_worker_health( + execution: Arc, + worker_health: Arc, +) -> axum::Router { + api_router_with_origins_and_worker_health(execution, Vec::new(), worker_health).await +} + pub(super) async fn api_router_with_origins( execution: Arc, cors_allowed_origins: Vec, +) -> axum::Router { + api_router_with_origins_and_worker_health( + execution, + cors_allowed_origins, + Arc::new(EmptyWorkerHealth), + ) + .await +} + +async fn api_router_with_origins_and_worker_health( + execution: Arc, + cors_allowed_origins: Vec, + worker_health: Arc, ) -> axum::Router { let admin = crate::admin::AdminTestFixture::new().await; gateway_api::initialize( @@ -59,7 +79,7 @@ pub(super) async fn api_router_with_origins( execution, admin.services, Vec::new(), - Arc::new(EmptyWorkerHealth), + worker_health, Arc::new(TestLifecycle::default()), ) .expect("API bundle")