From a3e24198cdf9683e565619ccfa007c9ee19bb34a Mon Sep 17 00:00:00 2001 From: Mason Ma <89764941+MasonSama404@users.noreply.github.com> Date: Fri, 30 Oct 2026 22:45:28 +0800 Subject: [PATCH] fix(openai): back off failed model catalog ETag refreshes (#561) Return catalog refresh failures to the host supervisor so queued ETags use the existing restart backoff. Preserve failure causes and report recovery only after a successful refresh. Co-authored-by: Codex --- .../providers/openai/src/provider/workers.rs | 13 +- .../crates/providers/openai/tests/admin.rs | 17 +- .../providers/openai/tests/provider/mod.rs | 1 + .../openai/tests/provider/workers.rs | 149 ++++++++++++++++++ 4 files changed, 169 insertions(+), 11 deletions(-) create mode 100644 backend/crates/providers/openai/tests/provider/workers.rs diff --git a/backend/crates/providers/openai/src/provider/workers.rs b/backend/crates/providers/openai/src/provider/workers.rs index 99cc94850..4a27179a6 100644 --- a/backend/crates/providers/openai/src/provider/workers.rs +++ b/backend/crates/providers/openai/src/provider/workers.rs @@ -308,14 +308,11 @@ impl DaemonTask for OpenAiCatalogEtagTask { () = cancellation.cancelled() => return Ok(()), () = self.catalog.wait_for_etag_refresh() => {}, }; - if let Err(error) = self.catalog.refresh().await { - tracing::warn!( - error = %error, - "OpenAI model catalog ETag refresh failed" - ); - } else { - context.succeeded(); - } + // 失败的 ETag 已重新入队，交回 Host 退避后再领取，避免立即重试 + self.catalog.refresh().await.map_err(|error| { + WorkerTaskError::safe("OpenAI 模型目录 ETag 刷新失败").with_source(error) + })?; + context.succeeded(); } }) } diff --git a/backend/crates/providers/openai/tests/admin.rs b/backend/crates/providers/openai/tests/admin.rs index 1bc30743a..8dbad64a6 100644 --- a/backend/crates/providers/openai/tests/admin.rs +++ b/backend/crates/providers/openai/tests/admin.rs @@ -1826,7 +1826,10 @@ fn reset_credit_command(account_id: ProviderAccountId) -> ConsumeProviderResetCr } } -fn initialized_provider_request(operation: Operation, account_id: &str) -> ProviderRequest { +pub(crate) fn initialized_provider_request( + operation: Operation, + account_id: &str, +) -> ProviderRequest { let provider = ProviderKind::new("openai").expect("provider"); let upstream_model = UpstreamModelId::new("gpt-5.4").expect("upstream model"); let public_model = PublicModelId::new(upstream_model.as_str()).expect("public model"); @@ -1856,7 +1859,7 @@ fn initialized_provider_request(operation: Operation, account_id: &str) -> Provi ProviderRequest::new(operation, plan.candidates()[0].clone()) } -fn initialized_attempt_context(request_id: &str, account_id: &str) -> AttemptContext { +pub(crate) fn initialized_attempt_context(request_id: &str, account_id: &str) -> AttemptContext { AttemptContext::new( RequestAttemptContext::new( ModelRequestId::new(request_id).expect("request id"), @@ -1887,6 +1890,15 @@ pub(crate) async fn initialized_test_provider( accounts: Arc, base_url: String, ) -> Arc { + initialized_test_bundle(accounts, base_url) + .await + .core_provider() +} + +pub(crate) async fn initialized_test_bundle( + accounts: Arc, + base_url: String, +) -> provider_openai::ProviderBundle { let mut config = valid_config(); config.config.api.base_url = base_url; provider_openai::initialize( @@ -1895,7 +1907,6 @@ pub(crate) async fn initialized_test_provider( ) .await .expect("initialized provider") - .core_provider() } fn provider_ports() -> ProviderStorePorts { diff --git a/backend/crates/providers/openai/tests/provider/mod.rs b/backend/crates/providers/openai/tests/provider/mod.rs index f87e8cfa8..c782d75ca 100644 --- a/backend/crates/providers/openai/tests/provider/mod.rs +++ b/backend/crates/providers/openai/tests/provider/mod.rs @@ -3,5 +3,6 @@ mod contract; mod failure; mod live; +mod workers; pub(crate) use contract::assert_local_connection_capacity_is_not_an_upstream_failure; diff --git a/backend/crates/providers/openai/tests/provider/workers.rs b/backend/crates/providers/openai/tests/provider/workers.rs new file mode 100644 index 000000000..e85bd0e92 --- /dev/null +++ b/backend/crates/providers/openai/tests/provider/workers.rs @@ -0,0 +1,149 @@ +//! 验证 OpenAI 目录 ETag worker 的失败交接、重启恢复与取消 + +use std::sync::Arc; +use std::time::Duration; + +use chrono::Utc; +use futures::StreamExt; +use gateway_core::lifecycle::CancellationToken; +use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload}; +use gateway_core::task::{WorkerContribution, WorkerDaemonContext, WorkerRunnable}; +use provider_openai::credential::ImportCodexOAuthCredential; +use serde_json::{Map, json}; +use tokio::time::timeout; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +use crate::admin::{ + initialized_attempt_context, initialized_provider_request, initialized_test_bundle, +}; +use crate::support::{MemoryAccountStore, profile, secret}; + +#[tokio::test] +async fn catalog_etag_failure_returns_to_supervisor_and_recovers_pending_refresh() { + let account_id = "acct_catalog_etag_worker"; + let store = Arc::new(MemoryAccountStore::default()); + store + .seed_oauth_credential(ImportCodexOAuthCredential { + account_id: account_id.to_owned(), + name: account_id.to_owned(), + secret: secret("at-catalog-etag-worker"), + verified_account: profile("chatgpt-catalog-etag-worker"), + next_refresh_at: Some(Utc::now() + chrono::Duration::minutes(30)), + enabled: true, + }) + .await; + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .insert_header("x-models-etag", "models-worker-v2") + .set_body_string(concat!( + "event: response.completed\n", + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_etag\",\"model\":\"gpt-5.4\",\"status\":\"completed\",\"output\":[]}}\n\n" + )), + ) + .expect(1) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with(ResponseTemplate::new(503)) + .with_priority(1) + .up_to_n_times(2) + .expect(2) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/codex/models")) + .respond_with(ResponseTemplate::new(200).set_body_raw( + include_bytes!("../transport/fixtures/official_models_snapshot.json"), + "application/json", + )) + .with_priority(2) + .expect(1) + .mount(&server) + .await; + let mut bundle = initialized_test_bundle(store, server.uri()).await; + let provider = bundle.core_provider(); + let operation = Operation::Generate(GenerateRequest::from_protocol_payload( + ProtocolPayload::json_object( + "openai", + json!({"model": "gpt-5.4", "input": "catalog refresh"}) + .as_object() + .unwrap() + .clone(), + ) + .unwrap() + .with_context(Map::from_iter([("use_websocket".to_owned(), json!(false))])), + )); + let mut stream = provider + .execute( + initialized_provider_request(operation, account_id), + initialized_attempt_context("req_catalog_etag_worker", account_id), + ) + .await + .expect("provider stream"); + while let Some(event) = stream.next().await { + event.expect("response carrying catalog ETag"); + } + let worker = bundle + .take_worker_contributions() + .into_iter() + .find_map(|contribution| match contribution { + WorkerContribution::Registration(registration) + if registration.id.owner() == "openai-model-etag" => + { + Some(registration) + } + _ => None, + }) + .expect("catalog ETag worker"); + let WorkerRunnable::Daemon { task, .. } = worker.runnable else { + panic!("catalog ETag worker must be a daemon"); + }; + + for expected_requests in 1..=2 { + let context = WorkerDaemonContext::new(CancellationToken::new()); + let result = timeout(Duration::from_secs(2), task.run(context.clone())).await; + let requests = server.received_requests().await.unwrap(); + let paths: Vec<_> = requests.iter().map(|request| request.url.path()).collect(); + assert!( + result.is_ok(), + "failed refresh must return control to the supervisor: {paths:?}, completed={}", + context.completed_cycles() + ); + let error = result.unwrap().expect_err("catalog unavailable"); + assert!( + error.error_details().is_some(), + "preserve the failure cause" + ); + assert_eq!(context.completed_cycles(), 0); + assert_eq!( + requests + .iter() + .filter(|request| request.url.path() == "/codex/models") + .count(), + expected_requests, + "each run must perform only one failed refresh" + ); + } + + let context = WorkerDaemonContext::new(CancellationToken::new()); + let run = task.run(context.clone()); + tokio::pin!(run); + tokio::select! { + result = &mut run => panic!("daemon exited before recovery: {result:?}"), + result = timeout(Duration::from_secs(2), context.progress_since(0)) => { + result.expect("pending ETag must survive failed runs"); + } + } + assert_eq!(context.completed_cycles(), 1); + context.cancellation().cancel(); + timeout(Duration::from_secs(2), run) + .await + .expect("idle worker must stop on cancellation") + .expect("cancelled worker"); +}