From 193e992a474669bae79a406fc3968cac69f6c738 Mon Sep 17 00:00:00 2001 From: zyycn Date: Mon, 15 Jun 2026 09:53:51 +0800 Subject: [PATCH] codex: persist failed request attempts Co-authored-by: Codex --- docs/OPENAI_CHAIN_PARITY_REAUDIT.md | 4 +- src/codex/serving/chat.rs | 16 +++++++ src/codex/serving/dispatch/mod.rs | 21 ++++++++- src/codex/serving/responses.rs | 52 +++++++++++++++++++++ tests/codex_serving/upstream_fallback.rs | 58 ++++++++++++++++++++++++ 5 files changed, 148 insertions(+), 3 deletions(-) diff --git a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md index 89aa98a34..46ffaf04b 100644 --- a/docs/OPENAI_CHAIN_PARITY_REAUDIT.md +++ b/docs/OPENAI_CHAIN_PARITY_REAUDIT.md @@ -495,11 +495,11 @@ Rust 证据： - SSE usage 提取已对齐原版最终 usage 语义，不再对多个事件里重复出现的同一份 usage 做累计；`extract_sse_usage_should_use_completed_response_usage_without_merging_earlier_usage` 覆盖该行为。 - least_used 依赖的 request_count、window_reset_at、quota_limited 字段已经进入运行时调度和数据库恢复路径。 - account retry decision 的失败请求持久 request_count 已对齐原版 release 语义：进入 429/402/403/401/model unsupported/Cloudflare 等账号 fallback/hold 处理时会写入一次无 token 的 request attempt；`v1_responses_should_mark_quota_exhausted_after_402_and_retry_next_account` 覆盖 402 fallback 失败账号计数。 +- 非 account retry decision 的最终失败请求持久 request_count 已对齐到原版 release 语义：未进入账号 fallback/hold 的最终错误会写入一次无 token 的 request attempt；`v1_responses_should_record_request_count_when_5xx_retries_are_exhausted` 覆盖 5xx 内部重试 3 次但持久 request_count 只计 1 次。 未完全对齐： - image_generation usage 未对齐。原版记录 image input/output tokens、image request success/failure 和对应窗口计数；Rust `TokenUsage` 只有 input/output/cached/total，没有解析或持久化 `tool_usage.image_gen`。 - dirty quota verification 缺失。原版离线 reset cachedQuota 后设置 `quotaVerifyRequired`，下一次请求会先调用 `/usage` 验证；Rust 只在本地窗口到期后清除 cooldown/推进窗口，没有等价 dirty 标记和请求前最多 5 次 upstream usage 校验。 -- 非 account retry decision 的失败请求持久 request_count 仍需确认。Rust 已覆盖账号 fallback/hold 分类错误，但纯 transport error、最终 5xx exhausted 等不进入 `UpstreamAccountRetry` 的失败是否应按原版 release 计数，还需要单独 golden test 决策。 - Cookie domain/path 语义不同。原版 CookieJar 是 per-account 简单 map，不按 domain/path 过滤；Rust 按 domain/path 存储和匹配。对 `chatgpt.com` 主链路更严格，但不是原版逐字行为。 - quota window reset 细节不完全一致。原版 reset 后设置 `window_counters_reset_at` 并可能标记 quotaVerifyRequired；Rust 设置 `window_started_at`，没有 `window_counters_reset_at` 字段，也不触发 quotaVerifyRequired。 - passive header 文档提到 `x-codex-active-limit`，原版和 Rust 当前都没有实际解析该 header；如果 Codex Desktop 当前依赖 active-limit，这两边都需要重新确认。 @@ -507,7 +507,7 @@ Rust 证据： 缺口/后续动作： - 补 `TokenUsage` 的 image_generation 字段与数据库/运行时窗口字段，或明确 Rust 不支持 image_generation usage 统计。 - 增加 dirty quota verification 机制：窗口离线 reset 后标记需要校验，请求前调用 usage endpoint，失败时保留 dirty 标记并限制放大次数。 -- 继续统一非 account retry decision 的持久 request_count 计数点；至少补 transport/5xx exhausted 的测试。 +- 继续补 transport error request_count 覆盖；5xx exhausted 已覆盖。 - 明确 cookie domain/path 精细化是 Rust 新架构选择还是需要回到原版 per-account map；如果保留，应补 domain/path 行为测试。 - 若 OpenAI/Codex 当前返回 `x-codex-active-limit`，补解析和 quota_json 存储；否则将其记录为双方未使用字段。 diff --git a/src/codex/serving/chat.rs b/src/codex/serving/chat.rs index 881046623..4a2c5b81f 100644 --- a/src/codex/serving/chat.rs +++ b/src/codex/serving/chat.rs @@ -181,6 +181,22 @@ impl ChatService { .await; return error_response.into_response(); } + if self + .upstream + .record_request_attempt(&acquired.account.id) + .await + .is_err() + { + self.upstream + .log_response( + &log_context, + StatusCode::OK, + EventLevel::Warn, + "v1 chat completions 记录失败请求计数失败", + json!({"stream": client_stream, "requestAttemptStoreError": true}), + ) + .await; + } let error_response = codex_client_error_response(error); self.upstream .log_response( diff --git a/src/codex/serving/dispatch/mod.rs b/src/codex/serving/dispatch/mod.rs index 757180476..0a9994a02 100644 --- a/src/codex/serving/dispatch/mod.rs +++ b/src/codex/serving/dispatch/mod.rs @@ -94,7 +94,7 @@ use self::{ completed_response_metadata, ensure_stream_metadata, has_terminal_sse_event, premature_close_failed_event, TupleStreamReconverter, }, - usage::{record_empty_response_with_deps, record_usage_with_deps}, + usage::{record_empty_response_with_deps, record_request_attempt, record_usage_with_deps}, }; #[derive(Clone)] @@ -320,6 +320,10 @@ impl CodexUpstreamService { record_empty_response_with_deps(&self.deps, account_id).await } + pub(crate) async fn record_request_attempt(&self, account_id: &str) -> Result<(), ()> { + record_request_attempt(&self.deps, account_id).await + } + pub(crate) async fn record_response_affinity( &self, request: &CodexResponsesRequest, @@ -379,6 +383,19 @@ async fn fallback_exhausted_message_with_deps( build_account_exhaustion_detail(summary, message) } +async fn record_failed_request_attempt_with_deps( + deps: &CodexUpstreamDependencies, + account_id: &str, +) { + if let Err(error) = record_request_attempt(deps, account_id).await { + tracing::warn!( + error = ?error, + account_id = %account_id, + "记录上游最终失败请求尝试失败" + ); + } +} + async fn apply_implicit_resume_with_deps( deps: &CodexUpstreamDependencies, request: &mut CodexResponsesRequest, @@ -616,6 +633,7 @@ async fn responses_http_sse_stream( .await; return error_response.into_response(); } + record_failed_request_attempt_with_deps(&deps, &acquired.account.id).await; let error_response = codex_client_error_response(error); log_codex_upstream_response_with_deps( &deps, @@ -1382,6 +1400,7 @@ async fn responses_websocket_stream( return error_response.into_response(); } } + record_failed_request_attempt_with_deps(&deps, &acquired.account.id).await; let error_response = codex_client_error_response(error); log_codex_upstream_response_with_deps( &deps, diff --git a/src/codex/serving/responses.rs b/src/codex/serving/responses.rs index ca351e1dd..9f09f85fc 100644 --- a/src/codex/serving/responses.rs +++ b/src/codex/serving/responses.rs @@ -353,6 +353,22 @@ impl ResponsesService { return error_response.into_response(); } } + if self + .upstream + .record_request_attempt(&acquired.account.id) + .await + .is_err() + { + self.upstream + .log_response( + &log_context, + StatusCode::OK, + EventLevel::Warn, + "v1 responses 记录失败请求计数失败", + json!({"stream": false, "requestAttemptStoreError": true}), + ) + .await; + } let error_response = codex_client_error_response(error); self.upstream .log_response( @@ -453,6 +469,22 @@ impl ResponsesService { return error_response.into_response(); } } + if self + .upstream + .record_request_attempt(&acquired.account.id) + .await + .is_err() + { + self.upstream + .log_response( + &log_context, + StatusCode::OK, + EventLevel::Warn, + "v1 responses 记录失败请求计数失败", + json!({"stream": false, "requestAttemptStoreError": true}), + ) + .await; + } let error_response = codex_client_error_response(error); self.upstream .log_response( @@ -790,6 +822,26 @@ impl ResponsesService { return error_response.into_response(); } } + if self + .upstream + .record_request_attempt(&acquired.account.id) + .await + .is_err() + { + self.upstream + .log_response( + &log_context, + StatusCode::OK, + EventLevel::Warn, + "v1 responses compact 记录失败请求计数失败", + json!({ + "stream": false, + "compact": true, + "requestAttemptStoreError": true, + }), + ) + .await; + } let error_response = codex_client_error_response(error); self.upstream .log_response( diff --git a/tests/codex_serving/upstream_fallback.rs b/tests/codex_serving/upstream_fallback.rs index 022ec283d..0cbd3cd57 100644 --- a/tests/codex_serving/upstream_fallback.rs +++ b/tests/codex_serving/upstream_fallback.rs @@ -305,6 +305,64 @@ async fn v1_responses_should_retry_same_account_after_5xx_before_fallback() { assert_eq!(access_b_posts, 0); } +#[tokio::test] +async fn v1_responses_should_record_request_count_when_5xx_retries_are_exhausted() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .and(header("authorization", "Bearer access-a")) + .respond_with(ResponseTemplate::new(500).set_body_json(json!({ + "error": {"message": "temporary upstream failure"} + }))) + .mount(&server) + .await; + let imported = build_imported_app_with_accounts( + server.uri(), + &[ImportAccount { + id: "acct_5xx_exhausted", + account_id: "chatgpt-a", + token: "access-a", + refresh_token: "refresh-a", + }], + ) + .await; + + let response = imported + .app + .oneshot( + Request::builder() + .method("POST") + .uri("/v1/responses") + .header( + "authorization", + format!("Bearer {}", imported.client_api_key), + ) + .header("content-type", "application/json") + .body(Body::from( + r#"{"model":"gpt-5.5","input":[],"stream":false,"use_websocket":false}"#, + )) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR); + let failed_usage: Option<(i64,)> = + sqlx::query_as("select request_count from account_usage where account_id = ?") + .bind("acct_5xx_exhausted") + .fetch_optional(&imported.pool) + .await + .unwrap(); + assert_eq!(failed_usage.map(|row| row.0).unwrap_or_default(), 1); + + let requests = server.received_requests().await.unwrap(); + let post_count = requests + .iter() + .filter(|request| request.method.as_str() == "POST") + .count(); + assert_eq!(post_count, 3); +} + #[tokio::test] async fn v1_responses_stream_should_retry_same_account_after_5xx_before_fallback() { let server = MockServer::start().await;