diff --git a/crates/aisix-admin/src/lib.rs b/crates/aisix-admin/src/lib.rs index bb235f02..87162f2c 100644 --- a/crates/aisix-admin/src/lib.rs +++ b/crates/aisix-admin/src/lib.rs @@ -802,6 +802,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -837,6 +838,7 @@ mod tests { last_error_kind: "schema_failed".into(), last_error: "schema validation failed at `/display_name`".into(), seen_at: chrono::Utc::now(), + serving_stale_since: None, }], partially_compatible: vec![aisix_core::config_status::PartialCompatResource { resource_kind: "api_keys".into(), @@ -844,6 +846,7 @@ mod tests { count: 2, }], partially_compatible_rows_by_kind: [("api_keys".to_string(), 2)].into_iter().collect(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -899,6 +902,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); diff --git a/crates/aisix-core/src/config.rs b/crates/aisix-core/src/config.rs index a4a97315..a75ee519 100644 --- a/crates/aisix-core/src/config.rs +++ b/crates/aisix-core/src/config.rs @@ -258,10 +258,16 @@ pub struct ManagedConfig { /// so the proxy can serve traffic from cached config across CP /// outages and full container restarts. /// - /// Empty string disables persistence — useful for ephemeral test - /// runs where you don't want a stale cache to mask a real failure. - #[serde(default = "ManagedConfig::default_snapshot_cache_path")] - pub snapshot_cache_path: String, + /// When the field is omitted, managed mode uses + /// `/var/lib/aisix/config_cache.json` and self-hosted etcd mode + /// leaves persistence off (unchanged defaults). Setting a path + /// enables the cache in either mode — self-hosted etcd deployments + /// gain the same offline resilience by opting in. Empty string + /// disables persistence everywhere — useful for ephemeral test runs + /// where you don't want a stale cache to mask a real failure. A + /// bare `snapshot_cache_path:` (YAML null) is treated as omitted. + #[serde(default)] + pub snapshot_cache_path: Option, /// Heartbeat interval, in seconds. The DP POSTs a heartbeat to /// dp-manager every `heartbeat_interval_secs`; CP surfaces a DP as @@ -296,15 +302,28 @@ impl ManagedConfig { has_pem || has_file } + /// Resolve the snapshot-cache path per the field docs: an explicit + /// path wins in any mode, an explicit empty string disables, and an + /// omitted field means "the default path in managed mode, disabled + /// in self-hosted etcd mode". + pub fn effective_snapshot_cache_path(&self) -> Option<&str> { + match self.snapshot_cache_path.as_deref() { + Some("") => None, + Some(path) => Some(path), + None if self.is_managed() => Some(Self::DEFAULT_SNAPSHOT_CACHE_PATH), + None => None, + } + } + + /// Default on-disk snapshot cache location for managed mode. + pub const DEFAULT_SNAPSHOT_CACHE_PATH: &'static str = "/var/lib/aisix/config_cache.json"; + fn default_mtls_dir() -> String { "/var/lib/aisix/mtls".into() } fn default_dp_id_file() -> String { "/var/lib/aisix/dp_id".into() } - fn default_snapshot_cache_path() -> String { - "/var/lib/aisix/config_cache.json".into() - } const fn default_heartbeat_interval_secs() -> u64 { 15 } @@ -2091,15 +2110,51 @@ managed: assert_eq!(cfg.managed.mtls_dir, "/var/lib/aisix/mtls"); assert_eq!(cfg.managed.dp_id_file, "/var/lib/aisix/dp_id"); // Default snapshot cache path keeps offline-resilience on by - // default; operators opt out by setting the field to "". + // default in managed mode; operators opt out by setting the + // field to "". assert_eq!( - cfg.managed.snapshot_cache_path, - "/var/lib/aisix/config_cache.json", + cfg.managed.effective_snapshot_cache_path(), + Some("/var/lib/aisix/config_cache.json"), ); // CP URL comes from env at runtime — empty here is fine. assert!(cfg.managed.cp_base_url.is_none()); } + /// #871: the snapshot cache resolves per mode — managed defaults on, + /// self-hosted etcd defaults off, an explicit path enables either, + /// an explicit "" disables either. + #[test] + fn snapshot_cache_path_resolution_per_mode() { + let mut managed = ManagedConfig { + enabled: true, + ..Default::default() + }; + assert_eq!( + managed.effective_snapshot_cache_path(), + Some(ManagedConfig::DEFAULT_SNAPSHOT_CACHE_PATH), + ); + managed.snapshot_cache_path = Some(String::new()); + assert_eq!(managed.effective_snapshot_cache_path(), None); + managed.snapshot_cache_path = Some("/tmp/cache.json".into()); + assert_eq!( + managed.effective_snapshot_cache_path(), + Some("/tmp/cache.json"), + ); + + let mut self_hosted = ManagedConfig { + enabled: false, + ..Default::default() + }; + assert_eq!(self_hosted.effective_snapshot_cache_path(), None); + self_hosted.snapshot_cache_path = Some("/tmp/cache.json".into()); + assert_eq!( + self_hosted.effective_snapshot_cache_path(), + Some("/tmp/cache.json"), + ); + self_hosted.snapshot_cache_path = Some(String::new()); + assert_eq!(self_hosted.effective_snapshot_cache_path(), None); + } + #[test] fn rejects_legacy_registration_token_field() { let f = write_yaml( diff --git a/crates/aisix-core/src/config_status.rs b/crates/aisix-core/src/config_status.rs index ebb5b4c7..9f975710 100644 --- a/crates/aisix-core/src/config_status.rs +++ b/crates/aisix-core/src/config_status.rs @@ -30,15 +30,15 @@ //! `key '\0' canonical_json_value '\n'`, concatenated in ascending key //! order. `canonical_json_value` recursively sorts object keys and drops //! insignificant whitespace ([`canonical_json`]); a value that is not JSON -//! is hashed as its raw bytes. `source_hash` covers every entry in the last -//! observed full snapshot; `config_hash` covers only the accepted subset -//! (the entries actually served). When a full snapshot is applied with no -//! rejections the two are equal. A resource rejected on a *live watch event* -//! is surfaced via `rejected[]` and does not enter `source_hash` until the -//! next full resync (the watch delta never carries the bad bytes into the -//! served entry map) — so `source_hash == config_hash` can hold while the -//! state is `degraded`; `rejected[]` is the authoritative partial-rejection -//! signal, not a hash diff. +//! is hashed as its raw bytes. `source_hash` covers every entry the DP has +//! observed from etcd — full snapshots and live watch events alike, +//! including rejected writes. `config_hash` covers the bytes each key +//! actually *serves*: the observed bytes for accepted keys, the pinned +//! last-known-good bytes for keys serving stale (#871, see +//! `serving_stale_since` on `rejected[]`), and nothing for a rejected key +//! with no last good. When everything is accepted the two hashes are +//! equal; a rejection makes them diverge, with `rejected[]` as the +//! authoritative per-resource explanation. //! - **file**: `sha256` over the raw file bytes. On a clean load the applied //! `config_hash` equals `source_hash` (the whole file is applied); on a //! rejected reload the applied hash stays at the last-good file's hash. @@ -142,6 +142,17 @@ pub struct RejectedResource { pub first_seen_at: String, /// RFC3339 UTC timestamp the rejection was most recently observed. pub last_seen_at: String, + /// RFC3339 UTC timestamp since when this resource has been serving its + /// last known good value instead of the rejected bytes (issue #871). + /// Absent when nothing is serving for this resource — the row either + /// never loaded successfully or its retention was dropped. + #[serde(skip_serializing_if = "Option::is_none")] + pub serving_stale_since: Option, + /// Seconds elapsed since `serving_stale_since`, recomputed on every + /// read so the staleness age is reported every cycle. Absent together + /// with `serving_stale_since`. + #[serde(skip_serializing_if = "Option::is_none")] + pub serving_stale_age_seconds: Option, } /// One rejected entry handed to [`ConfigStatus`] by a load path. `identity` @@ -155,6 +166,10 @@ pub struct IncomingRejection { pub last_error_kind: String, pub last_error: String, pub seen_at: DateTime, + /// When set, the resource is still serving its last known good value + /// (pinned before this rejection) and this is the instant that stale + /// serving began (#871). `None` means nothing serves for this row. + pub serving_stale_since: Option>, } /// One partially compatible observation, aggregated per (kind, field): @@ -207,6 +222,11 @@ pub struct LoadObservation { /// Row-based (a resource with two unknown fields counts once), for /// the `aisix_config_partially_compatible_resources` gauge. pub partially_compatible_rows_by_kind: BTreeMap, + /// Served resources per kind whose latest source bytes are rejected + /// and whose last known good value serves instead (#871), for the + /// `aisix_config_stale_served_resources` gauge. The per-resource + /// detail (which row, since when) rides on `rejected[]`. + pub stale_served_rows_by_kind: BTreeMap, /// Whether this load counts as a config reload for /// `aisix_config_reloads_total` (full (re)syncs and file loads do; /// incremental etcd events do not). @@ -230,6 +250,7 @@ struct RetainedRejection { last_error: String, first_seen_at: DateTime, last_seen_at: DateTime, + serving_stale_since: Option>, } #[derive(Debug)] @@ -269,6 +290,10 @@ struct ConfigStatusInner { partially_compatible: Vec, partially_compatible_rows_by_kind: BTreeMap, + // Rows serving their last known good value (#871), per kind. Replaced + // wholesale on every load, like the partially-compatible state. + stale_served_rows_by_kind: BTreeMap, + // Metric counters. reloads_total: u64, reload_failures: BTreeMap<&'static str, u64>, @@ -305,6 +330,7 @@ impl ConfigStatus { rejected: BTreeMap::new(), partially_compatible: Vec::new(), partially_compatible_rows_by_kind: BTreeMap::new(), + stale_served_rows_by_kind: BTreeMap::new(), reloads_total: 0, reload_failures: BTreeMap::new(), })), @@ -353,12 +379,14 @@ impl ConfigStatus { last_error: r.last_error, first_seen_at, last_seen_at: r.seen_at, + serving_stale_since: r.serving_stale_since, }, ); } inner.rejected = merged; inner.partially_compatible = obs.partially_compatible; inner.partially_compatible_rows_by_kind = obs.partially_compatible_rows_by_kind; + inner.stale_served_rows_by_kind = obs.stale_served_rows_by_kind; let clean = inner.rejected.is_empty(); inner.last_reload_successful = clean; @@ -498,6 +526,7 @@ impl ConfigStatusInner { last_error_kind: f.last_error_kind.clone(), last_error: f.last_error.clone(), }); + let now = Utc::now(); let mut rejected: Vec = self .rejected .values() @@ -508,6 +537,12 @@ impl ConfigStatusInner { last_error: r.last_error.clone(), first_seen_at: rfc3339(r.first_seen_at), last_seen_at: rfc3339(r.last_seen_at), + serving_stale_since: r.serving_stale_since.map(rfc3339), + // Recomputed on every read — the "reported every cycle" + // staleness age (#871). Clamped at zero for clock skew. + serving_stale_age_seconds: r + .serving_stale_since + .map(|s| now.signed_duration_since(s).num_seconds().max(0) as u64), }) .collect(); rejected.sort_by(|a, b| { @@ -541,6 +576,7 @@ impl ConfigStatusInner { reload_failures: self.reload_failures.iter().map(|(k, v)| (*k, *v)).collect(), rejected_by_kind, partially_compatible_by_kind: self.partially_compatible_rows_by_kind.clone(), + stale_served_by_kind: self.stale_served_rows_by_kind.clone(), observed_revision: if etcd { self.observed_revision } else { None }, applied_revision: if etcd { self.applied_revision } else { None }, config_hash: self.config_hash.clone(), @@ -618,6 +654,9 @@ pub struct ConfigMetricsView { pub rejected_by_kind: BTreeMap, /// Served resources per kind carrying at least one ignored field. pub partially_compatible_by_kind: BTreeMap, + /// Served resources per kind running on their last known good value + /// because the latest source bytes are rejected (#871). + pub stale_served_by_kind: BTreeMap, pub observed_revision: Option, pub applied_revision: Option, pub config_hash: Option, @@ -728,6 +767,7 @@ mod tests { last_error_kind: error_kind.to_string(), last_error: error.to_string(), seen_at: Utc::now(), + serving_stale_since: None, } } @@ -756,6 +796,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -783,6 +824,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -796,6 +838,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -810,6 +853,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: true, }); @@ -826,6 +870,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -849,6 +894,7 @@ mod tests { )], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -871,6 +917,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -888,6 +935,7 @@ mod tests { )], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: true, }); @@ -910,6 +958,7 @@ mod tests { )], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -926,6 +975,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: false, wholly_rejected: false, }; @@ -953,6 +1003,7 @@ mod tests { )], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -965,6 +1016,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -975,6 +1027,61 @@ mod tests { assert!(v.rejected.is_empty()); } + /// #871: a rejection whose key serves its last known good value + /// reports the stale-since instant and a freshly computed age on + /// every read; one with nothing serving omits both fields. + #[test] + fn stale_serving_rejections_report_since_and_age() { + let cs = ConfigStatus::new(SourceKind::Etcd); + let mut stale = incoming( + "/aisix/models/stale", + "models", + "stale", + "schema_failed", + "boom", + ); + stale.serving_stale_since = Some(Utc::now() - chrono::Duration::seconds(90)); + let dead = incoming( + "/aisix/models/dead", + "models", + "dead", + "schema_failed", + "boom", + ); + cs.record_load(LoadObservation { + source_hash: "s".into(), + observed_revision: Some(1), + applied: Some(applied("a", &[("models", 1)])), + rejected: vec![stale, dead], + partially_compatible: Vec::new(), + partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: [("models".to_string(), 1)].into_iter().collect(), + is_reload: true, + wholly_rejected: false, + }); + + let json = serde_json::to_value(cs.view()).unwrap(); + let rejected = json["rejected"].as_array().unwrap(); + let dead_row = rejected + .iter() + .find(|r| r["resource_id"] == "dead") + .unwrap(); + assert!( + dead_row.get("serving_stale_since").is_none(), + "a rejection with nothing serving must omit the stale fields", + ); + let stale_row = rejected + .iter() + .find(|r| r["resource_id"] == "stale") + .unwrap(); + assert!(stale_row["serving_stale_since"].is_string()); + let age = stale_row["serving_stale_age_seconds"].as_u64().unwrap(); + assert!((90..=95).contains(&age), "age ≈ 90s, got {age}"); + + // The per-kind stale row counts flow through to the metrics view. + assert_eq!(cs.metrics().stale_served_by_kind.get("models"), Some(&1)); + } + #[test] fn first_seen_is_preserved_across_reloads() { let cs = ConfigStatus::new(SourceKind::Etcd); @@ -993,6 +1100,7 @@ mod tests { rejected: vec![r], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -1013,6 +1121,7 @@ mod tests { rejected: vec![r2], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -1038,6 +1147,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -1058,6 +1168,7 @@ mod tests { rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); @@ -1087,6 +1198,7 @@ mod tests { ], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload: true, wholly_rejected: false, }); diff --git a/crates/aisix-core/src/filesource/status.rs b/crates/aisix-core/src/filesource/status.rs index 368e4452..925699e9 100644 --- a/crates/aisix-core/src/filesource/status.rs +++ b/crates/aisix-core/src/filesource/status.rs @@ -66,6 +66,7 @@ pub fn load_resources_file_tracked( rejected: vec![], partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload, wholly_rejected: false, }); @@ -85,6 +86,7 @@ pub fn load_resources_file_tracked( rejected, partially_compatible: Vec::new(), partially_compatible_rows_by_kind: Default::default(), + stale_served_rows_by_kind: Default::default(), is_reload, // The whole file was rejected; last-good retained. wholly_rejected: true, @@ -137,6 +139,10 @@ fn map_load_error(e: &LoadError, seen_at: chrono::DateTime) -> IncomingReje last_error_kind: classify(&e.message).to_string(), last_error: e.message.clone(), seen_at, + // The file source is all-or-nothing: a failed reload keeps the + // previous snapshot wholesale (reported via `wholly_rejected`), + // so per-row last-known-good retention does not apply. + serving_stale_since: None, } } diff --git a/crates/aisix-etcd/src/loader.rs b/crates/aisix-etcd/src/loader.rs index c1e00503..adb6c744 100644 --- a/crates/aisix-etcd/src/loader.rs +++ b/crates/aisix-etcd/src/loader.rs @@ -88,6 +88,12 @@ pub struct RejectedEntry { pub kind: RejectionKind, pub error: String, pub timestamp_unix_secs: u64, + /// Unix seconds since when this key has been serving its last known + /// good value instead of the rejected bytes (#871). Always `None` as + /// produced by the loader — the supervisor joins its retained + /// stale-serving state in on read, so the heartbeat reports the + /// staleness age next to the rejection. + pub stale_serving_since_unix_secs: Option, } impl RejectedEntry { @@ -101,6 +107,7 @@ impl RejectedEntry { kind, error: error.into(), timestamp_unix_secs: now, + stale_serving_since_unix_secs: None, } } } diff --git a/crates/aisix-etcd/src/snapshot_cache.rs b/crates/aisix-etcd/src/snapshot_cache.rs index 26d31cad..aa335f6f 100644 --- a/crates/aisix-etcd/src/snapshot_cache.rs +++ b/crates/aisix-etcd/src/snapshot_cache.rs @@ -34,10 +34,13 @@ use tokio::io::AsyncWriteExt; use tokio::sync::Mutex; use crate::provider::RawEntry; +use crate::supervisor::StaleServing; /// File-format version. Bumped whenever the wire shape of [`CachedFile`] /// changes incompatibly so old DPs ignore future caches instead of -/// crashing on a stale-format upgrade. +/// crashing on a stale-format upgrade. The `stale` section added for +/// #871 is additive (defaulted on read, ignored by older DPs), so it +/// stays at 1. const FORMAT_VERSION: u32 = 1; /// Owned, sync-or-async-safe snapshot cache. Cheap to clone — internally @@ -61,6 +64,12 @@ struct CachedFile { version: u32, revision: i64, entries: Vec, + /// Last-known-good values pinned for keys whose current bytes (in + /// `entries`) are rejected (#871). Restored before the boot resync so + /// stale-serving rows survive a restart. Defaulted so pre-#871 cache + /// files keep loading. + #[serde(default)] + stale: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -72,6 +81,27 @@ struct CachedEntry { revision: i64, } +#[derive(Debug, Clone, Serialize, Deserialize)] +struct CachedStaleEntry { + key: String, + /// Base64 of the pinned last-known-good value bytes. + value_b64: String, + /// Revision the pinned value was accepted at. + revision: i64, + /// Unix seconds when stale serving began — persisted so the reported + /// staleness age stays continuous across restarts. + since_unix_secs: u64, +} + +/// The parsed contents of a cache file: the raw entry set, the revision +/// it reflects, and the pinned last-known-good values (#871). +#[derive(Debug, Clone)] +pub struct CachedSnapshot { + pub entries: Vec, + pub revision: i64, + pub stale: Vec, +} + impl SnapshotCache { /// Construct a cache backed by `path`. The file is created on the /// first successful [`Self::store`]; missing path on [`Self::load`] @@ -102,7 +132,7 @@ impl SnapshotCache { /// recognised [`FORMAT_VERSION`], and every entry's value decodes. /// Anything else is logged and treated as cache-miss so a corrupt /// file can never wedge the DP. - pub fn load(&self) -> Option<(Vec, i64)> { + pub fn load(&self) -> Option { let path = self.inner.path.as_ref()?; let bytes = match std::fs::read(path) { Ok(b) => b, @@ -138,20 +168,38 @@ impl SnapshotCache { }) }) .collect(); - match entries { - Ok(entries) => Some((entries, cached.revision)), - Err(e) => { + let stale: Result, _> = cached + .stale + .into_iter() + .map(|s| { + B64.decode(&s.value_b64).map(|value| StaleServing { + entry: RawEntry { + key: s.key, + value, + revision: s.revision, + }, + since_unix_secs: s.since_unix_secs, + }) + }) + .collect(); + match (entries, stale) { + (Ok(entries), Ok(stale)) => Some(CachedSnapshot { + entries, + revision: cached.revision, + stale, + }), + (Err(e), _) | (_, Err(e)) => { tracing::warn!(error = %e, "snapshot cache entry decode failed; ignoring"); None } } } - /// Write the given entries + revision atomically. Errors are - /// logged-and-swallowed because losing the cache is not worth - /// blowing up an otherwise-healthy DP — at worst the next restart - /// rebuilds from etcd. - pub async fn store(&self, entries: &[RawEntry], revision: i64) { + /// Write the given entries + revision + pinned last-known-good values + /// atomically. Errors are logged-and-swallowed because losing the + /// cache is not worth blowing up an otherwise-healthy DP — at worst + /// the next restart rebuilds from etcd. + pub async fn store(&self, entries: &[RawEntry], revision: i64, stale: &[StaleServing]) { let Some(path) = self.inner.path.clone() else { return; }; @@ -166,6 +214,15 @@ impl SnapshotCache { revision: e.revision, }) .collect(), + stale: stale + .iter() + .map(|s| CachedStaleEntry { + key: s.entry.key.clone(), + value_b64: B64.encode(&s.entry.value), + revision: s.entry.revision, + since_unix_secs: s.since_unix_secs, + }) + .collect(), }; let bytes = match serde_json::to_vec(&cached) { Ok(b) => b, @@ -219,11 +276,47 @@ mod tests { entry("/aisix/models/m-1", br#"{"name":"m1"}"#, 7), entry("/aisix/api_keys/k-1", b"\xff\x00\x01raw", 8), ]; - cache.store(&entries, 42).await; + cache.store(&entries, 42, &[]).await; + + let cached = cache.load().expect("cache file exists"); + assert_eq!(cached.revision, 42); + assert_eq!(cached.entries, entries); + assert!(cached.stale.is_empty()); + } + + #[tokio::test] + async fn round_trips_stale_entries() { + let dir = tempdir().unwrap(); + let cache = SnapshotCache::new(dir.path().join("snap.json")); + let entries = vec![entry("/aisix/models/m-1", br#"{"bad":true}"#, 9)]; + let stale = vec![StaleServing { + entry: entry("/aisix/models/m-1", br#"{"name":"last-good"}"#, 7), + since_unix_secs: 1_770_000_000, + }]; + cache.store(&entries, 9, &stale).await; - let (loaded, rev) = cache.load().expect("cache file exists"); - assert_eq!(rev, 42); - assert_eq!(loaded, entries); + let cached = cache.load().expect("cache file exists"); + assert_eq!(cached.stale, stale); + } + + /// A pre-#871 cache file (no `stale` section) must keep loading. + #[tokio::test] + async fn legacy_file_without_stale_section_loads() { + let dir = tempdir().unwrap(); + let path = dir.path().join("snap.json"); + let legacy = serde_json::json!({ + "version": 1, + "revision": 5, + "entries": [ + {"key": "/aisix/models/m-1", "value_b64": B64.encode(br#"{"a":1}"#), "revision": 5} + ], + }); + tokio::fs::write(&path, serde_json::to_vec(&legacy).unwrap()) + .await + .unwrap(); + let cached = SnapshotCache::new(&path).load().expect("legacy file loads"); + assert_eq!(cached.entries.len(), 1); + assert!(cached.stale.is_empty()); } #[tokio::test] @@ -236,7 +329,7 @@ mod tests { #[tokio::test] async fn disabled_cache_is_a_noop() { let cache = SnapshotCache::disabled(); - cache.store(&[entry("/a", b"x", 1)], 1).await; + cache.store(&[entry("/a", b"x", 1)], 1, &[]).await; assert!(cache.load().is_none()); } @@ -270,10 +363,10 @@ mod tests { let dir = tempdir().unwrap(); let path = dir.path().join("snap.json"); let cache = SnapshotCache::new(&path); - cache.store(&[entry("/a", b"v1", 1)], 1).await; - cache.store(&[entry("/a", b"v2", 2)], 2).await; - let (loaded, rev) = cache.load().unwrap(); - assert_eq!(rev, 2); - assert_eq!(loaded[0].value, b"v2".to_vec()); + cache.store(&[entry("/a", b"v1", 1)], 1, &[]).await; + cache.store(&[entry("/a", b"v2", 2)], 2, &[]).await; + let cached = cache.load().unwrap(); + assert_eq!(cached.revision, 2); + assert_eq!(cached.entries[0].value, b"v2".to_vec()); } } diff --git a/crates/aisix-etcd/src/supervisor.rs b/crates/aisix-etcd/src/supervisor.rs index 7484a0c6..11e12ff7 100644 --- a/crates/aisix-etcd/src/supervisor.rs +++ b/crates/aisix-etcd/src/supervisor.rs @@ -138,6 +138,25 @@ const MAX_RETAINED_REJECTIONS: usize = 256; /// the aggregated report, with a WARN so the truncation is never silent. const MAX_RETAINED_PARTIAL_ROWS: usize = 1024; +/// One key whose latest etcd bytes are rejected while its last +/// successfully loaded value keeps serving (#871, xDS-NACK style). +/// `entry` pins the last-known-good raw document with the revision it +/// was accepted at; `since_unix_secs` is the instant stale serving began +/// (the first rejected replacement observed for the key), reported as +/// the staleness age and persisted in the snapshot cache so the age +/// stays continuous across restarts. +/// +/// Deliberately uncapped, unlike the rejection and partial-compat +/// buffers: dropping an entry here would take a served resource offline, +/// not truncate a report. The map is bounded by the number of rows that +/// ever loaded successfully — a subset of the served snapshot, which is +/// itself uncapped. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StaleServing { + pub entry: RawEntry, + pub since_unix_secs: u64, +} + /// One supervisor instance. Consumers call [`Supervisor::run`] once and /// drop the returned handle on shutdown. pub struct Supervisor { @@ -183,6 +202,19 @@ pub struct Supervisor { /// [`MAX_RETAINED_PARTIAL_ROWS`]. partial_compat: Mutex>, + /// Last-known-good state, keyed by etcd key: exactly the keys whose + /// latest etcd bytes are rejected while a previously accepted value + /// keeps serving (#871). A rejected put pins the serving bytes here; + /// a successful put or a delete removes the key; a resync drops keys + /// that now load or left etcd and re-injects the rest into the fresh + /// snapshot. Persisted via [`SnapshotCache`] so retention survives + /// restarts. Independent of the capped `rejections` buffer — buffer + /// overflow must never take a served resource offline. + /// + /// Locking: never held together with another supervisor lock; every + /// user snapshots or mutates it in its own scope. + stale_serving: Mutex>, + // JoinHandles for in-flight `flush_cache` writes. Tests use // [`Self::await_pending_cache_writes`] to deterministically wait // for these without relying on a wall-clock sleep, which proved @@ -217,6 +249,7 @@ impl Supervisor

{ config_status: ConfigStatus::new(SourceKind::Etcd), rejections: Mutex::new(Vec::new()), partial_compat: Mutex::new(HashMap::new()), + stale_serving: Mutex::new(HashMap::new()), pending_writes: Mutex::new(Vec::new()), } } @@ -243,33 +276,60 @@ impl Supervisor

{ /// `aisix_config_reloads_total` — set only on full (re)syncs, not on /// incremental watch events. fn sync_config_status(&self, is_reload: bool) { + // Snapshot the stale-serving state first, in its own lock scope + // (see the `stale_serving` field docs for the locking rule). + let stale: HashMap = self.stale_serving.lock().unwrap().clone(); let source_hash; let config_hash; let rejected: Vec; { let state = self.state.lock().unwrap(); let rejections = self.rejections.lock().unwrap(); - // `state` is the raw entry map the DP holds. A full resync inserts - // every entry (including rejected ones), so on that path source_hash - // covers the whole observed snapshot. A live watch Put that is - // rejected never enters `state`, so it is reported only via - // `rejected[]` and folds into source_hash at the next resync — see - // the `config_status` module docs. + // `state` is the raw entry map the DP holds — every observed + // etcd write lands here, including rejected ones (a resync + // inserts them wholesale; a rejected live Put mirrors its + // bytes in too, #871), so source_hash always covers the + // observed etcd state. source_hash = hash_entries(state.values().map(|e| (e.key.as_str(), e.value.as_slice()))); let rejected_keys: HashSet<&str> = rejections.iter().map(|r| r.key.as_str()).collect(); + // config_hash covers the bytes each key ACTUALLY serves: the + // observed etcd bytes for accepted keys, the pinned last-known- + // good bytes for stale-serving keys (#871), and nothing for a + // rejected key with no last good (it doesn't serve). The stale + // map — not the capped rejection buffer — decides which keys + // substitute, so buffer overflow can never flip a served key's + // hash contribution to the rejected bytes. config_hash = hash_entries( state .values() - .filter(|e| !rejected_keys.contains(e.key.as_str())) - .map(|e| (e.key.as_str(), e.value.as_slice())), + .filter(|e| { + !rejected_keys.contains(e.key.as_str()) && !stale.contains_key(&e.key) + }) + .map(|e| (e.key.as_str(), e.value.as_slice())) + .chain( + stale + .values() + .map(|s| (s.entry.key.as_str(), s.entry.value.as_slice())), + ), ); - rejected = rejections.iter().map(|r| self.map_rejection(r)).collect(); + rejected = rejections + .iter() + .map(|r| self.map_rejection(r, &stale)) + .collect(); } let revision = *self.revision.lock().unwrap(); let resource_counts = resource_counts(&self.handle.load()); let (partially_compatible, partially_compatible_rows_by_kind) = self.partial_compat_observation(); + let mut stale_served_rows_by_kind: BTreeMap = BTreeMap::new(); + for key_str in stale.keys() { + if let Ok(parsed) = key::parse(&self.prefix, key_str) { + *stale_served_rows_by_kind + .entry(parsed.kind.to_string()) + .or_insert(0) += 1; + } + } self.config_status.record_load(LoadObservation { source_hash, @@ -282,6 +342,7 @@ impl Supervisor

{ rejected, partially_compatible, partially_compatible_rows_by_kind, + stale_served_rows_by_kind, is_reload, // etcd always publishes the accepted subset (even an empty one); // it never retains a previous snapshot wholesale, so a wholly- @@ -293,8 +354,13 @@ impl Supervisor

{ /// Map a loader [`RejectedEntry`] to the source-agnostic wire shape. The /// key is split into `/` via [`key::parse`]; an unparseable key /// (the `bad_key` path) reports empty kind/id, mirroring the control - /// plane's rejected-resources surface. - fn map_rejection(&self, r: &RejectedEntry) -> IncomingRejection { + /// plane's rejected-resources surface. `stale` joins in the instant the + /// key began serving its last known good value, if it is (#871). + fn map_rejection( + &self, + r: &RejectedEntry, + stale: &HashMap, + ) -> IncomingRejection { let (kind, id) = match key::parse(&self.prefix, &r.key) { Ok(parsed) => (parsed.kind.to_string(), parsed.id.to_string()), Err(_) => (String::new(), String::new()), @@ -307,15 +373,30 @@ impl Supervisor

{ last_error: r.error.clone(), seen_at: DateTime::from_timestamp(r.timestamp_unix_secs as i64, 0) .unwrap_or_else(Utc::now), + serving_stale_since: stale + .get(&r.key) + .and_then(|s| DateTime::from_timestamp(s.since_unix_secs as i64, 0)), } } - /// Snapshot of the most recent loader rejections (capped). Used by - /// the heartbeat path to forward "DP rejected these resources" to + /// Snapshot of the most recent loader rejections (capped), with the + /// stale-serving instant joined in per key (#871). Used by the + /// heartbeat path to forward "DP rejected these resources" to /// cp-api. Returns a clone so the caller doesn't hold the lock /// across the heartbeat HTTP call. pub fn recent_rejections(&self) -> Vec { - self.rejections.lock().unwrap().clone() + let stale: HashMap = { + let guard = self.stale_serving.lock().unwrap(); + guard + .iter() + .map(|(k, s)| (k.clone(), s.since_unix_secs)) + .collect() + }; + let mut out = self.rejections.lock().unwrap().clone(); + for r in &mut out { + r.stale_serving_since_unix_secs = stale.get(&r.key).copied(); + } + out } /// Replace the retained rejection buffer wholesale. Called by the @@ -449,22 +530,33 @@ impl Supervisor

{ /// No-op when the cache is disabled or the file is missing / /// unparseable. pub fn restore_from_cache(&self) { - let Some((entries, revision)) = self.cache.load() else { + let Some(cached) = self.cache.load() else { return; }; - let stats = self.apply_resync(&entries); + // Seed the stale-serving state BEFORE the resync so keys whose + // cached bytes are rejected recover their pinned last-known-good + // values (#871) — apply_resync then re-validates each seed and + // drops any whose key now loads cleanly or left the entry set. + { + let mut stale = self.stale_serving.lock().unwrap(); + stale.clear(); + for s in cached.stale { + stale.insert(s.entry.key.clone(), s); + } + } + let stats = self.apply_resync(&cached.entries); // Track the last cached revision so the first live cycle's // resync reflects the right "from where" in logs. We don't // try to use it as the watch start revision — the etcd server // may have compacted past it; load_all + watch from latest is // always safer. - *self.revision.lock().unwrap() = revision; + *self.revision.lock().unwrap() = cached.revision; // Reflect the cached revision on the status view (apply_resync above // synced with the entry-max revision). self.sync_config_status(false); tracing::info!( accepted = stats.accepted, - revision, + revision = cached.revision, "snapshot restored from on-disk cache (offline-resilient boot)", ); } @@ -515,12 +607,72 @@ impl Supervisor

{ self.sync_config_status(false); } + /// Pin the currently served bytes for `key` as its last known good + /// (#871). Called when a watch put for the key is rejected. No-op + /// when the key is already stale-tracked (the original pin and its + /// `since` stand) or when nothing serves for the key (it never + /// loaded successfully — there is no good value to pin). + /// + /// The serving bytes are read from `state[key]`: the invariant is + /// that a key present in the served snapshot and NOT stale-tracked + /// has its served bytes in `state` (a rejected put pins here BEFORE + /// mirroring the rejected bytes into `state`, and a resync that + /// rejects a key either stale-tracks it or drops it from the + /// snapshot). + fn capture_last_good(&self, key_str: &str) { + if self.stale_serving.lock().unwrap().contains_key(key_str) { + return; + } + let Ok(parsed) = key::parse(&self.prefix, key_str) else { + return; + }; + if !snapshot_has(&self.handle.load(), parsed.kind, parsed.id) { + return; + } + let Some(good) = self.state.lock().unwrap().get(key_str).cloned() else { + return; + }; + // entry().or_insert_with keeps the original pin (and its `since`) + // if a concurrent caller won the race after the check above. + self.stale_serving + .lock() + .unwrap() + .entry(key_str.to_string()) + .or_insert_with(|| StaleServing { + entry: good, + since_unix_secs: now_unix_secs(), + }); + } + /// Apply a single Put event on top of the current snapshot. /// Returns `true` if the apply succeeded (schema + parse passed). pub fn apply_put(&self, entry: &RawEntry) -> bool { // Build a tiny snapshot out of just the new entry, then merge. let (tiny, mut stats) = loader::build_snapshot(&self.prefix, std::slice::from_ref(entry)); if stats.accepted == 0 { + // The previous good value keeps serving. Pin it now (#871): + // the next resync rebuilds from the rejected etcd bytes and + // needs the pinned bytes to keep this row alive. Must run + // BEFORE the state-map update below, which overwrites the + // serving bytes with the rejected ones. + self.capture_last_good(&entry.key); + // Mirror the rejected bytes into the observed-state map and + // the cache like any other observed etcd write: source_hash + // reflects the observed etcd state immediately, and a + // restart inside this window restores the same + // rejected-bytes + pinned-value shape a post-resync restart + // would — keeping the staleness clock continuous instead of + // resetting it at the next boot. + { + let mut state = self.state.lock().unwrap(); + state.insert(entry.key.clone(), entry.clone()); + } + { + let mut rev = self.revision.lock().unwrap(); + if entry.revision > *rev { + *rev = entry.revision; + } + } // Note: a previously retained partially-compatible entry for // this key is deliberately kept — the row's last-good value // (loaded with those fields ignored) is still what serves. @@ -534,6 +686,7 @@ impl Supervisor

{ // A rejected watch event still changes the reported state // (rejected[] gains this entry; last_reload flips unsuccessful). self.sync_config_status(false); + self.flush_cache(); return false; } @@ -547,51 +700,17 @@ impl Supervisor

{ // because the operation is "merge tiny into current". self.handle.rcu(|current| { let new = clone_snapshot(current); - - // Move any entries from `tiny` into `new`. Must cover every - // ResourceTable on AisixSnapshot — a missing kind here - // means a watch event silently drops on the floor and the - // snapshot never sees the new entry, even though the loader - // and the proxy both know about it. - for e in tiny.models.entries() { - new.models.insert(clone_entry(&e)); - } - for e in tiny.apikeys.entries() { - new.apikeys.insert(clone_entry(&e)); - } - for e in tiny.provider_keys.entries() { - new.provider_keys.insert(clone_entry(&e)); - } - for e in tiny.guardrails.entries() { - new.guardrails.insert(clone_entry(&e)); - } - for e in tiny.guardrail_attachments.entries() { - new.guardrail_attachments.insert(clone_entry(&e)); - } - for e in tiny.cache_policies.entries() { - new.cache_policies.insert(clone_entry(&e)); - } - for e in tiny.observability_exporters.entries() { - new.observability_exporters.insert(clone_entry(&e)); - } - for e in tiny.rate_limit_policies.entries() { - new.rate_limit_policies.insert(clone_entry(&e)); - } - for e in tiny.mcp_servers.entries() { - new.mcp_servers.insert(clone_entry(&e)); - } - for e in tiny.mcp_policies.entries() { - new.mcp_policies.insert(clone_entry(&e)); - } - for e in tiny.a2a_agents.entries() { - new.a2a_agents.insert(clone_entry(&e)); - } - for e in tiny.oidc_providers.entries() { - new.oidc_providers.insert(clone_entry(&e)); - } + // Move any entries from `tiny` into `new`. `merge_snapshot` + // must cover every ResourceTable on AisixSnapshot — a + // missing kind there means a watch event silently drops on + // the floor and the snapshot never sees the new entry, even + // though the loader and the proxy both know about it. + merge_snapshot(&new, &tiny); new }); self.remove_rejection_for_key(&entry.key); + // The key's latest bytes load again — retention ends (#871). + self.stale_serving.lock().unwrap().remove(&entry.key); // Refresh this key's partially-compatible signal: replaced when // the new value still carries unknown fields, cleared when it now // matches the schema exactly. @@ -641,34 +760,28 @@ impl Supervisor

{ // delete that wins the race observes the same key already // gone, so this caller returns false (nothing left to remove). let snap = self.handle.load(); - let present = match parsed.kind { - "models" => snap.models.get_by_id(parsed.id).is_some(), - "api_keys" => snap.apikeys.get_by_id(parsed.id).is_some(), - "provider_keys" => snap.provider_keys.get_by_id(parsed.id).is_some(), - "guardrails" => snap.guardrails.get_by_id(parsed.id).is_some(), - "guardrail_attachments" => snap.guardrail_attachments.get_by_id(parsed.id).is_some(), - "cache_policies" => snap.cache_policies.get_by_id(parsed.id).is_some(), - "observability_exporters" => { - snap.observability_exporters.get_by_id(parsed.id).is_some() - } - "rate_limit_policies" => snap.rate_limit_policies.get_by_id(parsed.id).is_some(), - "mcp_servers" => snap.mcp_servers.get_by_id(parsed.id).is_some(), - "mcp_policies" => snap.mcp_policies.get_by_id(parsed.id).is_some(), - "a2a_agents" => snap.a2a_agents.get_by_id(parsed.id).is_some(), - "oidc_providers" => snap.oidc_providers.get_by_id(parsed.id).is_some(), - _ => false, - }; + let present = snapshot_has(&snap, parsed.kind, parsed.id); let removed_rejection = self.remove_rejection_for_key(key_str); // A deleted key no longer serves, so its partially-compatible - // signal (if any) goes with it. + // signal (if any) goes with it — and so does its last-known-good + // retention (#871): the pin must never outlive the etcd key. self.update_partial_row(key_str, None); + self.stale_serving.lock().unwrap().remove(key_str); + // The observed-state map drops the key on BOTH branches below. + // A key can be absent from the snapshot yet present in `state`: + // a rejected put mirrors its bytes there even when the row never + // served. Leaving those bytes behind would keep the deleted key + // in source_hash until the next resync and persist the deleted + // document in the cache file. + let removed_state = self.state.lock().unwrap().remove(key_str).is_some(); drop(snap); if !present { - if removed_rejection { + if removed_rejection || removed_state { let cur_rev = *self.revision.lock().unwrap(); self.status.record_apply(cur_rev); // Clearing a rejected key changes the reported state. self.sync_config_status(false); + self.flush_cache(); } return removed_rejection; } @@ -722,7 +835,6 @@ impl Supervisor

{ } new }); - self.state.lock().unwrap().remove(key_str); // Stamp /admin/v1/health freshness on a successful delete. We // don't have a per-event revision on the wire delete // (the etcd watch revision is held at the cycle level); @@ -736,8 +848,96 @@ impl Supervisor

{ } /// Replace the current snapshot with a freshly loaded set (resync). + /// + /// Rejected keys don't simply vanish (#871): a key whose latest bytes + /// are rejected but whose previous good value was serving keeps + /// serving that value — the pre-existing "cliff" where a routine + /// resync/restart silently took a resource offline days after the + /// write that broke it. Retention ends when the key loads cleanly + /// again or leaves etcd. pub fn apply_resync(&self, entries: &[RawEntry]) -> BuildStats { - let (snap, stats) = loader::build_snapshot(&self.prefix, entries); + let (snap, mut stats) = loader::build_snapshot(&self.prefix, entries); + + // Reconcile the last-known-good state against this build, then + // inject the retained values into the fresh snapshot. + let rejected_keys: HashSet<&str> = + stats.rejections.iter().map(|r| r.key.as_str()).collect(); + let entry_keys: HashSet<&str> = entries.iter().map(|e| e.key.as_str()).collect(); + // Serving bytes for newly rejected keys come from the PRE-resync + // state map (see `capture_last_good` for the invariant). Collect + // them before `state` is replaced below. + let prev_state: HashMap = { + let state = self.state.lock().unwrap(); + stats + .rejections + .iter() + .filter_map(|r| state.get(&r.key).map(|e| (r.key.clone(), e.clone()))) + .collect() + }; + let prev_snap = self.handle.load(); + let injected: Vec = { + let mut stale = self.stale_serving.lock().unwrap(); + // Retention ends for keys that now load cleanly or left etcd + // entirely — the delete-side guarantee that a pinned value + // never outlives its key. + stale.retain(|k, _| { + entry_keys.contains(k.as_str()) && rejected_keys.contains(k.as_str()) + }); + // Newly rejected keys that were serving up to this resync: + // pin their serving bytes now. + for r in &stats.rejections { + if stale.contains_key(&r.key) { + continue; + } + let Ok(parsed) = key::parse(&self.prefix, &r.key) else { + continue; + }; + if !snapshot_has(&prev_snap, parsed.kind, parsed.id) { + continue; + } + if let Some(good) = prev_state.get(&r.key) { + stale.insert( + r.key.clone(), + StaleServing { + entry: good.clone(), + since_unix_secs: now_unix_secs(), + }, + ); + } + } + stale.values().map(|s| s.entry.clone()).collect() + }; + drop(prev_snap); + + // Re-build each pinned value from its bytes so every derived + // signal (typed value, YELLOW ignored-field paths) stays + // consistent with what actually serves. A pinned value this + // build can no longer parse (e.g. after a DP downgrade) drops + // its retention with an ERROR — same contract as any RED row. + if !injected.is_empty() { + let (lkg_snap, lkg_stats) = loader::build_snapshot(&self.prefix, &injected); + if !lkg_stats.rejections.is_empty() { + let mut stale = self.stale_serving.lock().unwrap(); + for r in &lkg_stats.rejections { + tracing::error!( + key = %r.key, + error = %r.error, + "pinned last-known-good value no longer parses; dropping retention", + ); + stale.remove(&r.key); + } + } + merge_snapshot(&snap, &lkg_snap); + stats.partial_rows.extend(lkg_stats.partial_rows); + stats.partially_compatible = loader::aggregate_partial_compat(&stats.partial_rows); + if lkg_stats.accepted > 0 { + tracing::info!( + count = lkg_stats.accepted, + "serving last-known-good values for rejected keys", + ); + } + } + self.handle.store(snap); // Replace the cache-tracking map wholesale and flush. @@ -787,6 +987,10 @@ impl Supervisor

{ let state = self.state.lock().unwrap(); state.values().cloned().collect() }; + let stale: Vec = { + let guard = self.stale_serving.lock().unwrap(); + guard.values().cloned().collect() + }; let revision = *self.revision.lock().unwrap(); let cache = self.cache.clone(); // Spawn the actual write so the apply path stays sync. If we @@ -797,7 +1001,8 @@ impl Supervisor

{ // raced the spawn (~50ms wasn't enough on heavily loaded // GitHub Actions runners). if let Ok(rt_handle) = tokio::runtime::Handle::try_current() { - let join = rt_handle.spawn(async move { cache.store(&entries, revision).await }); + let join = + rt_handle.spawn(async move { cache.store(&entries, revision, &stale).await }); self.pending_writes.lock().unwrap().push(join); } } @@ -937,43 +1142,110 @@ async fn wait_for_cancel(mut rx: tokio::sync::watch::Receiver) { /// it doesn't materialise a deep copy of the `T` payload. fn clone_snapshot(src: &AisixSnapshot) -> AisixSnapshot { let out = AisixSnapshot::new(); - for e in src.models.entries() { - out.models.insert(clone_entry(&e)); - } - for e in src.apikeys.entries() { - out.apikeys.insert(clone_entry(&e)); - } - for e in src.provider_keys.entries() { - out.provider_keys.insert(clone_entry(&e)); - } - for e in src.guardrails.entries() { - out.guardrails.insert(clone_entry(&e)); - } - for e in src.guardrail_attachments.entries() { - out.guardrail_attachments.insert(clone_entry(&e)); - } - for e in src.cache_policies.entries() { - out.cache_policies.insert(clone_entry(&e)); - } - for e in src.observability_exporters.entries() { - out.observability_exporters.insert(clone_entry(&e)); - } - for e in src.rate_limit_policies.entries() { - out.rate_limit_policies.insert(clone_entry(&e)); - } - for e in src.mcp_servers.entries() { - out.mcp_servers.insert(clone_entry(&e)); - } - for e in src.mcp_policies.entries() { - out.mcp_policies.insert(clone_entry(&e)); - } - for e in src.a2a_agents.entries() { - out.a2a_agents.insert(clone_entry(&e)); + merge_snapshot(&out, src); + out +} + +/// Insert every entry of `src` into `dst` (replacing same-id entries). +/// The exhaustive destructuring makes adding a ResourceTable to +/// [`AisixSnapshot`] a compile error here — a missing kind would mean +/// entries silently drop on the floor when a watch put merges or a +/// last-known-good row is re-injected on resync. +fn merge_snapshot(dst: &AisixSnapshot, src: &AisixSnapshot) { + let AisixSnapshot { + models, + apikeys, + provider_keys, + guardrails, + guardrail_attachments, + cache_policies, + observability_exporters, + rate_limit_policies, + mcp_servers, + mcp_policies, + a2a_agents, + oidc_providers, + } = src; + for e in models.entries() { + dst.models.insert(clone_entry(&e)); + } + for e in apikeys.entries() { + dst.apikeys.insert(clone_entry(&e)); + } + for e in provider_keys.entries() { + dst.provider_keys.insert(clone_entry(&e)); + } + for e in guardrails.entries() { + dst.guardrails.insert(clone_entry(&e)); + } + for e in guardrail_attachments.entries() { + dst.guardrail_attachments.insert(clone_entry(&e)); + } + for e in cache_policies.entries() { + dst.cache_policies.insert(clone_entry(&e)); + } + for e in observability_exporters.entries() { + dst.observability_exporters.insert(clone_entry(&e)); + } + for e in rate_limit_policies.entries() { + dst.rate_limit_policies.insert(clone_entry(&e)); + } + for e in mcp_servers.entries() { + dst.mcp_servers.insert(clone_entry(&e)); + } + for e in mcp_policies.entries() { + dst.mcp_policies.insert(clone_entry(&e)); + } + for e in a2a_agents.entries() { + dst.a2a_agents.insert(clone_entry(&e)); + } + for e in oidc_providers.entries() { + dst.oidc_providers.insert(clone_entry(&e)); } - for e in src.oidc_providers.entries() { - out.oidc_providers.insert(clone_entry(&e)); +} + +/// Whether the snapshot holds an entry for `(kind, id)`. An unknown +/// kind reads as absent. Exhaustively destructured for the same +/// drift-guard reason as [`merge_snapshot`]: a kind added to the +/// snapshot but missed here would silently never pin a last known good. +fn snapshot_has(snap: &AisixSnapshot, kind: &str, id: &str) -> bool { + let AisixSnapshot { + models, + apikeys, + provider_keys, + guardrails, + guardrail_attachments, + cache_policies, + observability_exporters, + rate_limit_policies, + mcp_servers, + mcp_policies, + a2a_agents, + oidc_providers, + } = snap; + match kind { + "models" => models.get_by_id(id).is_some(), + "api_keys" => apikeys.get_by_id(id).is_some(), + "provider_keys" => provider_keys.get_by_id(id).is_some(), + "guardrails" => guardrails.get_by_id(id).is_some(), + "guardrail_attachments" => guardrail_attachments.get_by_id(id).is_some(), + "cache_policies" => cache_policies.get_by_id(id).is_some(), + "observability_exporters" => observability_exporters.get_by_id(id).is_some(), + "rate_limit_policies" => rate_limit_policies.get_by_id(id).is_some(), + "mcp_servers" => mcp_servers.get_by_id(id).is_some(), + "mcp_policies" => mcp_policies.get_by_id(id).is_some(), + "a2a_agents" => a2a_agents.get_by_id(id).is_some(), + "oidc_providers" => oidc_providers.get_by_id(id).is_some(), + _ => false, } - out +} + +/// Wall-clock seconds since the Unix epoch; zero on a pre-epoch clock. +fn now_unix_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) } /// Per-kind counts of the served snapshot, keyed by the plural etcd resource @@ -1453,15 +1725,15 @@ mod tests { sup.await_pending_cache_writes().await; let cache = SnapshotCache::new(&cache_path); - let (entries, _) = cache.load().expect("cache file present"); - assert_eq!(entries.len(), 2); + let cached = cache.load().expect("cache file present"); + assert_eq!(cached.entries.len(), 2); sup.apply_delete("/aisix/models/m-1"); sup.await_pending_cache_writes().await; - let (entries, _) = cache.load().expect("cache file present"); - assert_eq!(entries.len(), 1); - assert_eq!(entries[0].key, "/aisix/models/m-2"); + let cached = cache.load().expect("cache file present"); + assert_eq!(cached.entries.len(), 1); + assert_eq!(cached.entries[0].key, "/aisix/models/m-2"); } // ---- regression coverage for issue #114 ------------------------- @@ -1729,4 +2001,273 @@ mod tests { assert_eq!(sup.recent_partial_compat().len(), 1); assert_eq!(sup.recent_rejections().len(), 1); } + + // ---- RED last-known-good retention across resync/restart (#871 PR2) ---- + // + // A watch put that is rejected already leaves the previous good value + // serving (pinned above). But the retention used to end at the next + // full resync: `apply_resync` rebuilt the snapshot from accepted rows + // only, so a key whose latest etcd bytes are rejected VANISHED — an + // api_key would 401 byte-identically to "no such key", days after the + // write that caused it. The tests below pin the xDS-NACK-style fix: + // the last known good value keeps serving for as long as the etcd key + // exists, across resync and restart, with the staleness reported. + + #[tokio::test] + async fn rejected_update_keeps_last_good_serving_across_resync() { + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::new(provider, "/aisix"); + sup.load_once().await.unwrap(); + + assert!(sup.apply_put(&entry("/aisix/models/m-1", VALID_MODEL, 1))); + assert!(!sup.apply_put(&entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2))); + assert_eq!(sup.handle().load().models.len(), 1); + + // The next resync re-reads the full etcd state — which still + // holds the rejected bytes for this key. + sup.apply_resync(&[entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2)]); + assert_eq!( + sup.handle().load().models.len(), + 1, + "resync must keep serving the last known good value for a rejected key", + ); + + // The rejection signal persists every cycle, and the row is + // reported as serving-stale with its age. + assert_eq!(sup.recent_rejections().len(), 1); + let view = serde_json::to_value(sup.config_status().view()).unwrap(); + assert_eq!(view["rejected"].as_array().unwrap().len(), 1); + assert!( + view["rejected"][0]["serving_stale_since"].is_string(), + "rejected[] must carry the stale-serving timestamp: {view}", + ); + assert!( + view["rejected"][0]["serving_stale_age_seconds"].is_u64(), + "rejected[] must carry the staleness age: {view}", + ); + // The served row keeps counting. + assert_eq!(view["applied"]["resource_counts"]["models"], 1); + } + + #[tokio::test] + async fn rejected_update_keeps_last_good_serving_across_restart() { + let dir = tempfile::tempdir().unwrap(); + let cache_path = dir.path().join("snap.json"); + + // First lifecycle: a good row loads, then a resync observes the + // rejected replacement bytes (the etcd state after a newer CP + // wrote an update this DP cannot represent). The flushed cache + // must carry enough to survive a restart. + { + let provider = Arc::new(FakeProvider::new( + vec![entry("/aisix/models/m-1", VALID_MODEL, 1)], + 1, + )); + let sup = Supervisor::with_cache(provider, "/aisix", SnapshotCache::new(&cache_path)); + sup.load_once().await.unwrap(); + assert_eq!(sup.handle().load().models.len(), 1); + sup.apply_resync(&[entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2)]); + assert_eq!( + sup.handle().load().models.len(), + 1, + "pre-restart: the last known good value serves through the resync", + ); + sup.await_pending_cache_writes().await; + } + + // Second lifecycle (process restart, etcd unreachable): restore + // from disk. The last known good value must come back — without + // it the restart is the cliff where the resource silently dies. + { + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::with_cache(provider, "/aisix", SnapshotCache::new(&cache_path)); + sup.restore_from_cache(); + assert_eq!( + sup.handle().load().models.len(), + 1, + "restart must restore the last known good value for a rejected key", + ); + assert_eq!( + sup.recent_rejections().len(), + 1, + "the rejection signal must survive the restart too", + ); + } + } + + #[tokio::test] + async fn deleting_a_rejected_never_serving_key_clears_observed_state() { + // Audit finding on #871 PR2: a rejected put now mirrors its + // bytes into the observed-state map even when the row never + // served (no pin). Deleting that key takes the `!present` early + // return in apply_delete, which must still drop the bytes from + // `state` — otherwise the deleted key haunts source_hash until + // the next resync and its document persists in the cache file. + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::new(provider, "/aisix"); + sup.load_once().await.unwrap(); + let clean_hash = sup.config_status().view().source.source_hash; + + // Never served: the very first put for the key is rejected. + assert!(!sup.apply_put(&entry("/aisix/models/m-bad", BAD_PROVIDER_MODEL, 1))); + assert!(sup.handle().load().models.is_empty()); + assert_ne!( + sup.config_status().view().source.source_hash, + clean_hash, + "the rejected bytes are part of the observed etcd state", + ); + + // The delete finds nothing in the snapshot but must still clear + // the observed-state entry (and the rejection — clearing it is + // "something removed", so the call reports true). + assert!(sup.apply_delete("/aisix/models/m-bad")); + assert!(sup.recent_rejections().is_empty()); + assert_eq!( + sup.config_status().view().source.source_hash, + clean_hash, + "a deleted key must leave the observed etcd state immediately", + ); + } + + #[tokio::test] + async fn rejected_put_persists_pin_for_immediate_restart() { + // A restart INSIDE the rejected-put window (before any resync + // fixed the state to disk) must behave like a post-resync + // restart: the rejected bytes and the pinned last-good ride the + // cache together, so the value keeps serving AND the staleness + // clock stays continuous instead of resetting at boot. + let dir = tempfile::tempdir().unwrap(); + let cache_path = dir.path().join("snap.json"); + let since_before; + + { + let provider = Arc::new(FakeProvider::new( + vec![entry("/aisix/models/m-1", VALID_MODEL, 1)], + 1, + )); + let sup = Supervisor::with_cache(provider, "/aisix", SnapshotCache::new(&cache_path)); + sup.load_once().await.unwrap(); + assert!(!sup.apply_put(&entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2))); + since_before = sup.recent_rejections()[0] + .stale_serving_since_unix_secs + .expect("rejected put with a serving value must report stale-since"); + sup.await_pending_cache_writes().await; + } + + { + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::with_cache(provider, "/aisix", SnapshotCache::new(&cache_path)); + sup.restore_from_cache(); + assert_eq!( + sup.handle().load().models.len(), + 1, + "restart in the rejected-put window must restore the pinned value", + ); + let rejections = sup.recent_rejections(); + assert_eq!(rejections.len(), 1); + assert_eq!( + rejections[0].stale_serving_since_unix_secs, + Some(since_before), + "the staleness clock must be continuous across the restart", + ); + } + } + + #[tokio::test] + async fn stale_served_row_dies_with_etcd_delete() { + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::new(provider, "/aisix"); + sup.load_once().await.unwrap(); + + assert!(sup.apply_put(&entry("/aisix/models/m-1", VALID_MODEL, 1))); + assert!(!sup.apply_put(&entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2))); + sup.apply_resync(&[entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2)]); + assert_eq!(sup.handle().load().models.len(), 1); + + // The admin deletes the resource: the last known good goes with + // it — retention must never outlive the etcd key. + assert!(sup.apply_delete("/aisix/models/m-1")); + assert!(sup.handle().load().models.is_empty()); + assert!(sup.recent_rejections().is_empty()); + // A later resync confirming the key's absence keeps it gone. + sup.apply_resync(&[]); + assert!(sup.handle().load().models.is_empty()); + } + + #[tokio::test] + async fn stale_served_row_dies_when_resync_no_longer_carries_the_key() { + // Same zombie guard for the resync-observed deletion: a key that + // disappears from the full etcd read (no watch Delete seen, e.g. + // reconnect after compaction) must drop its last known good. + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::new(provider, "/aisix"); + sup.load_once().await.unwrap(); + + assert!(sup.apply_put(&entry("/aisix/models/m-1", VALID_MODEL, 1))); + assert!(!sup.apply_put(&entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2))); + sup.apply_resync(&[entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2)]); + assert_eq!(sup.handle().load().models.len(), 1); + + sup.apply_resync(&[]); + assert!( + sup.handle().load().models.is_empty(), + "a key absent from the resynced etcd state must not keep serving", + ); + assert!(sup.recent_rejections().is_empty()); + } + + #[tokio::test] + async fn stale_last_good_that_was_yellow_keeps_its_partial_compat_signal() { + // The value actually serving is itself YELLOW (unknown field + // ignored), and the newer update is RED-rejected. Both signals + // must coexist across a resync: rejected[] describes the new + // bytes, partially_compatible[] describes the served old value. + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::new(provider, "/aisix"); + sup.load_once().await.unwrap(); + + assert!(sup.apply_put(&entry("/aisix/models/m-1", YELLOW_MODEL, 1))); + assert!(!sup.apply_put(&entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2))); + sup.apply_resync(&[entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2)]); + + assert_eq!(sup.handle().load().models.len(), 1); + assert_eq!(sup.recent_rejections().len(), 1); + let agg = sup.recent_partial_compat(); + assert_eq!( + agg.len(), + 1, + "the served YELLOW last-good keeps reporting its ignored fields", + ); + assert_eq!(agg[0].field, "future_knob"); + } + + #[tokio::test] + async fn config_hash_reflects_served_bytes_not_rejected_bytes() { + let provider = Arc::new(FakeProvider::new(vec![], 0)); + let sup = Supervisor::new(provider, "/aisix"); + sup.load_once().await.unwrap(); + + assert!(sup.apply_put(&entry("/aisix/models/m-1", VALID_MODEL, 1))); + let good_hash = sup.config_status().view().applied.unwrap().config_hash; + + sup.apply_resync(&[entry("/aisix/models/m-1", BAD_PROVIDER_MODEL, 2)]); + let view = sup.config_status().view(); + let applied = view.applied.unwrap(); + // What's served didn't change, so the served-config hash must not + // change either: the rejected bytes never enter config_hash (the + // hash must not claim the new value applied), and the row must + // not silently drop out of it (the hash must not claim the row + // stopped serving). + assert_eq!( + applied.config_hash, good_hash, + "config_hash must cover the bytes actually served (the last known good)", + ); + // source_hash reflects the observed etcd state (the rejected + // bytes), so the two hashes diverge — the honest "not converged" + // signal, explained by rejected[]. + assert_ne!( + Some(applied.config_hash.as_str()), + view.source.source_hash.as_deref(), + ); + } } diff --git a/crates/aisix-obs/src/metrics.rs b/crates/aisix-obs/src/metrics.rs index 2b9592e0..87c064da 100644 --- a/crates/aisix-obs/src/metrics.rs +++ b/crates/aisix-obs/src/metrics.rs @@ -184,6 +184,12 @@ pub const M_CONFIG_REJECTED_RESOURCES: &str = "aisix_config_rejected_resources"; /// data plane's rollout. pub const M_CONFIG_PARTIALLY_COMPATIBLE_RESOURCES: &str = "aisix_config_partially_compatible_resources"; +/// Served resources per kind whose latest source bytes are rejected and +/// whose last known good value serves instead (#871). Non-zero means the +/// gateway is running stale config for those rows; the per-row detail +/// (which resource, stale since when) lives in `/status/config` +/// `rejected[]`. +pub const M_CONFIG_STALE_SERVED_RESOURCES: &str = "aisix_config_stale_served_resources"; pub const M_CONFIG_OBSERVED_REVISION: &str = "aisix_config_observed_revision"; pub const M_CONFIG_APPLIED_REVISION: &str = "aisix_config_applied_revision"; pub const M_CONFIG_HASH_INFO: &str = "aisix_config_hash_info"; @@ -350,6 +356,7 @@ struct ConfigLabelState { last_hash: Option, last_rejected_kinds: std::collections::HashSet, last_partial_kinds: std::collections::HashSet, + last_stale_kinds: std::collections::HashSet, } const REQUEST_SERIES_CACHE_CAPACITY: usize = 1024; @@ -715,6 +722,19 @@ impl Metrics { .set(*count as f64); } labels.last_partial_kinds = view.partially_compatible_by_kind.keys().cloned().collect(); + + // Stale-served gauge per kind (#871), same zeroing discipline. + for kind in &labels.last_stale_kinds { + if !view.stale_served_by_kind.contains_key(kind) { + metrics::gauge!(M_CONFIG_STALE_SERVED_RESOURCES, "kind" => kind.clone()) + .set(0.0); + } + } + for (kind, count) in &view.stale_served_by_kind { + metrics::gauge!(M_CONFIG_STALE_SERVED_RESOURCES, "kind" => kind.clone()) + .set(*count as f64); + } + labels.last_stale_kinds = view.stale_served_by_kind.keys().cloned().collect(); }); } @@ -3050,6 +3070,7 @@ mod tests { reload_failures: std::collections::BTreeMap::new(), rejected_by_kind: std::collections::BTreeMap::new(), partially_compatible_by_kind: std::collections::BTreeMap::new(), + stale_served_by_kind: std::collections::BTreeMap::new(), observed_revision: Some(42), applied_revision: Some(42), config_hash: Some("abc123".into()), @@ -3066,6 +3087,7 @@ mod tests { view.rejected_by_kind.insert("models".to_string(), 1); view.partially_compatible_by_kind .insert("api_keys".to_string(), 3); + view.stale_served_by_kind.insert("models".to_string(), 1); m.sync_config_status(&view); let out = m.render(); @@ -3081,6 +3103,9 @@ mod tests { assert!(out.contains(&format!( "{M_CONFIG_PARTIALLY_COMPATIBLE_RESOURCES}{{kind=\"api_keys\"}} 3" ))); + assert!(out.contains(&format!( + "{M_CONFIG_STALE_SERVED_RESOURCES}{{kind=\"models\"}} 1" + ))); assert!(out.contains(&format!("{M_CONFIG_OBSERVED_REVISION} 42"))); assert!(out.contains(&format!("{M_CONFIG_APPLIED_REVISION} 42"))); assert!(out.contains(&format!("{M_CONFIG_HASH_INFO}{{hash=\"abc123\"}} 1"))); @@ -3111,6 +3136,7 @@ mod tests { first .partially_compatible_by_kind .insert("api_keys".to_string(), 1); + first.stale_served_by_kind.insert("models".to_string(), 1); m.sync_config_status(&first); // The applied config changes and the models rejection clears. @@ -3131,6 +3157,10 @@ mod tests { assert!(out.contains(&format!( "{M_CONFIG_PARTIALLY_COMPATIBLE_RESOURCES}{{kind=\"api_keys\"}} 0" ))); + // And for the stale-served gauge (#871). + assert!(out.contains(&format!( + "{M_CONFIG_STALE_SERVED_RESOURCES}{{kind=\"models\"}} 0" + ))); } /// Issue #408 audit MEDIUM-2: pin every boundary of diff --git a/crates/aisix-server/src/heartbeat.rs b/crates/aisix-server/src/heartbeat.rs index e37323f2..2f041fcc 100644 --- a/crates/aisix-server/src/heartbeat.rs +++ b/crates/aisix-server/src/heartbeat.rs @@ -413,6 +413,14 @@ struct RejectedResourceWire { kind: &'static str, error: String, timestamp_unix_secs: u64, + /// Unix seconds since when this key has been serving its last known + /// good value instead of the rejected bytes (#871) — present on + /// every beat while stale serving lasts, so cp-api can derive the + /// staleness age each cycle. Omitted when nothing serves for the + /// key, preserving the historical body shape for older CPs (whose + /// heartbeat handler tolerates unknown fields either way). + #[serde(skip_serializing_if = "Option::is_none")] + stale_serving_since_unix_secs: Option, } impl From<&RejectedEntry> for RejectedResourceWire { @@ -422,6 +430,7 @@ impl From<&RejectedEntry> for RejectedResourceWire { kind: r.kind.as_str(), error: r.error.clone(), timestamp_unix_secs: r.timestamp_unix_secs, + stale_serving_since_unix_secs: r.stale_serving_since_unix_secs, } } } @@ -793,6 +802,57 @@ mod tests { ); } + /// #871: a rejection whose key still serves its last known good value + /// carries the stale-serving instant on every beat; one with nothing + /// serving omits the field (historical body shape). + #[tokio::test] + async fn send_includes_stale_serving_since_on_rejections() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/dp/heartbeat")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "ok": true + }))) + .mount(&server) + .await; + + let dir = tempfile::tempdir().unwrap(); + let mtls = write_test_bundle(dir.path()); + let cfg = cfg_with_bundle(format!("{}/dp/heartbeat", server.uri()), mtls) + .with_rejection_fetcher(Arc::new(|| { + vec![ + RejectedEntry { + key: "/aisix/models/stale-served".into(), + kind: aisix_etcd::loader::RejectionKind::SchemaFailed, + error: "schema validation failed".into(), + timestamp_unix_secs: 1_770_000_100, + stale_serving_since_unix_secs: Some(1_770_000_000), + }, + RejectedEntry { + key: "/aisix/models/never-loaded".into(), + kind: aisix_etcd::loader::RejectionKind::SchemaFailed, + error: "schema validation failed".into(), + timestamp_unix_secs: 1_770_000_100, + stale_serving_since_unix_secs: None, + }, + ] + })); + send(&plain_client(), &cfg, 7).await.unwrap(); + + let received = server.received_requests().await.unwrap(); + let body: serde_json::Value = serde_json::from_slice(&received[0].body).unwrap(); + let rejected = body["rejected_resources"].as_array().unwrap(); + assert_eq!(rejected.len(), 2); + assert_eq!( + rejected[0]["stale_serving_since_unix_secs"], + 1_770_000_000_i64, + ); + assert!( + rejected[1].get("stale_serving_since_unix_secs").is_none(), + "a rejection with nothing serving must omit the stale field", + ); + } + #[tokio::test] async fn send_includes_partially_compatible_resources_when_wired() { let server = MockServer::start().await; diff --git a/crates/aisix-server/src/main.rs b/crates/aisix-server/src/main.rs index 322230c0..34e1ab1d 100644 --- a/crates/aisix-server/src/main.rs +++ b/crates/aisix-server/src/main.rs @@ -575,17 +575,15 @@ async fn run(mut cfg: Config) -> anyhow::Result<()> { etcd_prefix.clone(), )) }; - // Snapshot cache: in managed mode persist to disk (default - // /var/lib/aisix/config_cache.json) so the DP can serve traffic + // Snapshot cache: persist to disk so the DP can serve traffic // from the last-known config across CP outages and restarts. - // Disabled outside managed mode and when the operator clears the - // path explicitly. - let snapshot_cache = - if cfg.managed.is_managed() && !cfg.managed.snapshot_cache_path.is_empty() { - SnapshotCache::new(&cfg.managed.snapshot_cache_path) - } else { - SnapshotCache::disabled() - }; + // Managed mode defaults to /var/lib/aisix/config_cache.json; + // self-hosted etcd mode enables it only when the operator sets + // a path explicitly; "" disables it in either mode. + let snapshot_cache = match cfg.managed.effective_snapshot_cache_path() { + Some(path) => SnapshotCache::new(path), + None => SnapshotCache::disabled(), + }; let supervisor = Arc::new(Supervisor::with_cache( provider, etcd_prefix, diff --git a/tests/e2e/src/cases/config-last-known-good-e2e.test.ts b/tests/e2e/src/cases/config-last-known-good-e2e.test.ts new file mode 100644 index 00000000..16f84f31 --- /dev/null +++ b/tests/e2e/src/cases/config-last-known-good-e2e.test.ts @@ -0,0 +1,252 @@ +import { createHash, randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + EtcdClient, + ProxyClient, + SeedClient, + spawnApp, + startOpenAiUpstream, + waitConfigPropagation, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// E2E for RED last-known-good retention (issue #871, PR 2). When an etcd +// document is updated to bytes this data-plane version cannot represent +// (RED — schema violation), the previously accepted value must keep +// serving real traffic for as long as the etcd key exists: +// +// - immediately after the rejected watch update (already true pre-PR); +// - across a full restart, which replays the snapshot cache AND runs a +// fresh load_all + resync against the still-rejected etcd bytes — the +// two paths that used to silently drop the row (the "cliff"); +// - with the staleness observable every cycle: `rejected[]` on +// `GET /status/config` carries `serving_stale_since` (stable across +// the restart) and a recomputed `serving_stale_age_seconds`, and the +// metrics listener exposes a per-kind stale-served gauge; +// - until the etcd key is DELETED, at which point the pinned value dies +// with it — retention must never outlive the key. + +const CALLER_PLAINTEXT = "sk-last-known-good-caller"; +const CALLER_KEY_HASH = createHash("sha256").update(CALLER_PLAINTEXT).digest("hex"); + +interface StatusConfig { + state: string; + applied?: { resource_counts: Record }; + rejected: Array<{ + resource_kind: string; + resource_id: string; + serving_stale_since?: string; + serving_stale_age_seconds?: number; + }>; + partially_compatible: Array<{ resource_kind: string; field: string; count: number }>; +} + +async function getStatusConfig(app: SpawnedApp): Promise { + const res = await fetch(`${app.metricsUrl}/status/config`); + expect(res.status).toBe(200); + return (await res.json()) as StatusConfig; +} + +async function scrape(app: SpawnedApp): Promise { + const res = await fetch(`${app.metricsUrl}/metrics`); + expect(res.status).toBe(200); + return res.text(); +} + +describe("config last-known-good: rejected updates keep serving across resync and restart", () => { + let app: SpawnedApp | undefined; + let stoppedApp: SpawnedApp | undefined; + let upstream: OpenAiUpstream | undefined; + let etcd: EtcdClient | undefined; + let etcdReachable = false; + let cacheDir: string | undefined; + + const etcdPrefix = `/aisix-e2e-lkg-${randomUUID()}`; + let modelId: string; + let modelKey: string; + let staleSinceBeforeRestart: string | undefined; + + beforeAll(async () => { + etcd = new EtcdClient(); + etcdReachable = await etcd.ping(); + if (!etcdReachable) return; + + upstream = await startOpenAiUpstream(); + cacheDir = await mkdtemp(join(tmpdir(), "aisix-lkg-cache-")); + app = await spawnApp({ + etcdPrefix, + snapshotCachePath: join(cacheDir, "config_cache.json"), + }); + + const seed = new SeedClient(etcd, etcdPrefix); + const pk = await seed.createProviderKey({ + display_name: "lkg-pk", + secret: "sk-mock", + api_base: `${upstream.baseUrl}/v1`, + }); + const model = await seed.createModel({ + display_name: "lkg-model", + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: pk.id, + }); + modelId = model.id; + modelKey = `${etcdPrefix}/models/${modelId}`; + await seed.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: ["lkg-model"], + }); + }); + + afterAll(async () => { + await app?.exit(); + // The pre-restart app was stop()ped without cleanup; exit() is + // idempotent on the dead process and reclaims its tmp dir. + await stoppedApp?.exit(); + await upstream?.close(); + if (cacheDir) await rm(cacheDir, { recursive: true, force: true }); + }); + + test("a rejected update leaves the old value serving real traffic, with staleness reported", async (ctx) => { + if (!etcdReachable || !app || !etcd) { + ctx.skip(); + return; + } + + await waitConfigPropagation(async () => { + const cfg = await getStatusConfig(app!); + return (cfg.applied?.resource_counts.models ?? 0) >= 1; + }); + + // Baseline: the model serves. + const proxy = new ProxyClient(app.proxyUrl, CALLER_PLAINTEXT); + const before = await proxy.chat({ + model: "lkg-model", + messages: [{ role: "user", content: "baseline" }], + }); + expect(before.status, JSON.stringify(before.body)).toBe(200); + + // A newer control plane (or a bug) replaces the document with bytes + // this DP rejects: empty display_name violates the schema (RED). + await etcd.put( + modelKey, + JSON.stringify({ + display_name: "", + provider: "openai", + model_name: "gpt-4o-mini", + }), + ); + + let cfg: StatusConfig | undefined; + await waitConfigPropagation(async () => { + cfg = await getStatusConfig(app!); + return cfg.rejected.some((r) => r.resource_id === modelId); + }); + + // The old value keeps serving REAL traffic — the user journey the + // cliff used to break. + const during = await proxy.chat({ + model: "lkg-model", + messages: [{ role: "user", content: "still serving?" }], + }); + expect(during.status, JSON.stringify(during.body)).toBe(200); + + // The rejection is reported with the staleness attached. + expect(cfg!.state).toBe("degraded"); + const rejection = cfg!.rejected.find((r) => r.resource_id === modelId)!; + expect(rejection.resource_kind).toBe("models"); + expect(rejection.serving_stale_since).toBeTypeOf("string"); + expect(rejection.serving_stale_age_seconds).toBeTypeOf("number"); + staleSinceBeforeRestart = rejection.serving_stale_since; + // The served row keeps counting. + expect(cfg!.applied?.resource_counts.models).toBe(1); + + // And the per-kind gauge on the metrics listener. + const text = await scrape(app); + expect(text).toMatch(/aisix_config_stale_served_resources\{kind="models"\} 1/); + }); + + test("the last known good survives a restart (cache replay + live resync against rejected bytes)", async (ctx) => { + if (!etcdReachable || !app || !etcd) { + ctx.skip(); + return; + } + + // Full process restart on the same etcd prefix + snapshot cache. + stoppedApp = app; + await app.stop(); + app = await spawnApp({ + etcdPrefix, + snapshotCachePath: join(cacheDir!, "config_cache.json"), + }); + + // Prove the LIVE etcd read completed (not just the cache replay): + // a sentinel written after the restart can only appear via the new + // process's load_all/watch. By then the boot resync has re-read the + // rejected bytes for the model key — the exact path that used to + // drop the row. + await etcd.put( + `${etcdPrefix}/api_keys/${randomUUID()}`, + JSON.stringify({ + key_hash: createHash("sha256").update(`sentinel-${randomUUID()}`).digest("hex"), + allowed_models: [], + }), + ); + await waitConfigPropagation(async () => { + const cfg = await getStatusConfig(app!); + return (cfg.applied?.resource_counts.api_keys ?? 0) >= 2; + }); + + // The model still serves real traffic from its pinned value. + const proxy = new ProxyClient(app.proxyUrl, CALLER_PLAINTEXT); + const after = await proxy.chat({ + model: "lkg-model", + messages: [{ role: "user", content: "post-restart" }], + }); + expect(after.status, JSON.stringify(after.body)).toBe(200); + + // The rejection + staleness survive, and the stale-since instant is + // CONTINUOUS across the restart (persisted in the snapshot cache), + // so the age keeps growing instead of resetting. + const cfg = await getStatusConfig(app); + const rejection = cfg.rejected.find((r) => r.resource_id === modelId)!; + expect(rejection).toBeDefined(); + expect(rejection.serving_stale_since).toBe(staleSinceBeforeRestart); + expect(rejection.serving_stale_age_seconds).toBeTypeOf("number"); + expect(cfg.applied?.resource_counts.models).toBe(1); + + const text = await scrape(app); + expect(text).toMatch(/aisix_config_stale_served_resources\{kind="models"\} 1/); + }); + + test("deleting the etcd key kills the pinned value — no zombie config", async (ctx) => { + if (!etcdReachable || !app || !etcd) { + ctx.skip(); + return; + } + + await etcd.delete(modelKey); + await waitConfigPropagation(async () => { + const cfg = await getStatusConfig(app!); + return (cfg.applied?.resource_counts.models ?? 0) === 0; + }); + + // The resource is gone for real traffic... + const proxy = new ProxyClient(app.proxyUrl, CALLER_PLAINTEXT); + const gone = await proxy.chat({ + model: "lkg-model", + messages: [{ role: "user", content: "should be gone" }], + }); + expect(gone.status, JSON.stringify(gone.body)).toBe(404); + + // ...and every stale/rejected signal clears with it. + const cfg = await getStatusConfig(app); + expect(cfg.rejected).toHaveLength(0); + const text = await scrape(app); + expect(text).toMatch(/aisix_config_stale_served_resources\{kind="models"\} 0/); + }); +}); diff --git a/tests/e2e/src/harness/app.ts b/tests/e2e/src/harness/app.ts index d5bc9ea8..f8dbc7ee 100644 --- a/tests/e2e/src/harness/app.ts +++ b/tests/e2e/src/harness/app.ts @@ -86,6 +86,21 @@ export interface AppOverrides { * and send SIGHUP to exercise reloads. */ resourcesFile?: string; + /** + * Reuse a fixed etcd prefix instead of generating a fresh one. For + * restart scenarios: `stop()` the first app (keeps etcd data), then + * spawn a second one with the same prefix so it loads the survivor + * state. The LAST app spawned on the prefix should `exit()` to clean + * it up. + */ + etcdPrefix?: string; + /** + * `managed.snapshot_cache_path` — enables the on-disk snapshot cache + * (#871) without managed mode. Point two sequential apps (same + * `etcdPrefix`) at one path to exercise cache-restored restarts. + * The caller owns the file's lifecycle. + */ + snapshotCachePath?: string; } export interface SpawnedApp { @@ -112,6 +127,13 @@ export interface SpawnedApp { output(): string; signal(signal: NodeJS.Signals): void; exit(): Promise; + /** + * Terminate the binary WITHOUT cleaning up: the etcd prefix, the tmp + * config dir, and any snapshot cache file survive. For restart + * scenarios — spawn a successor with the same `etcdPrefix` / + * `snapshotCachePath`, and let the successor's `exit()` clean up. + */ + stop(): Promise; } const BIN_PATH = @@ -186,7 +208,7 @@ async function spawnAppOnce(overrides: AppOverrides = {}): Promise { } const [proxyPort, adminPort, metricsPort] = await pickFreePorts(3); const adminKey = overrides.adminKey ?? `admin-${randomUUID()}`; - const etcdPrefix = `/aisix-e2e-${randomUUID()}`; + const etcdPrefix = overrides.etcdPrefix ?? `/aisix-e2e-${randomUUID()}`; const dir = await mkdtemp(join(tmpdir(), "aisix-e2e-")); let resourcesPath: string | undefined; @@ -233,6 +255,9 @@ async function spawnAppOnce(overrides: AppOverrides = {}): Promise { tracing: { otlp: { enabled: false, endpoint: "http://127.0.0.1:4317", sample_ratio: 1 } }, }, cache: { backend: "memory" }, + ...(overrides.snapshotCachePath + ? { managed: { snapshot_cache_path: overrides.snapshotCachePath } } + : {}), ...(overrides.extra ?? {}), }; @@ -350,6 +375,9 @@ async function spawnAppOnce(overrides: AppOverrides = {}): Promise { await terminate(child); await cleanup(fileMode ? undefined : etcd, etcdPrefix, dir); }, + async stop() { + await terminate(child); + }, }; }