fix(runtime): reload full snapshot on config revision races (#584)

* fix(runtime): reload raced snapshot facts before suspending publication Co-authored-by: Codex <noreply@openai.com> * fix(runtime): retry extension preparation when its revision becomes stale Co-authored-by: Codex <noreply@openai.com> --------- Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-11-03 11:38:47 +08:00
parent a92b95c48c
commit 26de671226
3 changed files with 165 additions and 9 deletions
@@ -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<ProviderKind, BTreeMap<String, BTreeSet<ProviderAccountId>>>;

@@ -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)
    }
 }

@@ -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<TestSnapshotStore>,
+        changes: usize,
+        calls: AtomicUsize,
+    }
+
+    impl ExtensionPreparationPort for RacingPreparation {
+        fn prepare(
+            &self,
+            _: ConfigRevision,
+        ) -> BoxFuture<'_, Result<ExtensionSetReference, ExtensionPreparationError>> {
+            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]
@@ -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(&current));
+        assert!(removed_account_allowed(&frozen));
+    });
+}
+
 #[test]
 fn publisher_should_contribute_reconciliation_and_subscription_workers() {
    let store = Arc::new(TestSnapshotStore::new(Ok(facts(1, 1))));