fix(accounts): avoid provider catalog waits when saving (#260)

Publish committed account and permission facts with the current model catalog while the extension set is unchanged. Preempt in-flight background catalog reconciliation for a new configuration commit; periodic reconciliation refreshes the catalog afterward. Confirm account saves before asynchronously reloading lists and groups.

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-core/src/routing/snapshot.rs

@@ -349,6 +349,21 @@ impl RuntimeSnapshotCompiler {
 
     /// 读取一个 revision，并为已注册 Provider 查询实时模型目录。
     pub async fn compile(&self) -> Result<RuntimeSnapshot, RuntimeSnapshotCompileError> {
+        self.compile_inner(None).await
+    }
+
+    /// 配置提交复用同一扩展集合的已发布目录，目录留待对账刷新。
+    pub(crate) async fn compile_with_cached_catalog(
+        &self,
+        previous: &RuntimeSnapshot,
+    ) -> Result<RuntimeSnapshot, RuntimeSnapshotCompileError> {
+        self.compile_inner(Some(previous)).await
+    }
+
+    async fn compile_inner(
+        &self,
+        previous: Option<&RuntimeSnapshot>,
+    ) -> Result<RuntimeSnapshot, RuntimeSnapshotCompileError> {
         // 同一配置 revision 的重试复用扩展候选，保留插件策略的发布一致性。
         let mut prepared_extensions: Option<(
             ConfigRevision,
@@ -388,8 +403,11 @@ impl RuntimeSnapshotCompiler {
             let catalogs = &self.catalogs;
             let catalog_generations = catalogs.catalog_generations();
             let provider_kinds = catalog_generations.keys().cloned().collect();
+            let cached = previous.filter(|previous| {
+                previous.extensions().map(|set| set.id()) == extensions.as_ref().map(|set| set.id())
+            });
             let snapshot =
-                compile_runtime_snapshot(facts, catalogs.as_ref(), provider_kinds).await?;
+                compile_runtime_snapshot(facts, catalogs.as_ref(), provider_kinds, cached).await?;
             let observed_generations = catalogs.catalog_generations();
             if catalog_generations == observed_generations {
                 if extensions.is_some()
@@ -403,7 +421,11 @@ impl RuntimeSnapshotCompiler {
                     return Err(RuntimeSnapshotCompileError::RevisionChanged);
                 }
                 return Ok(snapshot
-                    .with_provider_catalog_generations(observed_generations)
+                    .with_provider_catalog_generations(if cached.is_some() {
+                        BTreeMap::new()
+                    } else {
+                        observed_generations
+                    })
                     .with_extensions(extensions));
             }
         }
@@ -415,11 +437,32 @@ async fn compile_runtime_snapshot(
     facts: SnapshotFacts,
     catalogs: &dyn ProviderCatalogPort,
     provider_kinds: Vec<ProviderKind>,
+    previous: Option<&RuntimeSnapshot>,
 ) -> Result<RuntimeSnapshot, RuntimeSnapshotCompileError> {
     // 只有成功取得的完整目录能证明模型缺项；发现型目录和查询失败均交由上游验证。
     let mut provider_models = Vec::new();
     let mut exhaustive_provider_catalogs = BTreeSet::new();
     for provider in &provider_kinds {
+        if let Some(previous) = previous {
+            if previous.exhaustive_provider_catalogs.contains(provider) {
+                exhaustive_provider_catalogs.insert(provider.clone());
+            }
+            if let Some(models) = previous.provider_models.get(provider) {
+                provider_models.extend(models.iter().map(|(model, capabilities)| {
+                    let compiled =
+                        ProviderModel::new(provider.clone(), model.clone(), capabilities.clone());
+                    match previous
+                        .provider_model_presentations
+                        .get(provider)
+                        .and_then(|presentations| presentations.get(model))
+                    {
+                        Some(presentation) => compiled.with_presentation(presentation.clone()),
+                        None => compiled,
+                    }
+                }));
+            }
+            continue;
+        }
         let Ok(models) = catalogs.query_model_capabilities(provider).await else {
             continue;
         };

backend/crates/gateway-core/src/runtime/mod.rs

@@ -3,8 +3,9 @@
 pub mod extensions;
 
 use std::collections::BTreeMap;
+use std::future::Future;
 use std::pin::Pin;
-use std::sync::{Arc, RwLock};
+use std::sync::{Arc, Mutex as SyncMutex, RwLock};
 use std::time::Duration;
 
 use futures::future::BoxFuture;
@@ -138,6 +139,81 @@ pub trait SnapshotControl: Send + Sync {
     fn publish_committed(&self, committed_revision: ConfigRevision) -> BoxFuture<'_, ()>;
 }
 
+#[derive(Default)]
+struct RefreshPriorityState {
+    pending_commits: usize,
+    active_background: Option<CancellationToken>,
+}
+
+struct CommittedRefreshGuard {
+    priority: Arc<SyncMutex<RefreshPriorityState>>,
+}
+
+impl CommittedRefreshGuard {
+    fn begin(priority: Arc<SyncMutex<RefreshPriorityState>>) -> Self {
+        let active = {
+            let mut state = lock_unpoisoned(&priority);
+            state.pending_commits += 1;
+            state.active_background.clone()
+        };
+        // 先登记待发布提交，再取消正在等待目录的后台刷新，避免后者抢先重入。
+        if let Some(active) = active {
+            active.cancel();
+        }
+        Self { priority }
+    }
+}
+
+impl Drop for CommittedRefreshGuard {
+    fn drop(&mut self) {
+        lock_unpoisoned(&self.priority).pending_commits -= 1;
+    }
+}
+
+struct BackgroundRefreshGuard {
+    priority: Arc<SyncMutex<RefreshPriorityState>>,
+    cancellation: CancellationToken,
+}
+
+impl BackgroundRefreshGuard {
+    fn begin(priority: Arc<SyncMutex<RefreshPriorityState>>) -> Option<Self> {
+        let mut state = lock_unpoisoned(&priority);
+        if state.pending_commits != 0 {
+            return None;
+        }
+        let cancellation = CancellationToken::new();
+        state.active_background = Some(cancellation.clone());
+        drop(state);
+        Some(Self {
+            priority,
+            cancellation,
+        })
+    }
+}
+
+impl Drop for BackgroundRefreshGuard {
+    fn drop(&mut self) {
+        lock_unpoisoned(&self.priority).active_background = None;
+    }
+}
+
+// 仅中断未完成的后台读取或编译；配置提交负责发布最新事实并发送通知。
+async fn preempt_for_committed_refresh<T>(
+    cancellation: Option<&CancellationToken>,
+    operation: impl Future<Output = T>,
+) -> Result<T, RuntimeSnapshotCompileError> {
+    let Some(cancellation) = cancellation else {
+        return Ok(operation.await);
+    };
+    let cancelled = cancellation.cancelled().fuse();
+    let operation = operation.fuse();
+    pin_mut!(cancelled, operation);
+    select_biased! {
+        _ = cancelled => Err(RuntimeSnapshotCompileError::RevisionChanged),
+        result = operation => Ok(result),
+    }
+}
+
 /// 配置提交后的本进程快照发布与跨进程失效通知。
 #[derive(Clone)]
 pub struct RuntimeSnapshotPublisher {
@@ -145,11 +221,13 @@ pub struct RuntimeSnapshotPublisher {
     snapshots: RuntimeSnapshotHandle,
     subscriptions: Arc<dyn SnapshotSubscriptionPort>,
     refresh_lock: Arc<Mutex<()>>,
+    refresh_priority: Arc<SyncMutex<RefreshPriorityState>>,
 }
 
 enum RefreshMode {
     Required,
     Reconcile,
+    Committed,
 }
 
 impl RuntimeSnapshotPublisher {
@@ -164,6 +242,7 @@ impl RuntimeSnapshotPublisher {
             snapshots,
             subscriptions,
             refresh_lock: Arc::new(Mutex::new(())),
+            refresh_priority: Arc::new(SyncMutex::new(RefreshPriorityState::default())),
         }
     }
 
@@ -176,26 +255,41 @@ impl RuntimeSnapshotPublisher {
         &self,
         mode: RefreshMode,
     ) -> Result<ConfigRevision, RuntimeSnapshotCompileError> {
-        // 所有入口共享从读取事实到发布或暂停的完整临界区；只锁最后的替换会让
-        // 旧编译覆盖新授权，或让晚到的失败暂停新快照。请求读取不等待此锁。
+        // 提交先登记并中断慢后台读取；发布/暂停仍在完整临界区内串行执行。
+        let _committed = matches!(mode, RefreshMode::Committed)
+            .then(|| CommittedRefreshGuard::begin(Arc::clone(&self.refresh_priority)));
         let _refresh = self.refresh_lock.lock().await;
-        self.refresh_locked(mode).await
+        self.refresh_locked_with_priority(mode).await
+    }
+
+    async fn refresh_locked_with_priority(
+        &self,
+        mode: RefreshMode,
+    ) -> Result<ConfigRevision, RuntimeSnapshotCompileError> {
+        if matches!(mode, RefreshMode::Committed) {
+            return self.refresh_locked(mode, None).await;
+        }
+        let background = BackgroundRefreshGuard::begin(Arc::clone(&self.refresh_priority))
+            .ok_or(RuntimeSnapshotCompileError::RevisionChanged)?;
+        self.refresh_locked(mode, Some(&background.cancellation))
+            .await
     }
 
     async fn refresh_locked(
         &self,
         mode: RefreshMode,
+        cancellation: Option<&CancellationToken>,
     ) -> Result<ConfigRevision, RuntimeSnapshotCompileError> {
         let configuration_changed = if matches!(mode, RefreshMode::Reconcile) {
-            let persisted_revision = self
-                .compiler
-                .store()
-                .current_config_revision()
-                .await
-                .map_err(|_| {
-                    self.snapshots.suspend();
-                    RuntimeSnapshotCompileError::StoreUnavailable
-                })?;
+            let persisted_revision = preempt_for_committed_refresh(
+                cancellation,
+                self.compiler.store().current_config_revision(),
+            )
+            .await?
+            .map_err(|_| {
+                self.snapshots.suspend();
+                RuntimeSnapshotCompileError::StoreUnavailable
+            })?;
             let changed = runtime_revision_needs_refresh(
                 self.published_revision().map(ConfigRevision::get),
                 persisted_revision.get(),
@@ -213,11 +307,18 @@ impl RuntimeSnapshotPublisher {
         } else {
             true
         };
-        let snapshot = match self.compiler.compile().await {
+        let compiled = if matches!(mode, RefreshMode::Committed) {
+            match self.snapshots.snapshot_for_diagnostics() {
+                Some(previous) => self.compiler.compile_with_cached_catalog(&previous).await,
+                None => self.compiler.compile().await,
+            }
+        } else {
+            preempt_for_committed_refresh(cancellation, self.compiler.compile()).await?
+        };
+        let snapshot = match compiled {
             Ok(snapshot) => snapshot,
             Err(error) => {
-                // 仅目录代次变化时保留旧快照，下一周期继续对账；配置缺失、变化
-                // 或持久 revision 回退均须 fail closed，不按 revision 大小丢弃刷新。
+                // 目录对账失败可继续服务旧快照；配置或权限变化无法确认时须暂停。
                 if configuration_changed {
                     self.snapshots.suspend();
                 }
@@ -244,7 +345,7 @@ impl RuntimeSnapshotPublisher {
 
     /// 数据库提交不能被目录或通知基础设施的暂时故障伪装成回滚。
     async fn publish_committed_inner(&self, committed_revision: ConfigRevision) {
-        if let Err(error) = self.refresh().await {
+        if let Err(error) = self.refresh_with_mode(RefreshMode::Committed).await {
             tracing::warn!(?error, "committed runtime snapshot refresh failed");
         }
         self.notify_committed_revision(committed_revision).await;
@@ -407,3 +508,8 @@ fn write_unpoisoned<T>(lock: &RwLock<T>) -> std::sync::RwLockWriteGuard<'_, T> {
     lock.write()
         .unwrap_or_else(std::sync::PoisonError::into_inner)
 }
+
+fn lock_unpoisoned<T>(lock: &SyncMutex<T>) -> std::sync::MutexGuard<'_, T> {
+    lock.lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}

backend/crates/gateway-core/tests/runtime/mod.rs

@@ -18,8 +18,9 @@ use gateway_core::routing::snapshot::{
     SnapshotStoreError, SnapshotStorePort,
 };
 use gateway_core::routing::{
-    AccountGroupId, ConfigRevision, ProviderCatalogGeneration, ProviderCatalogPort,
-    ProviderCatalogUnavailable, ProviderKind, ProviderModelCapabilities, RuntimeSnapshot,
+    AccountGroupId, ConfigRevision, ModelCapabilities, ProviderCatalogGeneration,
+    ProviderCatalogPort, ProviderCatalogUnavailable, ProviderKind, ProviderModelCapabilities,
+    RuntimeSnapshot, UpstreamModelId,
 };
 use gateway_core::runtime::{
     RuntimeSnapshotHandle, RuntimeSnapshotPublisher, SnapshotControl, SnapshotRevisionStream,
@@ -162,6 +163,119 @@ fn publisher_should_refresh_locally_and_notify_committed_revision() {
     );
 }
 
+#[test]
+fn committed_account_change_should_publish_without_waiting_for_provider_catalog() {
+    block_on(async {
+        let store = Arc::new(TestSnapshotStore::new(Ok(scoped_facts(9, true))));
+        let catalog = Arc::new(TestCatalog::default());
+        catalog
+            .models
+            .lock()
+            .expect("models lock")
+            .push(ProviderModelCapabilities::new(
+                UpstreamModelId::new("listed-model").expect("model"),
+                ModelCapabilities::new(Default::default(), None),
+            ));
+        let compiler = Arc::new(catalog_compiler(store.clone(), catalog.clone()));
+        let initial = compiler.compile().await.expect("initial snapshot");
+        let provider = ProviderKind::new("audit").expect("provider");
+        let initial_models = initial.public_models_for_provider(&provider);
+        let handle = RuntimeSnapshotHandle::new(initial);
+        let publisher = RuntimeSnapshotPublisher::new(
+            compiler,
+            handle.clone(),
+            Arc::new(TestSnapshotSubscriptions::default()),
+        );
+        *store.facts.lock().expect("facts lock") = Ok(scoped_facts(10, false));
+        *store.current_revision.lock().expect("revision lock") = Ok(revision(10));
+        let (release, gate) = oneshot::channel();
+        *catalog.next_query.lock().expect("catalog gate lock") = Some(gate);
+
+        publisher.publish_committed(revision(10)).await;
+        let committed = handle.acquire().expect("committed snapshot");
+        assert_eq!(committed.revision(), revision(10));
+        assert!(!removed_account_allowed(&committed));
+        assert_eq!(
+            committed.public_models_for_provider(&provider),
+            initial_models
+        );
+        assert_eq!(catalog.queries.load(Ordering::SeqCst), 1);
+        assert_ne!(
+            committed.provider_catalog_generations(),
+            &catalog.catalog_generations(),
+        );
+
+        let mut reconcile = Box::pin(publisher.refresh());
+        assert!(futures::poll!(reconcile.as_mut()).is_pending());
+        release.send(()).expect("release catalog query");
+        reconcile.await.expect("catalog reconciliation");
+        assert_eq!(
+            handle
+                .acquire()
+                .expect("reconciled snapshot")
+                .provider_catalog_generations(),
+            &catalog.catalog_generations(),
+        );
+    });
+}
+
+#[test]
+fn committed_account_change_should_preempt_inflight_catalog_reconciliation() {
+    block_on(async {
+        let store = Arc::new(TestSnapshotStore::new(Ok(scoped_facts(9, true))));
+        let catalog = Arc::new(TestCatalog::default());
+        let compiler = Arc::new(catalog_compiler(store.clone(), catalog.clone()));
+        let initial = compiler.compile().await.expect("initial snapshot");
+        let handle = RuntimeSnapshotHandle::new(initial);
+        let publisher = RuntimeSnapshotPublisher::new(
+            compiler,
+            handle.clone(),
+            Arc::new(TestSnapshotSubscriptions::default()),
+        );
+
+        // 首次提交复用目录，随后对账停在慢 Provider 查询；下一次提交仍须及时撤权。
+        *store.facts.lock().expect("facts lock") = Ok(scoped_facts(10, true));
+        *store.current_revision.lock().expect("revision lock") = Ok(revision(10));
+        publisher.publish_committed(revision(10)).await;
+        let (release, gate) = oneshot::channel();
+        *catalog.next_query.lock().expect("catalog gate lock") = Some(gate);
+        let (task, context) = reconciliation_task(&publisher);
+        let mut reconcile = task.run_cycle(context.clone());
+        assert!(futures::poll!(reconcile.as_mut()).is_pending());
+        assert_eq!(catalog.queries.load(Ordering::SeqCst), 2);
+
+        *store.facts.lock().expect("facts lock") = Ok(scoped_facts(11, false));
+        *store.current_revision.lock().expect("revision lock") = Ok(revision(11));
+        let mut committed = publisher.publish_committed(revision(11));
+        assert!(futures::poll!(committed.as_mut()).is_pending());
+        assert!(matches!(
+            futures::poll!(reconcile.as_mut()),
+            std::task::Poll::Ready(Err(_)),
+        ));
+        assert_eq!(
+            futures::poll!(committed.as_mut()),
+            std::task::Poll::Ready(()),
+        );
+        assert!(release.send(()).is_err());
+        let current = handle.acquire().expect("committed snapshot");
+        assert_eq!(current.revision(), revision(11));
+        assert!(!removed_account_allowed(&current));
+        assert_ne!(
+            current.provider_catalog_generations(),
+            &catalog.catalog_generations(),
+        );
+
+        task.run_cycle(context).await.expect("next reconciliation");
+        assert_eq!(
+            handle
+                .acquire()
+                .expect("reconciled snapshot")
+                .provider_catalog_generations(),
+            &catalog.catalog_generations(),
+        );
+    });
+}
+
 #[test]
 fn publisher_should_suspend_but_still_notify_after_committed_refresh_failure() {
     let store = Arc::new(TestSnapshotStore::new(Err(
@@ -323,15 +437,22 @@ fn reconciliation_revision_failure_should_not_suspend_later_committed_refresh()
         assert!(futures::poll!(reconcile.as_mut()).is_pending());
         let mut committed = publisher.publish_committed(revision(11));
         assert!(futures::poll!(committed.as_mut()).is_pending());
-        assert_eq!(store.loads.load(Ordering::SeqCst), 0);
-
-        release
-            .send(Err(SnapshotStoreError::unavailable()))
-            .expect("release failed revision read");
-        assert!(reconcile.await.is_err());
-        assert!(handle.acquire().is_err());
-        committed.await;
+        assert!(matches!(
+            futures::poll!(reconcile.as_mut()),
+            std::task::Poll::Ready(Err(_)),
+        ));
+        assert_eq!(
+            futures::poll!(committed.as_mut()),
+            std::task::Poll::Ready(()),
+        );
+        assert!(
+            release
+                .send(Err(SnapshotStoreError::unavailable()))
+                .is_err()
+        );
+        assert_eq!(store.loads.load(Ordering::SeqCst), 1);
         assert_eq!(handle.revision(), Some(revision(11)));
+        assert!(handle.acquire().is_ok());
     });
 }
 
@@ -547,6 +668,7 @@ struct TestCatalog {
     generation: AtomicU64,
     queries: AtomicUsize,
     next_query: Mutex<Option<oneshot::Receiver<()>>>,
+    models: Mutex<Vec<ProviderModelCapabilities>>,
 }
 
 impl ProviderCatalogPort for TestCatalog {
@@ -567,7 +689,7 @@ impl ProviderCatalogPort for TestCatalog {
             if let Some(gate) = gate {
                 gate.await.expect("release catalog query");
             }
-            Ok(Vec::new())
+            Ok(self.models.lock().expect("models lock").clone())
         })
     }
 }

frontend/src/views/accounts/composables/useAccountBatchEditor.ts

@@ -94,8 +94,8 @@ export function useAccountBatchEditor(options: {
       })
       showBatchEditModal.value = false
       options.selectedIds.value = new Set()
-      await Promise.all([options.reloadAccounts(), options.reloadGroups()])
       toast.success(`已更新 ${accountIds.length} 个账号`)
+      void Promise.allSettled([options.reloadAccounts(), options.reloadGroups()])
     }, { onError: () => void options.reloadAccounts() })
   }
 

frontend/src/views/accounts/composables/useAccountEditor.ts

@@ -134,8 +134,8 @@ export function useAccountEditor(options: {
           : undefined,
       })
       showEditModal.value = false
-      await Promise.all([options.reloadAccounts(), options.reloadGroups()])
       toast.success('账号已更新')
+      void Promise.allSettled([options.reloadAccounts(), options.reloadGroups()])
     })
   }