From e0e51c7da69b4054fd0c35babc04eaf93b27eec9 Mon Sep 17 00:00:00 2001 From: Jiahao <58540575+jiahao6635@users.noreply.github.com> Date: Thu, 24 Sep 2026 23:26:58 +0800 Subject: [PATCH] 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 --- .../gateway-core/src/routing/snapshot.rs | 47 +++++- .../crates/gateway-core/src/runtime/mod.rs | 140 ++++++++++++++--- .../crates/gateway-core/tests/runtime/mod.rs | 144 ++++++++++++++++-- .../composables/useAccountBatchEditor.ts | 2 +- .../accounts/composables/useAccountEditor.ts | 2 +- 5 files changed, 303 insertions(+), 32 deletions(-) diff --git a/backend/crates/gateway-core/src/routing/snapshot.rs b/backend/crates/gateway-core/src/routing/snapshot.rs index 85b5b2f01..8562721aa 100644 --- a/backend/crates/gateway-core/src/routing/snapshot.rs +++ b/backend/crates/gateway-core/src/routing/snapshot.rs @@ -349,6 +349,21 @@ impl RuntimeSnapshotCompiler { /// 读取一个 revision，并为已注册 Provider 查询实时模型目录。 pub async fn compile(&self) -> Result { + self.compile_inner(None).await + } + + /// 配置提交复用同一扩展集合的已发布目录，目录留待对账刷新。 + pub(crate) async fn compile_with_cached_catalog( + &self, + previous: &RuntimeSnapshot, + ) -> Result { + self.compile_inner(Some(previous)).await + } + + async fn compile_inner( + &self, + previous: Option<&RuntimeSnapshot>, + ) -> Result { // 同一配置 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, + previous: Option<&RuntimeSnapshot>, ) -> Result { // 只有成功取得的完整目录能证明模型缺项；发现型目录和查询失败均交由上游验证。 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; }; diff --git a/backend/crates/gateway-core/src/runtime/mod.rs b/backend/crates/gateway-core/src/runtime/mod.rs index fb2ef9776..4483999fb 100644 --- a/backend/crates/gateway-core/src/runtime/mod.rs +++ b/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, +} + +struct CommittedRefreshGuard { + priority: Arc>, +} + +impl CommittedRefreshGuard { + fn begin(priority: Arc>) -> 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>, + cancellation: CancellationToken, +} + +impl BackgroundRefreshGuard { + fn begin(priority: Arc>) -> Option { + 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( + cancellation: Option<&CancellationToken>, + operation: impl Future, +) -> Result { + 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, refresh_lock: Arc>, + refresh_priority: Arc>, } 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 { - // 所有入口共享从读取事实到发布或暂停的完整临界区；只锁最后的替换会让 - // 旧编译覆盖新授权，或让晚到的失败暂停新快照。请求读取不等待此锁。 + // 提交先登记并中断慢后台读取；发布/暂停仍在完整临界区内串行执行。 + 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 { + 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 { 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(lock: &RwLock) -> std::sync::RwLockWriteGuard<'_, T> { lock.write() .unwrap_or_else(std::sync::PoisonError::into_inner) } + +fn lock_unpoisoned(lock: &SyncMutex) -> std::sync::MutexGuard<'_, T> { + lock.lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) +} diff --git a/backend/crates/gateway-core/tests/runtime/mod.rs b/backend/crates/gateway-core/tests/runtime/mod.rs index fd122d917..98c56137f 100644 --- a/backend/crates/gateway-core/tests/runtime/mod.rs +++ b/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(¤t)); + 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>>, + models: Mutex>, } 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()) }) } } diff --git a/frontend/src/views/accounts/composables/useAccountBatchEditor.ts b/frontend/src/views/accounts/composables/useAccountBatchEditor.ts index 1c0bc9164..cfcbf3c18 100644 --- a/frontend/src/views/accounts/composables/useAccountBatchEditor.ts +++ b/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() }) } diff --git a/frontend/src/views/accounts/composables/useAccountEditor.ts b/frontend/src/views/accounts/composables/useAccountEditor.ts index af4c8260c..7e208e021 100644 --- a/frontend/src/views/accounts/composables/useAccountEditor.ts +++ b/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()]) }) }