From 26de671226322bad6b79bb8897907d39f184111f Mon Sep 17 00:00:00 2001 From: Ke Wang Date: Tue, 3 Nov 2026 11:35:22 +0800 Subject: [PATCH] fix(runtime): reload full snapshot on config revision races (#584) * fix(runtime): reload raced snapshot facts before suspending publication Co-authored-by: Codex * fix(runtime): retry extension preparation when its revision becomes stale Co-authored-by: Codex --------- Co-authored-by: Codex --- .../gateway-core/src/routing/snapshot.rs | 33 ++++-- .../gateway-core/tests/routing/snapshot.rs | 108 +++++++++++++++++- .../crates/gateway-core/tests/runtime/mod.rs | 33 ++++++ 3 files changed, 165 insertions(+), 9 deletions(-) diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index b41b7ef67..0703ee48f 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -23,7 +23,7 @@ use super::{ RuntimeAccountDirectory, UpstreamModelId, }; -const MAXIMUM_CATALOG_STABILITY_ATTEMPTS: usize = 4; +const MAXIMUM_SNAPSHOT_STABILITY_ATTEMPTS: usize = 4; type ModelCatalogAccounts = BTreeMap>>; @@ -328,14 +328,17 @@ impl RuntimeSnapshotCompiler { ConfigRevision, crate::routing::extensions::ExtensionSetReference, )> = None; - for _ in 0..MAXIMUM_CATALOG_STABILITY_ATTEMPTS { + let mut instability = RuntimeSnapshotCompileError::CatalogChanged; + for _ in 0..MAXIMUM_SNAPSHOT_STABILITY_ATTEMPTS { let facts = self .store .load_snapshot_facts() .await .map_err(|_| RuntimeSnapshotCompileError::StoreUnavailable)?; if facts.config_revision != facts.observed_current_revision { - return Err(RuntimeSnapshotCompileError::RevisionChanged); + // 并发提交使本轮事实过期时重读完整快照，不能混用不同版本的权限 + instability = RuntimeSnapshotCompileError::RevisionChanged; + continue; } let revision = facts.config_revision; let extensions = match &self.extensions { @@ -347,7 +350,7 @@ impl RuntimeSnapshotCompiler { *prepared_revision != revision || !prepared.is_ready() }); if must_prepare { - let prepared = preparer + let prepared = match preparer .prepare_with_request_body_limit( revision, usize::try_from( @@ -356,7 +359,21 @@ impl RuntimeSnapshotCompiler { .unwrap_or(usize::MAX), ) .await - .map_err(|_| RuntimeSnapshotCompileError::ExtensionsUnavailable)?; + { + Ok(prepared) => prepared, + Err(_) => { + // 扩展读取也可能遇到并发提交；只有版本已变才重读事实 + let observed = + self.store.current_config_revision().await.map_err(|_| { + RuntimeSnapshotCompileError::StoreUnavailable + })?; + if observed == revision { + return Err(RuntimeSnapshotCompileError::ExtensionsUnavailable); + } + instability = RuntimeSnapshotCompileError::RevisionChanged; + continue; + } + }; prepared_extensions = Some((revision, prepared)); } prepared_extensions @@ -390,7 +407,8 @@ impl RuntimeSnapshotCompiler { .map_err(|_| RuntimeSnapshotCompileError::StoreUnavailable)? != revision { - return Err(RuntimeSnapshotCompileError::RevisionChanged); + instability = RuntimeSnapshotCompileError::RevisionChanged; + continue; } let snapshot = snapshot .with_provider_catalog_generations(if cached.is_some() && !refresh_catalog { @@ -401,8 +419,9 @@ impl RuntimeSnapshotCompiler { .with_extensions(extensions); return Ok(snapshot); } + instability = RuntimeSnapshotCompileError::CatalogChanged; } - Err(RuntimeSnapshotCompileError::CatalogChanged) + Err(instability) } } diff --git a/backend/crates/gateway-core/tests/routing/snapshot.rs b/backend/crates/gateway-core/tests/routing/snapshot.rs index aab41dc0d..cd59f743d 100644 --- a/backend/crates/gateway-core/tests/routing/snapshot.rs +++ b/backend/crates/gateway-core/tests/routing/snapshot.rs @@ -730,12 +730,116 @@ impl ProviderCatalogPort for PublishingCatalog { #[test] fn compiler_should_reject_revision_changed_during_consistent_read() { - let facts = facts(1, 2); - let compiler = compiler(Arc::new(TestSnapshotStore::new(Ok(facts)))); + let store = Arc::new(SequencedSnapshotStore { + facts: vec![facts(1, 2)], + loads: AtomicUsize::new(0), + current_revision: revision(2), + }); + let compiler = compiler(store.clone()); let error = block_on(compiler.compile()).expect_err("revision drift must fail closed"); assert_eq!(error, RuntimeSnapshotCompileError::RevisionChanged); + assert_eq!(store.loads.load(Ordering::SeqCst), 4); +} + +#[test] +fn compiler_reloads_complete_facts_after_revision_changes_during_read() { + let store = Arc::new(SequencedSnapshotStore { + facts: vec![facts(1, 2), facts(2, 2)], + loads: AtomicUsize::new(0), + current_revision: revision(2), + }); + + let snapshot = block_on(compiler(store.clone()).compile()).expect("latest consistent facts"); + + assert_eq!(snapshot.revision(), revision(2)); + assert_eq!(store.loads.load(Ordering::SeqCst), 2); +} + +#[test] +fn compiler_reloads_revision_changed_after_extension_preparation() { + let drops = Arc::new(AtomicUsize::new(0)); + let extensions = Arc::new(CountingExtensionPreparation { + prepares: AtomicUsize::new(0), + drops: Arc::clone(&drops), + }); + let compiler = RuntimeSnapshotCompiler::new( + Arc::new(SequencedSnapshotStore { + facts: vec![facts(1, 1), facts(2, 2)], + loads: AtomicUsize::new(0), + current_revision: revision(2), + }), + Arc::new(TestCatalog::Empty), + ) + .with_extensions(extensions.clone()); + + let snapshot = block_on(compiler.compile()).expect("latest prepared snapshot"); + + assert_eq!(snapshot.revision(), revision(2)); + assert_eq!(snapshot.extensions().unwrap().id().as_str(), "prepared-2"); + assert_eq!(extensions.prepares.load(Ordering::SeqCst), 2); + assert_eq!(drops.load(Ordering::SeqCst), 1); + drop(snapshot); + assert_eq!(drops.load(Ordering::SeqCst), 2); +} + +#[test] +fn compiler_rechecks_revision_when_extension_preparation_rejects_stale_facts() { + struct RacingPreparation { + store: Arc, + changes: usize, + calls: AtomicUsize, + } + + impl ExtensionPreparationPort for RacingPreparation { + fn prepare( + &self, + _: ConfigRevision, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + let call = self.calls.fetch_add(1, Ordering::SeqCst); + if self.changes == 0 || call < self.changes { + if self.changes != 0 { + let next = u64::try_from(call).unwrap() + 2; + *self.store.facts.lock().unwrap() = Ok(facts(next, next)); + *self.store.current_revision.lock().unwrap() = Ok(revision(next)); + } + return Err(ExtensionPreparationError); + } + Ok(ExtensionSetReference::new( + ExtensionSetId::new("ready".to_owned()).unwrap(), + Arc::new(AliasLease(vec![])), + )) + }) + } + } + + for (changes, expected_calls) in [(0, 1), (1, 2), (4, 4)] { + let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1)))); + let extensions = Arc::new(RacingPreparation { + store: store.clone(), + changes, + calls: AtomicUsize::new(0), + }); + let compiler = RuntimeSnapshotCompiler::new(store, Arc::new(TestCatalog::Empty)) + .with_extensions(extensions.clone()); + + let result = block_on(compiler.compile()); + + match changes { + 0 => assert_eq!( + result.unwrap_err(), + RuntimeSnapshotCompileError::ExtensionsUnavailable + ), + 1 => assert_eq!(result.unwrap().revision(), revision(2)), + _ => assert_eq!( + result.unwrap_err(), + RuntimeSnapshotCompileError::RevisionChanged + ), + } + assert_eq!(extensions.calls.load(Ordering::SeqCst), expected_calls); + } } #[test] diff --git a/backend/crates/gateway-core/tests/runtime/mod.rs b/backend/crates/gateway-core/tests/runtime/mod.rs index 73523ad9f..0a26e7fb2 100644 --- a/backend/crates/gateway-core/tests/runtime/mod.rs +++ b/backend/crates/gateway-core/tests/runtime/mod.rs @@ -326,6 +326,39 @@ fn publisher_should_suspend_but_still_notify_after_committed_refresh_failure() { ); } +#[test] +fn committed_refresh_reloads_raced_facts_without_losing_revoked_scope() { + block_on(async { + let store = Arc::new(TestSnapshotStore::new(Ok(scoped_facts(10, true)))); + let initial = compiler(store.clone()) + .compile() + .await + .expect("initial scope"); + let handle = RuntimeSnapshotHandle::new(initial); + let frozen = handle.acquire().expect("initial request"); + assert!(removed_account_allowed(&frozen)); + let publisher = RuntimeSnapshotPublisher::new( + Arc::new(compiler(store.clone())), + handle.clone(), + Arc::new(TestSnapshotSubscriptions::default()), + ); + let (release, gate) = oneshot::channel(); + *store.next_load.lock().expect("load gate lock") = Some(gate); + let mut committed = publisher.publish_committed(revision(11)); + assert!(futures::poll!(committed.as_mut()).is_pending()); + *store.facts.lock().expect("facts lock") = Ok(scoped_facts(11, false)); + *store.current_revision.lock().expect("revision lock") = Ok(revision(11)); + release.send(Ok(facts(10, 11))).expect("release raced read"); + + committed.await; + + let current = handle.acquire().expect("latest snapshot remains available"); + assert_eq!(current.revision(), revision(11)); + assert!(!removed_account_allowed(¤t)); + assert!(removed_account_allowed(&frozen)); + }); +} + #[test] fn publisher_should_contribute_reconciliation_and_subscription_workers() { let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1))));