From f27817a883696b0d46032ef275518ed434186018 Mon Sep 17 00:00:00 2001 From: Yuansheng Wang Date: Tue, 4 Aug 2026 16:35:34 +0800 Subject: [PATCH 1/3] feat(config): serve last-known-good values for rejected keys across resync and restart MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A watch put that fails validation already leaves the previous good value serving, but the retention ended at the next full resync: the snapshot was rebuilt from accepted rows only, so a key whose latest etcd bytes are rejected silently vanished — an api_key would 401 identically to no such key, days after the write that broke it (issue #871, the pre- existing cliff PR #872 deliberately left in place). The supervisor now pins the serving bytes as the key's last known good the moment a replacement is rejected, and re-injects the pinned value on every resync (xDS-NACK style). Retention ends exactly when the etcd key loads cleanly again or is deleted — including a key observed absent from a resynced full read — so a pin can never outlive its key. The pinned values ride in the snapshot cache (additive stale section, same format version) so retention and the staleness clock survive restarts. Accounting picks the honest option on both axes: config_hash covers the bytes each key actually serves (pinned bytes for stale keys, nothing for rejected keys with no last good), so it neither claims the rejected bytes applied nor that the row stopped serving; resource_counts keeps counting served rows. Staleness is reported every cycle: rejected[] on /status/config gains serving_stale_since plus a recomputed serving_stale_age_seconds, the heartbeat rejected_resources entries gain stale_serving_since_unix_secs, and a per-kind aisix_config_stale_served_resources gauge lands next to the existing partially-compatible gauge with the same label-zeroing discipline. The pinned bytes rebuild through the normal loader on injection, so a last good value that is itself partially compatible keeps reporting its ignored fields next to the rejection for the same key. The stale map is deliberately uncapped and independent of the 256-cap rejection buffer: overflow there truncates a report, never takes a served row offline. managed.snapshot_cache_path becomes optional: omitted keeps today's defaults (managed on at /var/lib/aisix/config_cache.json, self-hosted etcd off), an explicit path now enables the cache in self-hosted etcd mode too, and an empty string still disables it everywhere. --- crates/aisix-admin/src/lib.rs | 4 + crates/aisix-core/src/config.rs | 74 ++- crates/aisix-core/src/config_status.rs | 112 ++++ crates/aisix-core/src/filesource/status.rs | 6 + crates/aisix-etcd/src/loader.rs | 7 + crates/aisix-etcd/src/snapshot_cache.rs | 133 ++++- crates/aisix-etcd/src/supervisor.rs | 590 +++++++++++++++++---- crates/aisix-obs/src/metrics.rs | 30 ++ crates/aisix-server/src/heartbeat.rs | 60 +++ crates/aisix-server/src/main.rs | 18 +- 10 files changed, 902 insertions(+), 132 deletions(-) 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..1258b4a9 100644 --- a/crates/aisix-core/src/config.rs +++ b/crates/aisix-core/src/config.rs @@ -258,10 +258,15 @@ 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. + #[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 +301,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 +2109,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..94a4b7a4 100644 --- a/crates/aisix-core/src/config_status.rs +++ b/crates/aisix-core/src/config_status.rs @@ -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..f973489c 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,6 +276,9 @@ 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; @@ -258,18 +294,43 @@ impl Supervisor

{ 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 +343,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 +355,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 +374,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 +531,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 +608,52 @@ 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 never touches + /// `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. + self.capture_last_good(&entry.key); // 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. @@ -547,51 +680,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,27 +740,13 @@ 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); drop(snap); if !present { if removed_rejection { @@ -736,8 +821,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 +960,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 +974,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 +1115,80 @@ 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(); + merge_snapshot(&out, src); + out +} + +/// Insert every entry of `src` into `dst` (replacing same-id entries). +/// Must cover every ResourceTable on [`AisixSnapshot`] — a missing kind +/// here means entries of that kind 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) { for e in src.models.entries() { - out.models.insert(clone_entry(&e)); + dst.models.insert(clone_entry(&e)); } for e in src.apikeys.entries() { - out.apikeys.insert(clone_entry(&e)); + dst.apikeys.insert(clone_entry(&e)); } for e in src.provider_keys.entries() { - out.provider_keys.insert(clone_entry(&e)); + dst.provider_keys.insert(clone_entry(&e)); } for e in src.guardrails.entries() { - out.guardrails.insert(clone_entry(&e)); + dst.guardrails.insert(clone_entry(&e)); } for e in src.guardrail_attachments.entries() { - out.guardrail_attachments.insert(clone_entry(&e)); + dst.guardrail_attachments.insert(clone_entry(&e)); } for e in src.cache_policies.entries() { - out.cache_policies.insert(clone_entry(&e)); + dst.cache_policies.insert(clone_entry(&e)); } for e in src.observability_exporters.entries() { - out.observability_exporters.insert(clone_entry(&e)); + dst.observability_exporters.insert(clone_entry(&e)); } for e in src.rate_limit_policies.entries() { - out.rate_limit_policies.insert(clone_entry(&e)); + dst.rate_limit_policies.insert(clone_entry(&e)); } for e in src.mcp_servers.entries() { - out.mcp_servers.insert(clone_entry(&e)); + dst.mcp_servers.insert(clone_entry(&e)); } for e in src.mcp_policies.entries() { - out.mcp_policies.insert(clone_entry(&e)); + dst.mcp_policies.insert(clone_entry(&e)); } for e in src.a2a_agents.entries() { - out.a2a_agents.insert(clone_entry(&e)); + dst.a2a_agents.insert(clone_entry(&e)); } for e in src.oidc_providers.entries() { - out.oidc_providers.insert(clone_entry(&e)); + dst.oidc_providers.insert(clone_entry(&e)); + } +} + +/// Whether the snapshot holds an entry for `(kind, id)`. Mirrors the +/// kind dispatch of [`Supervisor::apply_delete`]'s presence probe; an +/// unknown kind reads as absent. +fn snapshot_has(snap: &AisixSnapshot, kind: &str, id: &str) -> bool { + match kind { + "models" => snap.models.get_by_id(id).is_some(), + "api_keys" => snap.apikeys.get_by_id(id).is_some(), + "provider_keys" => snap.provider_keys.get_by_id(id).is_some(), + "guardrails" => snap.guardrails.get_by_id(id).is_some(), + "guardrail_attachments" => snap.guardrail_attachments.get_by_id(id).is_some(), + "cache_policies" => snap.cache_policies.get_by_id(id).is_some(), + "observability_exporters" => snap.observability_exporters.get_by_id(id).is_some(), + "rate_limit_policies" => snap.rate_limit_policies.get_by_id(id).is_some(), + "mcp_servers" => snap.mcp_servers.get_by_id(id).is_some(), + "mcp_policies" => snap.mcp_policies.get_by_id(id).is_some(), + "a2a_agents" => snap.a2a_agents.get_by_id(id).is_some(), + "oidc_providers" => snap.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 +1668,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 +1944,195 @@ 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 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, From a37a3abeb2610c3eeeff9c0dfe6270ff00efd3c7 Mon Sep 17 00:00:00 2001 From: Yuansheng Wang Date: Tue, 4 Aug 2026 16:46:20 +0800 Subject: [PATCH 2/3] test(e2e): last-known-good survives restart with a continuous staleness clock Real-etcd coverage for the issue scenario: a seeded model serves a real chat request, a schema-invalid replacement lands in etcd, and the old value keeps serving through the rejected watch put, a full process restart (snapshot-cache replay plus a live load_all resync against the still-rejected bytes), and dies only when the etcd key is deleted. The staleness surface is asserted at every step: serving_stale_since and serving_stale_age_seconds on rejected[], and the per-kind aisix_config_stale_served_resources gauge including its zeroing. The restart leg caught a real gap: a rejected watch put neither mirrored the rejected bytes into the observed-state map nor flushed the cache, so a restart inside that window rebuilt a cache shape with no pin and the staleness clock reset at boot. A rejected put now records its bytes in the state map (source_hash reflects the observed etcd state immediately, matching what a resync would report) and flushes, so a restart in any window restores the same rejected-bytes + pinned-value shape. Pinned by a new unit test asserting stale-since continuity across a restart with no intervening resync. Harness: spawnApp gains etcdPrefix and snapshotCachePath overrides and apps gain stop() (terminate without cleanup) so restart scenarios can hand etcd state and the cache file to a successor process. --- crates/aisix-core/src/config_status.rs | 18 +- crates/aisix-etcd/src/supervisor.rs | 84 +++++- .../cases/config-last-known-good-e2e.test.ts | 252 ++++++++++++++++++ tests/e2e/src/harness/app.ts | 30 ++- 4 files changed, 364 insertions(+), 20 deletions(-) create mode 100644 tests/e2e/src/cases/config-last-known-good-e2e.test.ts diff --git a/crates/aisix-core/src/config_status.rs b/crates/aisix-core/src/config_status.rs index 94a4b7a4..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. diff --git a/crates/aisix-etcd/src/supervisor.rs b/crates/aisix-etcd/src/supervisor.rs index f973489c..9a84b5ae 100644 --- a/crates/aisix-etcd/src/supervisor.rs +++ b/crates/aisix-etcd/src/supervisor.rs @@ -285,12 +285,11 @@ impl Supervisor

{ { 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(); @@ -616,9 +615,10 @@ impl Supervisor

{ /// /// 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 never touches - /// `state`, and a resync that rejects a key either stale-tracks it - /// or drops it from the snapshot). + /// 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; @@ -652,8 +652,27 @@ impl Supervisor

{ 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. + // 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. @@ -667,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; } @@ -2038,6 +2058,50 @@ mod tests { } } + #[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)); 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); + }, }; } From 480dbca0c4971bd9356581d190eea73096297a17 Mon Sep 17 00:00:00 2001 From: Yuansheng Wang Date: Tue, 4 Aug 2026 17:07:18 +0800 Subject: [PATCH 3/3] fix(config): drop deleted keys from observed state on the not-present delete path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Independent audit finding (MEDIUM) on the last-known-good change: a rejected put now mirrors its bytes into the observed-state map even when the row never served, so a subsequent watch delete takes the not-present early return in apply_delete — which cleared the rejection and the pin but left the key's bytes in the state map and never re-flushed the cache. The deleted key then haunted source_hash until the next resync and its document (secret-bearing for provider keys) persisted in the cache file; a cache-restored boot during an etcd outage resurrected the rejection for a key deleted long ago. The state-map removal now runs on both delete branches, and the not-present branch syncs status and flushes the cache whenever it actually removed something. Pinned by a test asserting source_hash returns to its pre-put value the moment the key is deleted. Also from the audit: merge_snapshot and snapshot_has exhaustively destructure AisixSnapshot so adding a resource kind without wiring these paths becomes a compile error instead of a silently unpinnable kind, and the snapshot_cache_path docs state that a bare key (YAML null) reads as omitted. --- crates/aisix-core/src/config.rs | 3 +- crates/aisix-etcd/src/supervisor.rs | 135 +++++++++++++++++++++------- 2 files changed, 105 insertions(+), 33 deletions(-) diff --git a/crates/aisix-core/src/config.rs b/crates/aisix-core/src/config.rs index 1258b4a9..a75ee519 100644 --- a/crates/aisix-core/src/config.rs +++ b/crates/aisix-core/src/config.rs @@ -264,7 +264,8 @@ pub struct ManagedConfig { /// 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. + /// 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, diff --git a/crates/aisix-etcd/src/supervisor.rs b/crates/aisix-etcd/src/supervisor.rs index 9a84b5ae..11e12ff7 100644 --- a/crates/aisix-etcd/src/supervisor.rs +++ b/crates/aisix-etcd/src/supervisor.rs @@ -767,13 +767,21 @@ impl Supervisor

{ // 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; } @@ -827,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); @@ -1140,65 +1147,95 @@ fn clone_snapshot(src: &AisixSnapshot) -> AisixSnapshot { } /// Insert every entry of `src` into `dst` (replacing same-id entries). -/// Must cover every ResourceTable on [`AisixSnapshot`] — a missing kind -/// here means entries of that kind silently drop on the floor when a -/// watch put merges or a last-known-good row is re-injected on resync. +/// 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) { - for e in src.models.entries() { + 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 src.apikeys.entries() { + for e in apikeys.entries() { dst.apikeys.insert(clone_entry(&e)); } - for e in src.provider_keys.entries() { + for e in provider_keys.entries() { dst.provider_keys.insert(clone_entry(&e)); } - for e in src.guardrails.entries() { + for e in guardrails.entries() { dst.guardrails.insert(clone_entry(&e)); } - for e in src.guardrail_attachments.entries() { + for e in guardrail_attachments.entries() { dst.guardrail_attachments.insert(clone_entry(&e)); } - for e in src.cache_policies.entries() { + for e in cache_policies.entries() { dst.cache_policies.insert(clone_entry(&e)); } - for e in src.observability_exporters.entries() { + for e in observability_exporters.entries() { dst.observability_exporters.insert(clone_entry(&e)); } - for e in src.rate_limit_policies.entries() { + for e in rate_limit_policies.entries() { dst.rate_limit_policies.insert(clone_entry(&e)); } - for e in src.mcp_servers.entries() { + for e in mcp_servers.entries() { dst.mcp_servers.insert(clone_entry(&e)); } - for e in src.mcp_policies.entries() { + for e in mcp_policies.entries() { dst.mcp_policies.insert(clone_entry(&e)); } - for e in src.a2a_agents.entries() { + for e in a2a_agents.entries() { dst.a2a_agents.insert(clone_entry(&e)); } - for e in src.oidc_providers.entries() { + for e in oidc_providers.entries() { dst.oidc_providers.insert(clone_entry(&e)); } } -/// Whether the snapshot holds an entry for `(kind, id)`. Mirrors the -/// kind dispatch of [`Supervisor::apply_delete`]'s presence probe; an -/// unknown kind reads as absent. +/// 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" => snap.models.get_by_id(id).is_some(), - "api_keys" => snap.apikeys.get_by_id(id).is_some(), - "provider_keys" => snap.provider_keys.get_by_id(id).is_some(), - "guardrails" => snap.guardrails.get_by_id(id).is_some(), - "guardrail_attachments" => snap.guardrail_attachments.get_by_id(id).is_some(), - "cache_policies" => snap.cache_policies.get_by_id(id).is_some(), - "observability_exporters" => snap.observability_exporters.get_by_id(id).is_some(), - "rate_limit_policies" => snap.rate_limit_policies.get_by_id(id).is_some(), - "mcp_servers" => snap.mcp_servers.get_by_id(id).is_some(), - "mcp_policies" => snap.mcp_policies.get_by_id(id).is_some(), - "a2a_agents" => snap.a2a_agents.get_by_id(id).is_some(), - "oidc_providers" => snap.oidc_providers.get_by_id(id).is_some(), + "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, } } @@ -2058,6 +2095,40 @@ mod tests { } } + #[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