diff --git a/README.md b/README.md index ead22b45a..5ab347567 100644 --- a/README.md +++ b/README.md @@ -29,7 +29,7 @@ Codex Proxy RS 为多个 Codex 账号提供统一的 Responses API。客户端 - 单个账号发生额度、认证、模型或传输故障时，继续使用账号池中的其他可用账号。 - 支持跨账号会话续接，并隔离不同账号的会话状态。 - 提供账号、API Key、模型映射、运行参数、用量、错误记录和系统更新管理界面。 -- 使用 PostgreSQL 保存权威业务数据，使用 Redis 保存会话、租约、模型快照和会话亲和数据。 +- 使用 PostgreSQL 保存权威业务数据，使用 Redis 保存管理会话、租约、模型快照和响应归属。 - 内置日志轮转、遥测保留期清理和 PostgreSQL/Redis 健康检查。 除账号调度和会话恢复外，请求与响应尽量保持原有语义。 @@ -180,7 +180,7 @@ Compose 默认只监听宿主机 `127.0.0.1`。需要从其他设备访问时， | 数据 | 默认位置 | | --- | --- | | 账号、API Key、设置、用量和错误记录 | Compose 命名卷 `postgres-data` | -| 管理会话、租约、模型快照和会话亲和 | Compose 命名卷 `redis-data` | +| 管理会话、租约、模型快照和响应归属 | Compose 命名卷 `redis-data` | | 身份派生密钥和更新状态 | `.runtime/data` | | 文件日志 | `.runtime/logs` | | 启动配置 | `.runtime/config.yaml` | diff --git a/backend/src/api/client/responses/websocket.rs b/backend/src/api/client/responses/websocket.rs index 971e3c7df..62413bf92 100644 --- a/backend/src/api/client/responses/websocket.rs +++ b/backend/src/api/client/responses/websocket.rs @@ -1,5 +1,7 @@ //! 官方 Responses WebSocket 入站协议。 +use std::sync::Arc; + use axum::{ body::Body, extract::{Request, State}, @@ -22,7 +24,7 @@ use tokio_tungstenite::{ use crate::{ api::{AppState, middleware::request_id::RequestId}, - dispatch::service::ResponseDispatchStream, + dispatch::{recovery::history::sanitize_replay_items, service::ResponseDispatchStream}, upstream::openai::protocol::{ sse::{SseEvent, SseEventDecoder}, websocket::{is_terminal_websocket_event, websocket_event_type}, @@ -131,6 +133,135 @@ struct ResponsesWebSocketSession { connection_id: String, } +struct ConnectionReplayState { + last_response_id: Option, + input: Option>>, +} + +enum PendingReplayUpdate { + Replace(Vec), + Append(Vec), + Unavailable, +} + +struct CompletedTurn { + response_id: String, + output: Vec, +} + +#[derive(Default)] +struct CompletedTurnCollector { + completed_items: Vec, + completed: Option, +} + +enum ForwardDispatchOutcome { + Continue(Option), + Disconnect, +} + +impl ConnectionReplayState { + fn new() -> Self { + Self { + last_response_id: None, + input: None, + } + } + + fn prepare( + &self, + request: &mut crate::upstream::openai::protocol::responses::CodexResponsesRequest, + ) -> PendingReplayUpdate { + let turn_input = request.input().to_vec(); + match request.previous_response_id() { + None => { + request.local_replay_input = None; + PendingReplayUpdate::Replace(turn_input) + } + Some(previous_response_id) + if self.last_response_id.as_deref() == Some(previous_response_id) + && self.input.is_some() => + { + request.local_replay_input.clone_from(&self.input); + PendingReplayUpdate::Append(turn_input) + } + Some(_) => { + request.local_replay_input = None; + PendingReplayUpdate::Unavailable + } + } + } + + fn commit(&mut self, pending: PendingReplayUpdate, completed: CompletedTurn) { + self.last_response_id = Some(completed.response_id); + match pending { + PendingReplayUpdate::Replace(input) => self.replace(input, completed.output), + PendingReplayUpdate::Append(input) => self.append(input, completed.output), + PendingReplayUpdate::Unavailable => self.clear_input(), + } + } + + fn replace(&mut self, mut input: Vec, mut output: Vec) { + sanitize_replay_items(&mut input); + sanitize_replay_items(&mut output); + input.append(&mut output); + self.input = Some(Arc::new(input)); + } + + fn append(&mut self, mut input: Vec, mut output: Vec) { + sanitize_replay_items(&mut input); + sanitize_replay_items(&mut output); + input.append(&mut output); + let Some(history) = self.input.as_mut() else { + return; + }; + Arc::make_mut(history).append(&mut input); + } + + fn clear_input(&mut self) { + self.input = None; + } +} + +impl CompletedTurnCollector { + fn observe(&mut self, event: &SseEvent) { + let Ok(value) = serde_json::from_str::(&event.data) else { + return; + }; + let event_type = event + .event + .as_deref() + .or_else(|| value.get("type").and_then(Value::as_str)); + match event_type { + Some("response.output_item.done") => { + if let Some(item) = value.get("item") { + self.completed_items.push(item.clone()); + } + } + Some("response.completed") => { + let Some(response_id) = value + .pointer("/response/id") + .and_then(Value::as_str) + .map(ToString::to_string) + else { + return; + }; + let output = value + .pointer("/response/output") + .and_then(Value::as_array) + .filter(|output| !output.is_empty()) + .cloned() + .unwrap_or_else(|| self.completed_items.clone()); + self.completed = Some(CompletedTurn { + response_id, + output, + }); + } + _ => {} + } + } +} + async fn serve_responses_websocket( mut socket: ResponsesWebSocket, session: ResponsesWebSocketSession, @@ -140,6 +271,7 @@ async fn serve_responses_websocket( "Responses WebSocket connected" ); let mut request_count = 0u64; + let mut replay_state = ConnectionReplayState::new(); while let Some(message) = socket.next().await { let payload = match message { @@ -190,7 +322,7 @@ async fn serve_responses_websocket( continue; } }; - let prepared = match prepare_responses_request( + let mut prepared = match prepare_responses_request( &session.state, body, &session.headers, @@ -229,9 +361,14 @@ async fn serve_responses_websocket( continue; } }; + let pending_replay = replay_state.prepare(&mut prepared.request); - if !dispatch_response_create(&mut socket, &session, request_id, prepared).await { - break; + match dispatch_response_create(&mut socket, &session, request_id, prepared).await { + ForwardDispatchOutcome::Continue(Some(completed)) => { + replay_state.commit(pending_replay, completed); + } + ForwardDispatchOutcome::Continue(None) => {} + ForwardDispatchOutcome::Disconnect => break, } } @@ -247,7 +384,7 @@ async fn dispatch_response_create( session: &ResponsesWebSocketSession, request_id: RequestId, prepared: PreparedResponsesRequest, -) -> bool { +) -> ForwardDispatchOutcome { let PreparedResponsesRequest { mut request, requested_model, @@ -270,7 +407,11 @@ async fn dispatch_response_create( Ok(stream) => stream, Err(error) => { let event = responses_websocket_dispatch_error_event(&error, request_id.as_str()); - return send_text(socket, event).await; + return if send_text(socket, event).await { + ForwardDispatchOutcome::Continue(None) + } else { + ForwardDispatchOutcome::Disconnect + }; } }; forward_dispatch_stream(socket, stream, request_id.as_str()).await @@ -280,7 +421,7 @@ async fn forward_dispatch_stream( socket: &mut ResponsesWebSocket, stream: ResponseDispatchStream, request_id: &str, -) -> bool { +) -> ForwardDispatchOutcome { let ResponseDispatchStream { body, response_headers, @@ -291,12 +432,13 @@ async fn forward_dispatch_stream( ) .await { - return false; + return ForwardDispatchOutcome::Disconnect; } let mut body = body; let mut decoder = SseEventDecoder::default(); let mut terminal_sent = false; + let mut completed = CompletedTurnCollector::default(); loop { let next = if terminal_sent { ActiveResponseInput::Upstream(body.next().await) @@ -322,19 +464,31 @@ async fn forward_dispatch_stream( ) .await; drop(body); - return sent; + return if sent { + ForwardDispatchOutcome::Continue(None) + } else { + ForwardDispatchOutcome::Disconnect + }; } }; - if !forward_sse_events(socket, events, request_id, &mut terminal_sent).await { - return false; + if !forward_sse_events( + socket, + events, + request_id, + &mut terminal_sent, + &mut completed, + ) + .await + { + return ForwardDispatchOutcome::Disconnect; } } ActiveResponseInput::Upstream(Some(Err(error))) => { tracing::warn!(request_id, error = %error, "Responses dispatch stream failed"); if terminal_sent { - return true; + return ForwardDispatchOutcome::Continue(completed.completed); } - return send_error( + return if send_error( socket, StatusCode::BAD_GATEWAY, "server_error", @@ -342,14 +496,19 @@ async fn forward_dispatch_stream( STREAM_DISCONNECTED_MESSAGE, request_id, ) - .await; + .await + { + ForwardDispatchOutcome::Continue(None) + } else { + ForwardDispatchOutcome::Disconnect + }; } ActiveResponseInput::Upstream(None) => { let events = match decoder.finish() { Ok(events) => events, Err(error) => { tracing::warn!(request_id, error = %error, "failed to finish Responses SSE decoding"); - return send_error( + return if send_error( socket, StatusCode::BAD_GATEWAY, "server_error", @@ -357,16 +516,29 @@ async fn forward_dispatch_stream( INVALID_UPSTREAM_EVENT_MESSAGE, request_id, ) - .await; + .await + { + ForwardDispatchOutcome::Continue(None) + } else { + ForwardDispatchOutcome::Disconnect + }; } }; - if !forward_sse_events(socket, events, request_id, &mut terminal_sent).await { - return false; + if !forward_sse_events( + socket, + events, + request_id, + &mut terminal_sent, + &mut completed, + ) + .await + { + return ForwardDispatchOutcome::Disconnect; } if terminal_sent { - return true; + return ForwardDispatchOutcome::Continue(completed.completed); } - return send_error( + return if send_error( socket, StatusCode::BAD_GATEWAY, "server_error", @@ -374,23 +546,30 @@ async fn forward_dispatch_stream( STREAM_DISCONNECTED_MESSAGE, request_id, ) - .await; + .await + { + ForwardDispatchOutcome::Continue(None) + } else { + ForwardDispatchOutcome::Disconnect + }; } ActiveResponseInput::Downstream(Some(Ok( Message::Ping(_) | Message::Pong(_) | Message::Frame(_), ))) => {} - ActiveResponseInput::Downstream(Some(Ok(Message::Close(_))) | None) => return false, + ActiveResponseInput::Downstream(Some(Ok(Message::Close(_))) | None) => { + return ForwardDispatchOutcome::Disconnect; + } ActiveResponseInput::Downstream(Some(Ok(Message::Text(_) | Message::Binary(_)))) => { close_for_protocol_violation( socket, "Only one response.create may be active per connection", ) .await; - return false; + return ForwardDispatchOutcome::Disconnect; } ActiveResponseInput::Downstream(Some(Err(error))) => { tracing::info!(request_id, error = %error, "Responses WebSocket client disconnected during response"); - return false; + return ForwardDispatchOutcome::Disconnect; } } } @@ -406,6 +585,7 @@ async fn forward_sse_events( events: Vec, request_id: &str, terminal_sent: &mut bool, + completed: &mut CompletedTurnCollector, ) -> bool { for event in events { let Some(event_type) = websocket_event_type(&event.data) else { @@ -424,6 +604,7 @@ async fn forward_sse_events( .await; return false; }; + completed.observe(&event); if !send_text(socket, event.data).await { return false; } diff --git a/backend/src/auth/service.rs b/backend/src/auth/service.rs index 09d555765..2ed655379 100644 --- a/backend/src/auth/service.rs +++ b/backend/src/auth/service.rs @@ -17,6 +17,7 @@ use super::{ const LOGIN_FAILURE_WINDOW: Duration = Duration::minutes(15); const LOGIN_LOCK_DURATION: Duration = Duration::minutes(15); const LOGIN_MAX_FAILURES: u32 = 5; +const LOGIN_FAILURE_SOURCE_LIMIT: usize = 10_000; /// 管理员会话服务。 #[derive(Clone)] @@ -134,6 +135,9 @@ impl SessionService { let login_locked = { let mut failures = self.login_failures.lock().await; retain_active_login_failures(&mut failures, now); + if !failures.contains_key(source) && failures.len() >= LOGIN_FAILURE_SOURCE_LIMIT { + return Err(SessionError::LoginThrottled); + } failures .get(source) .and_then(|state| state.locked_until) diff --git a/backend/src/dispatch/affinity/resolve.rs b/backend/src/dispatch/affinity/resolve.rs index d449d2c1f..ad388c0a0 100644 --- a/backend/src/dispatch/affinity/resolve.rs +++ b/backend/src/dispatch/affinity/resolve.rs @@ -88,7 +88,6 @@ pub(in crate::dispatch) async fn record_response_affinity( } else { crate::upstream::openai::protocol::responses::PreviousResponseScope::ConnectionLocal }, - replay: history.completed_replay(original_request, &metadata.output), created_at: Utc::now(), }; if let Err(error) = session_affinity diff --git a/backend/src/dispatch/affinity/service.rs b/backend/src/dispatch/affinity/service.rs index 01b34547b..9ea73cd0c 100644 --- a/backend/src/dispatch/affinity/service.rs +++ b/backend/src/dispatch/affinity/service.rs @@ -9,9 +9,6 @@ use super::{ }; const DEFAULT_SESSION_AFFINITY_TTL_SECS: i64 = 4 * 60 * 60; -pub const MAX_REPLAY_SNAPSHOT_BYTES: u64 = 2 * 1024 * 1024; -pub const MAX_REPLAY_SESSION_BYTES: u64 = 16 * 1024 * 1024; -pub const MAX_REPLAY_DEPTH: u16 = 128; /// 运行时会话亲和性服务。 #[derive(Clone)] @@ -49,36 +46,6 @@ impl SessionAffinityService { self.entry(response_id, now).await } - pub async fn replay_input( - &self, - response_id: &str, - entry: &SessionAffinityEntry, - now: DateTime, - ) -> Option> { - match self - .store - .replay_input( - response_id, - entry, - now, - self.ttl, - MAX_REPLAY_DEPTH, - MAX_REPLAY_SESSION_BYTES, - ) - .await - { - Ok(input) => input, - Err(error) => { - tracing::warn!( - response_id, - error = %error, - "failed to reconstruct Redis response replay history" - ); - None - } - } - } - pub async fn lookup_latest_account_by_conversation( &self, conversation_id: &str, diff --git a/backend/src/dispatch/affinity/store.rs b/backend/src/dispatch/affinity/store.rs index 80a35aea3..c852936c7 100644 --- a/backend/src/dispatch/affinity/store.rs +++ b/backend/src/dispatch/affinity/store.rs @@ -1,16 +1,277 @@ //! Redis 会话亲和存储。 -use std::collections::HashMap; - use chrono::{DateTime, Duration, Utc}; use redis::AsyncCommands; -use serde_json::Value; use thiserror::Error; use crate::infra::redis::RedisConnection; use super::types::SessionAffinityEntry; +/// 单会话保留的最近响应元数据数量。 +pub const MAX_CONVERSATION_AFFINITIES: usize = 128; +/// 单账号保留的最近响应索引数量。 +pub const MAX_ACCOUNT_AFFINITIES: usize = 16_384; +/// Redis 全局保留的最近响应元数据数量。 +pub const MAX_GLOBAL_AFFINITIES: usize = 65_536; +/// Redis 全局响应元数据 JSON 字节预算。 +pub const MAX_GLOBAL_AFFINITY_BYTES: u64 = 128 * 1024 * 1024; + +const MAX_AFFINITY_METADATA_BYTES: usize = 64 * 1024; +const GLOBAL_STALE_CLEANUP_BATCH: usize = 256; +const AFFINITY_VERSION: &str = "affinity:v3"; + +const UPSERT_SCRIPT: &str = r#" +local response_id = ARGV[1] +local metadata_json = ARGV[2] +local score = tonumber(ARGV[3]) +local expires_at = tonumber(ARGV[4]) +local ttl_seconds = tonumber(ARGV[5]) +local cutoff_ms = tonumber(ARGV[6]) +local response_prefix = ARGV[7] +local conversation_prefix = ARGV[8] +local account_prefix = ARGV[9] +local max_conversation = tonumber(ARGV[10]) +local max_account = tonumber(ARGV[11]) +local max_global = tonumber(ARGV[12]) +local max_global_bytes = tonumber(ARGV[13]) +local stale_cleanup_batch = tonumber(ARGV[14]) + +local current = cjson.decode(metadata_json) +local conversation_id = current.conversationId +local account_id = current.accountId + +local function decrement_bytes(size) + if size <= 0 then + return + end + local remaining = redis.call('DECRBY', KEYS[5], size) + if remaining <= 0 then + redis.call('DEL', KEYS[5]) + end +end + +local function read_info(id) + local raw = redis.call('HGET', KEYS[6], id) + if not raw then + return nil + end + local ok, info = pcall(cjson.decode, raw) + if not ok then + redis.call('HDEL', KEYS[6], id) + return nil + end + return info +end + +local function remove_response(id) + local response_key = response_prefix .. id + local info = read_info(id) + local metadata = nil + local raw = redis.call('GET', response_key) + if raw then + local ok, decoded = pcall(cjson.decode, raw) + if ok then + metadata = decoded + end + end + + local stored_conversation_id = info and info.conversationId or metadata and metadata.conversationId + local stored_account_id = info and info.accountId or metadata and metadata.accountId + local metadata_bytes = info and tonumber(info.metadataBytes or 0) + or redis.call('STRLEN', response_key) + + redis.call('UNLINK', response_key) + redis.call('ZREM', KEYS[4], id) + redis.call('HDEL', KEYS[6], id) + redis.call('ZREM', KEYS[2], id) + redis.call('ZREM', KEYS[3], id) + decrement_bytes(metadata_bytes) + if stored_conversation_id then + redis.call('ZREM', conversation_prefix .. stored_conversation_id, id) + end + if stored_account_id then + redis.call('ZREM', account_prefix .. stored_account_id, id) + end +end + +local function trim_index(key, max_members) + local excess = redis.call('ZCARD', key) - max_members + if excess <= 0 then + return + end + local ids = redis.call('ZRANGE', key, 0, excess - 1) + for _, id in ipairs(ids) do + remove_response(id) + end +end + +local previous_info = read_info(response_id) +if previous_info then + if previous_info.conversationId and previous_info.conversationId ~= conversation_id then + redis.call('ZREM', conversation_prefix .. previous_info.conversationId, response_id) + end + if previous_info.accountId and previous_info.accountId ~= account_id then + redis.call('ZREM', account_prefix .. previous_info.accountId, response_id) + end +end +local previous_bytes = previous_info and tonumber(previous_info.metadataBytes or 0) + or redis.call('STRLEN', KEYS[1]) +local metadata_bytes = string.len(metadata_json) + +redis.call('SET', KEYS[1], metadata_json, 'EXAT', expires_at) +redis.call('ZADD', KEYS[2], score, response_id) +redis.call('EXPIRE', KEYS[2], ttl_seconds) +redis.call('ZADD', KEYS[3], score, response_id) +redis.call('EXPIRE', KEYS[3], ttl_seconds) +redis.call('ZADD', KEYS[4], score, response_id) +redis.call('EXPIRE', KEYS[4], ttl_seconds) +redis.call('INCRBY', KEYS[5], metadata_bytes - previous_bytes) +redis.call('EXPIRE', KEYS[5], ttl_seconds) +redis.call('HSET', KEYS[6], response_id, cjson.encode({ + conversationId = conversation_id, + accountId = account_id, + metadataBytes = metadata_bytes, +})) +redis.call('EXPIRE', KEYS[6], ttl_seconds) + +local stale_conversation_ids = redis.call('ZRANGEBYSCORE', KEYS[2], '-inf', cutoff_ms) +for _, id in ipairs(stale_conversation_ids) do + remove_response(id) +end +local stale_account_ids = redis.call( + 'ZRANGEBYSCORE', KEYS[3], '-inf', cutoff_ms, 'LIMIT', 0, stale_cleanup_batch +) +for _, id in ipairs(stale_account_ids) do + remove_response(id) +end +local stale_global_ids = redis.call( + 'ZRANGEBYSCORE', KEYS[4], '-inf', cutoff_ms, 'LIMIT', 0, stale_cleanup_batch +) +for _, id in ipairs(stale_global_ids) do + remove_response(id) +end + +trim_index(KEYS[2], max_conversation) +trim_index(KEYS[3], max_account) +trim_index(KEYS[4], max_global) + +while tonumber(redis.call('GET', KEYS[5]) or '0') > max_global_bytes do + local oldest = redis.call('ZRANGE', KEYS[4], 0, 0) + if #oldest == 0 then + redis.call('DEL', KEYS[5]) + break + end + remove_response(oldest[1]) +end + +return 1 +"#; + +const FORGET_SCRIPT: &str = r#" +local response_id = ARGV[1] +local response_prefix = ARGV[2] +local conversation_prefix = ARGV[3] +local account_prefix = ARGV[4] +local response_key = response_prefix .. response_id +local info_json = redis.call('HGET', KEYS[3], response_id) +local metadata_json = redis.call('GET', response_key) +if not info_json and not metadata_json then + return 0 +end + +local info = nil +if info_json then + local ok, decoded = pcall(cjson.decode, info_json) + if ok then + info = decoded + end +end +local metadata = nil +if metadata_json then + local ok, decoded = pcall(cjson.decode, metadata_json) + if ok then + metadata = decoded + end +end +local conversation_id = info and info.conversationId or metadata and metadata.conversationId +local account_id = info and info.accountId or metadata and metadata.accountId +local metadata_bytes = info and tonumber(info.metadataBytes or 0) + or redis.call('STRLEN', response_key) + +redis.call('UNLINK', response_key) +redis.call('ZREM', KEYS[1], response_id) +redis.call('HDEL', KEYS[3], response_id) +if metadata_bytes > 0 then + local remaining = redis.call('DECRBY', KEYS[2], metadata_bytes) + if remaining <= 0 then + redis.call('DEL', KEYS[2]) + end +end +if conversation_id then + redis.call('ZREM', conversation_prefix .. conversation_id, response_id) +end +if account_id then + redis.call('ZREM', account_prefix .. account_id, response_id) +end +return 1 +"#; + +const FORGET_ACCOUNT_SCRIPT: &str = r#" +local account_id = ARGV[1] +local response_prefix = ARGV[2] +local conversation_prefix = ARGV[3] +local account_prefix = ARGV[4] +local account_key = account_prefix .. account_id +local response_ids = redis.call('ZRANGE', account_key, 0, -1) + +local function decrement_bytes(size) + if size <= 0 then + return + end + local remaining = redis.call('DECRBY', KEYS[3], size) + if remaining <= 0 then + redis.call('DEL', KEYS[3]) + end +end + +for _, response_id in ipairs(response_ids) do + local response_key = response_prefix .. response_id + local info_json = redis.call('HGET', KEYS[4], response_id) + local metadata_json = redis.call('GET', response_key) + local info = nil + if info_json then + local ok, decoded = pcall(cjson.decode, info_json) + if ok then + info = decoded + end + end + local metadata = nil + if metadata_json then + local ok, decoded = pcall(cjson.decode, metadata_json) + if ok then + metadata = decoded + end + end + local stored_account_id = info and info.accountId or metadata and metadata.accountId + if stored_account_id == account_id then + local conversation_id = info and info.conversationId or metadata and metadata.conversationId + local metadata_bytes = info and tonumber(info.metadataBytes or 0) + or redis.call('STRLEN', response_key) + redis.call('UNLINK', response_key) + redis.call('ZREM', KEYS[2], response_id) + redis.call('HDEL', KEYS[4], response_id) + decrement_bytes(metadata_bytes) + if conversation_id then + redis.call('ZREM', conversation_prefix .. conversation_id, response_id) + end + end +end + +redis.call('DEL', account_key) +return #response_ids +"#; + /// Redis 会话亲和性存储。 #[derive(Clone)] pub struct RedisSessionAffinityStore { @@ -24,6 +285,8 @@ pub enum RedisSessionAffinityStoreError { Redis(#[from] redis::RedisError), #[error("invalid Redis session affinity value: {0}")] Json(#[from] serde_json::Error), + #[error("session affinity metadata is too large: {bytes} bytes exceeds {max} bytes")] + MetadataTooLarge { bytes: usize, max: usize }, } pub type RedisSessionAffinityStoreResult = Result; @@ -39,7 +302,6 @@ impl RedisSessionAffinityStore { entry: &SessionAffinityEntry, ttl: Duration, ) -> RedisSessionAffinityStoreResult<()> { - let value = serde_json::to_string(entry)?; let expires_at = entry .created_at .checked_add_signed(ttl) @@ -48,50 +310,40 @@ impl RedisSessionAffinityStore { return Ok(()); } - let response_key = self.response_key(response_id); - let conversation_key = self.conversation_key(&entry.conversation_id); - let account_key = self.account_key(&entry.account_id); + let metadata_json = serde_json::to_string(entry)?; + if metadata_json.len() > MAX_AFFINITY_METADATA_BYTES { + return Err(RedisSessionAffinityStoreError::MetadataTooLarge { + bytes: metadata_json.len(), + max: MAX_AFFINITY_METADATA_BYTES, + }); + } + let ttl_seconds = ttl.num_seconds().max(1); let cutoff_ms = Utc::now().timestamp_millis() - ttl.num_milliseconds(); let mut connection = self.redis.manager(); - let mut transaction = redis::pipe(); - transaction - .atomic() - .cmd("SET") - .arg(response_key) - .arg(value) - .arg("EXAT") + let _: i64 = redis::Script::new(UPSERT_SCRIPT) + .key(self.response_key(response_id)) + .key(self.conversation_key(&entry.conversation_id)) + .key(self.account_key(&entry.account_id)) + .key(self.global_index_key()) + .key(self.global_bytes_key()) + .key(self.global_info_key()) + .arg(response_id) + .arg(metadata_json) + .arg(entry.created_at.timestamp_millis()) .arg(expires_at.timestamp()) - .ignore() - .cmd("ZADD") - .arg(&conversation_key) - .arg(entry.created_at.timestamp_millis()) - .arg(response_id) - .ignore() - .cmd("EXPIRE") - .arg(&conversation_key) .arg(ttl_seconds) - .ignore() - .cmd("ZADD") - .arg(&account_key) - .arg(entry.created_at.timestamp_millis()) - .arg(response_id) - .ignore() - .cmd("EXPIRE") - .arg(&account_key) - .arg(ttl_seconds) - .ignore() - .cmd("ZREMRANGEBYSCORE") - .arg(conversation_key) - .arg("-inf") .arg(cutoff_ms) - .ignore() - .cmd("ZREMRANGEBYSCORE") - .arg(&account_key) - .arg("-inf") - .arg(cutoff_ms) - .ignore(); - let _: () = transaction.query_async(&mut connection).await?; + .arg(self.response_key_prefix()) + .arg(self.conversation_key_prefix()) + .arg(self.account_key_prefix()) + .arg(MAX_CONVERSATION_AFFINITIES) + .arg(MAX_ACCOUNT_AFFINITIES) + .arg(MAX_GLOBAL_AFFINITIES) + .arg(MAX_GLOBAL_AFFINITY_BYTES) + .arg(GLOBAL_STALE_CLEANUP_BATCH) + .invoke_async(&mut connection) + .await?; Ok(()) } @@ -128,163 +380,14 @@ impl RedisSessionAffinityStore { ) -> RedisSessionAffinityStoreResult> { let conversation_key = self.conversation_key(conversation_id); let mut connection = self.redis.manager(); - let response_ids: Vec = connection.zrevrange(&conversation_key, 0, -1).await?; - for response_id in response_ids { - let Some(entry) = self.get(&response_id, now, ttl).await? else { - let _: usize = connection.zrem(&conversation_key, &response_id).await?; - continue; - }; - if entry.conversation_id != conversation_id { - let _: usize = connection.zrem(&conversation_key, &response_id).await?; - continue; - } - if variant_hash.is_some_and(|expected| entry.variant_hash.as_deref() != Some(expected)) - { - continue; - } - if max_age.is_some_and(|max_age| now.signed_duration_since(entry.created_at) > max_age) - { - continue; - } - return Ok(Some((response_id, entry))); - } - Ok(None) - } - - pub async fn replay_input( - &self, - response_id: &str, - head: &SessionAffinityEntry, - now: DateTime, - ttl: Duration, - max_depth: u16, - max_total_bytes: u64, - ) -> RedisSessionAffinityStoreResult>> { - let Some(head_snapshot) = head.replay.as_ref() else { - return Ok(None); - }; - if head_snapshot.depth == 0 - || head_snapshot.depth > max_depth - || head_snapshot.total_bytes > max_total_bytes - { - return Ok(None); - } - - let conversation_key = self.conversation_key(&head.conversation_id); - let mut connection = self.redis.manager(); - let response_ids: Vec = redis::cmd("ZREVRANGEBYSCORE") + let response_ids: Vec = redis::cmd("ZREVRANGE") .arg(&conversation_key) - .arg(head.created_at.timestamp_millis()) - .arg("-inf") - .arg("LIMIT") .arg(0) - .arg(max_depth) + .arg(MAX_CONVERSATION_AFFINITIES.saturating_sub(1)) .query_async(&mut connection) .await?; - if !response_ids - .iter() - .any(|candidate| candidate == response_id) - { - return Ok(None); - } - - let response_keys = response_ids - .iter() - .map(|candidate| self.response_key(candidate)) - .collect::>(); - let values: Vec> = redis::cmd("MGET") - .arg(&response_keys) - .query_async(&mut connection) - .await?; - let mut entries = HashMap::with_capacity(values.len()); - for (candidate, value) in response_ids.into_iter().zip(values) { - let Some(value) = value else { - continue; - }; - let entry: SessionAffinityEntry = serde_json::from_str(&value)?; - if entry.conversation_id == head.conversation_id - && entry - .created_at - .checked_add_signed(ttl) - .is_some_and(|expires_at| expires_at > now) - { - entries.insert(candidate, entry); - } - } - - let mut current_id = response_id.to_string(); - let mut expected_depth = head_snapshot.depth; - let mut expected_total_bytes = head_snapshot.total_bytes; - let mut nodes = Vec::with_capacity(usize::from(expected_depth)); - loop { - let Some(entry) = entries.get(¤t_id) else { - return Ok(None); - }; - let Some(snapshot) = entry.replay.as_ref() else { - return Ok(None); - }; - if snapshot.depth != expected_depth || snapshot.total_bytes != expected_total_bytes { - return Ok(None); - } - let node_bytes = replay_node_bytes(snapshot)?; - if node_bytes > expected_total_bytes { - return Ok(None); - } - nodes.push((snapshot.turn_input.clone(), snapshot.turn_output.clone())); - - match snapshot.parent_response_id.as_deref() { - None if expected_depth == 1 && expected_total_bytes == node_bytes => break, - Some(parent_response_id) if expected_depth > 1 => { - expected_depth -= 1; - expected_total_bytes -= node_bytes; - current_id = parent_response_id.to_string(); - } - None | Some(_) => return Ok(None), - } - } - - nodes.reverse(); - let mut full_input = Vec::new(); - for (mut input, mut output) in nodes { - full_input.append(&mut input); - full_input.append(&mut output); - } - Ok(Some(full_input)) - } - - pub async fn forget(&self, response_id: &str) -> RedisSessionAffinityStoreResult { - let response_key = self.response_key(response_id); - let mut connection = self.redis.manager(); - let value: Option = connection.get(&response_key).await?; - let Some(value) = value else { - return Ok(false); - }; - let entry: SessionAffinityEntry = serde_json::from_str(&value)?; - let mut transaction = redis::pipe(); - transaction - .atomic() - .cmd("DEL") - .arg(response_key) - .ignore() - .cmd("ZREM") - .arg(self.conversation_key(&entry.conversation_id)) - .arg(response_id) - .ignore() - .cmd("ZREM") - .arg(self.account_key(&entry.account_id)) - .arg(response_id) - .ignore(); - let _: () = transaction.query_async(&mut connection).await?; - Ok(true) - } - - pub async fn forget_account(&self, account_id: &str) -> RedisSessionAffinityStoreResult { - let account_key = self.account_key(account_id); - let mut connection = self.redis.manager(); - let response_ids: Vec = connection.zrange(&account_key, 0, -1).await?; if response_ids.is_empty() { - let _: usize = connection.del(account_key).await?; - return Ok(0); + return Ok(None); } let response_keys = response_ids @@ -292,67 +395,105 @@ impl RedisSessionAffinityStore { .map(|response_id| self.response_key(response_id)) .collect::>(); let values: Vec> = redis::cmd("MGET") - .arg(&response_keys) + .arg(response_keys) .query_async(&mut connection) .await?; - let mut conversation_members = Vec::with_capacity(response_ids.len()); - for (response_id, value) in response_ids.iter().zip(values) { + let mut stale_ids = Vec::new(); + let mut result = None; + for (response_id, value) in response_ids.into_iter().zip(values) { let Some(value) = value else { + stale_ids.push(response_id); continue; }; - match serde_json::from_str::(&value) { - Ok(entry) if entry.account_id == account_id => { - conversation_members.push((response_id, entry.conversation_id)); - } - Ok(_) => { - tracing::warn!( - response_id, - account_id, - "account affinity index points to another account" - ); - } - Err(error) => { - tracing::warn!( - response_id, - account_id, - error = %error, - "invalid account affinity entry removed during account cleanup" - ); - } + let entry: SessionAffinityEntry = serde_json::from_str(&value)?; + if entry.conversation_id != conversation_id + || entry + .created_at + .checked_add_signed(ttl) + .is_none_or(|expires_at| expires_at <= now) + { + stale_ids.push(response_id); + continue; } + if variant_hash.is_some_and(|expected| entry.variant_hash.as_deref() != Some(expected)) + || max_age + .is_some_and(|max_age| now.signed_duration_since(entry.created_at) > max_age) + { + continue; + } + result = Some((response_id, entry)); + break; } + if !stale_ids.is_empty() { + let _: usize = connection.zrem(conversation_key, stale_ids).await?; + } + Ok(result) + } - let mut transaction = redis::pipe(); - transaction.atomic().cmd("DEL").arg(&response_keys).ignore(); - for (response_id, conversation_id) in &conversation_members { - transaction - .cmd("ZREM") - .arg(self.conversation_key(conversation_id)) - .arg(response_id) - .ignore(); - } - transaction.cmd("DEL").arg(account_key).ignore(); - let _: () = transaction.query_async(&mut connection).await?; - Ok(response_ids.len() as u64) + pub async fn forget(&self, response_id: &str) -> RedisSessionAffinityStoreResult { + let mut connection = self.redis.manager(); + let forgotten: i64 = redis::Script::new(FORGET_SCRIPT) + .key(self.global_index_key()) + .key(self.global_bytes_key()) + .key(self.global_info_key()) + .arg(response_id) + .arg(self.response_key_prefix()) + .arg(self.conversation_key_prefix()) + .arg(self.account_key_prefix()) + .invoke_async(&mut connection) + .await?; + Ok(forgotten > 0) + } + + pub async fn forget_account(&self, account_id: &str) -> RedisSessionAffinityStoreResult { + let mut connection = self.redis.manager(); + let forgotten: u64 = redis::Script::new(FORGET_ACCOUNT_SCRIPT) + .key(self.account_key(account_id)) + .key(self.global_index_key()) + .key(self.global_bytes_key()) + .key(self.global_info_key()) + .arg(account_id) + .arg(self.response_key_prefix()) + .arg(self.conversation_key_prefix()) + .arg(self.account_key_prefix()) + .invoke_async(&mut connection) + .await?; + Ok(forgotten) } fn response_key(&self, response_id: &str) -> String { - self.redis.key(&format!("affinity:v2:resp:{response_id}")) + format!("{}{response_id}", self.response_key_prefix()) } fn conversation_key(&self, conversation_id: &str) -> String { - self.redis - .key(&format!("affinity:v2:conv:{conversation_id}")) + format!("{}{conversation_id}", self.conversation_key_prefix()) } fn account_key(&self, account_id: &str) -> String { - self.redis.key(&format!("affinity:v2:account:{account_id}")) + format!("{}{account_id}", self.account_key_prefix()) + } + + fn global_index_key(&self) -> String { + self.redis.key(&format!("{AFFINITY_VERSION}:global:index")) + } + + fn global_bytes_key(&self) -> String { + self.redis.key(&format!("{AFFINITY_VERSION}:global:bytes")) + } + + fn global_info_key(&self) -> String { + self.redis.key(&format!("{AFFINITY_VERSION}:global:info")) + } + + fn response_key_prefix(&self) -> String { + self.redis.key(&format!("{AFFINITY_VERSION}:resp:")) + } + + fn conversation_key_prefix(&self) -> String { + self.redis.key(&format!("{AFFINITY_VERSION}:conv:")) + } + + fn account_key_prefix(&self) -> String { + self.redis.key(&format!("{AFFINITY_VERSION}:account:")) } } - -fn replay_node_bytes( - snapshot: &super::types::ResponseReplaySnapshot, -) -> Result { - serde_json::to_vec(&(&snapshot.turn_input, &snapshot.turn_output)) - .map(|value| value.len() as u64) -} diff --git a/backend/src/dispatch/affinity/types.rs b/backend/src/dispatch/affinity/types.rs index c071f2aca..bb4d8fec0 100644 --- a/backend/src/dispatch/affinity/types.rs +++ b/backend/src/dispatch/affinity/types.rs @@ -2,7 +2,6 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; -use serde_json::Value; use crate::upstream::openai::protocol::responses::PreviousResponseScope; @@ -18,22 +17,5 @@ pub struct SessionAffinityEntry { pub function_call_ids: Vec, pub variant_hash: Option, pub continuation_scope: PreviousResponseScope, - pub replay: Option, pub created_at: DateTime, } - -/// 单轮 completed response 的有界、无凭据重放增量。 -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub struct ResponseReplaySnapshot { - /// 上一轮由本代理记录的 completed response。 - pub parent_response_id: Option, - /// 本轮客户端新增的输入，已剥离账号绑定字段。 - pub turn_input: Vec, - /// 本轮上游输出，已剥离账号绑定字段和上游对象 ID。 - pub turn_output: Vec, - /// 从会话起点到本节点的增量节点数。 - pub depth: u16, - /// 从会话起点到本节点的 JSON 编码累计字节数。 - pub total_bytes: u64, -} diff --git a/backend/src/dispatch/attempts.rs b/backend/src/dispatch/attempts.rs index 585ecd3a1..b414bdd06 100644 --- a/backend/src/dispatch/attempts.rs +++ b/backend/src/dispatch/attempts.rs @@ -3,11 +3,14 @@ use std::collections::{BTreeSet, VecDeque}; use chrono::Utc; +use tokio::time::{Duration, Instant, timeout_at}; use crate::fleet::pool::{ AccountAcquireRequest, AccountCandidateLease, AccountLease, AccountPoolService, }; +const ACCOUNT_CANDIDATE_WAIT_TIMEOUT: Duration = Duration::from_secs(10); + /// 请求开始时冻结的完整候选集，以及候选的最终处理状态。 pub(in crate::dispatch) struct AccountAttemptLedger { model: String, @@ -16,6 +19,7 @@ pub(in crate::dispatch) struct AccountAttemptLedger { busy: VecDeque, attempted: BTreeSet, state_excluded: BTreeSet, + busy_wait_deadline: Instant, } impl AccountAttemptLedger { @@ -31,6 +35,7 @@ impl AccountAttemptLedger { busy: VecDeque::new(), attempted: BTreeSet::new(), state_excluded: BTreeSet::new(), + busy_wait_deadline: Instant::now() + ACCOUNT_CANDIDATE_WAIT_TIMEOUT, } } @@ -63,7 +68,22 @@ impl AccountAttemptLedger { return None; } - account_pool.wait_for_candidate_change().await; + if timeout_at( + self.busy_wait_deadline, + account_pool.wait_for_candidate_change(), + ) + .await + .is_err() + { + tracing::warn!( + candidate_count = self.candidate_ids.len(), + attempted_count = self.attempted.len(), + busy_count = self.busy.len(), + timeout_ms = ACCOUNT_CANDIDATE_WAIT_TIMEOUT.as_millis(), + "timed out waiting for a busy account candidate" + ); + return None; + } std::mem::swap(&mut self.pending, &mut self.busy); } } diff --git a/backend/src/dispatch/errors.rs b/backend/src/dispatch/errors.rs index 5b277fb8f..6d966cbcb 100644 --- a/backend/src/dispatch/errors.rs +++ b/backend/src/dispatch/errors.rs @@ -206,8 +206,11 @@ pub(crate) fn is_retryable_upstream_5xx_error(error: &CodexClientError) -> bool pub(crate) fn is_retryable_account_transport_error(error: &CodexClientError) -> bool { match error { CodexClientError::Http(error) => error.is_connect() || error.is_timeout(), + CodexClientError::StreamIdleTimeout { .. } => true, CodexClientError::WebSocket( crate::upstream::openai::transport::websocket::CodexWebSocketExchangeError::Transport(_) + | crate::upstream::openai::transport::websocket::CodexWebSocketExchangeError::ConnectTimeout { .. } + | crate::upstream::openai::transport::websocket::CodexWebSocketExchangeError::SendTimeout { .. } | crate::upstream::openai::transport::websocket::CodexWebSocketExchangeError::ClosedBeforeTerminal | crate::upstream::openai::transport::websocket::CodexWebSocketExchangeError::ReceiveIdleTimeout { .. } | crate::upstream::openai::transport::websocket::CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput { .. } diff --git a/backend/src/dispatch/recording.rs b/backend/src/dispatch/recording.rs index 8849e7f8d..118e82745 100644 --- a/backend/src/dispatch/recording.rs +++ b/backend/src/dispatch/recording.rs @@ -204,11 +204,13 @@ pub(super) async fn record_prefetched_response_stream_failure_event( diagnostics_status_code(record.diagnostics), ); insert_response_request_summary_object(object, record.request, record.transport); - object.insert("requestBody".to_string(), json!(record.request)); - object.insert( - "responseBody".to_string(), - Value::String(String::from_utf8_lossy(record.prefetched).to_string()), - ); + if record.recorder.captures_body() { + object.insert("requestBody".to_string(), json!(record.request)); + object.insert( + "responseBody".to_string(), + Value::String(String::from_utf8_lossy(record.prefetched).to_string()), + ); + } insert_upstream_diagnostics_metadata(object, record.diagnostics); insert_response_trace_metadata_object( object, @@ -295,12 +297,14 @@ fn enrich_live_response_stream_metadata( insert_response_request_summary_object(object, &context.request, context.transport); insert_upstream_diagnostics_metadata(object, &context.diagnostics); insert_response_trace_metadata_object(object, &context.attempts, Some(&context.attempt)); - object - .entry("requestBody".to_string()) - .or_insert_with(|| serde_json::json!(context.request)); - object - .entry("responseBody".to_string()) - .or_insert_with(|| Value::String(body.to_string())); + if context.recorder.captures_body() { + object + .entry("requestBody".to_string()) + .or_insert_with(|| serde_json::json!(context.request)); + object + .entry("responseBody".to_string()) + .or_insert_with(|| Value::String(body.to_string())); + } } pub(super) fn insert_response_upstream_diagnostics( diff --git a/backend/src/dispatch/recovery/history.rs b/backend/src/dispatch/recovery/history.rs index 208341aa6..5b555be25 100644 --- a/backend/src/dispatch/recovery/history.rs +++ b/backend/src/dispatch/recovery/history.rs @@ -1,14 +1,12 @@ -//! previous response 所有权与完整上下文重放。 +//! previous response 所有权与连接内上下文重放。 + +use std::sync::Arc; use chrono::Utc; use serde_json::Value; use crate::{ - dispatch::affinity::{ - SessionAffinityEntry, SessionAffinityService, - service::{MAX_REPLAY_DEPTH, MAX_REPLAY_SESSION_BYTES, MAX_REPLAY_SNAPSHOT_BYTES}, - types::ResponseReplaySnapshot, - }, + dispatch::affinity::{SessionAffinityEntry, SessionAffinityService}, upstream::openai::protocol::responses::{CodexResponsesRequest, StreamCommitPolicy}, }; @@ -17,6 +15,7 @@ use crate::{ pub(in crate::dispatch) enum HistorySource { None, Managed, + ConnectionLocal, ExternalUnknown, } @@ -24,7 +23,7 @@ pub(in crate::dispatch) enum HistorySource { pub(in crate::dispatch) struct HistoryRecoveryPlan { source: HistorySource, managed_entry: Option, - replay_input: Option>, + local_replay_input: Option>>, external_attempted_account_id: Option, replay_owner_account: bool, } @@ -38,7 +37,7 @@ impl HistoryRecoveryPlan { return Self { source: HistorySource::None, managed_entry: None, - replay_input: None, + local_replay_input: None, external_attempted_account_id: None, replay_owner_account: false, }; @@ -46,22 +45,16 @@ impl HistoryRecoveryPlan { let managed_entry = session_affinity .lookup(previous_response_id, Utc::now()) .await; - let replay_input = match managed_entry.as_ref() { - Some(entry) => { - session_affinity - .replay_input(previous_response_id, entry, Utc::now()) - .await - } - None => None, + let local_replay_input = request.local_replay_input.clone(); + let source = match (&managed_entry, &local_replay_input) { + (Some(_), _) => HistorySource::Managed, + (None, Some(_)) => HistorySource::ConnectionLocal, + (None, None) => HistorySource::ExternalUnknown, }; Self { - source: if managed_entry.is_some() { - HistorySource::Managed - } else { - HistorySource::ExternalUnknown - }, + source, managed_entry, - replay_input, + local_replay_input, external_attempted_account_id: None, replay_owner_account: false, } @@ -83,23 +76,17 @@ impl HistoryRecoveryPlan { HistorySource::ExternalUnknown => match self.external_attempted_account_id.as_deref() { None => { self.external_attempted_account_id = Some(account_id.to_string()); - let mut request = original.clone(); - request.stream_commit_policy = StreamCommitPolicy::UntilOutputOrTerminal; - request.previous_response_scope = Some( - crate::upstream::openai::protocol::responses::PreviousResponseScope::ExternalUnknown, - ); - Some(request) + Some(external_history_request(original)) } Some(first_account_id) if first_account_id == account_id => { - let mut request = original.clone(); - request.stream_commit_policy = StreamCommitPolicy::UntilOutputOrTerminal; - request.previous_response_scope = Some( - crate::upstream::openai::protocol::responses::PreviousResponseScope::ExternalUnknown, - ); - Some(request) + Some(external_history_request(original)) } Some(_) => None, }, + HistorySource::ConnectionLocal => self + .local_replay_input + .as_deref() + .map(|replay_input| full_replay_request(original, replay_input)), HistorySource::Managed => { let entry = self.managed_entry.as_ref()?; if entry.account_id == account_id && !self.replay_owner_account { @@ -112,7 +99,7 @@ impl HistoryRecoveryPlan { } Some(request) } else { - self.replay_input.as_ref().map(|replay_input| { + self.local_replay_input.as_deref().map(|replay_input| { let mut request = full_replay_request(original, replay_input); request.local_conversation_id = Some(entry.conversation_id.clone()); request @@ -128,7 +115,7 @@ impl HistoryRecoveryPlan { return false; }; if entry.account_id != account_id - || self.replay_input.is_none() + || self.local_replay_input.is_none() || self.replay_owner_account { return false; @@ -138,67 +125,17 @@ impl HistoryRecoveryPlan { } pub(in crate::dispatch) fn can_failover(&self) -> bool { - !matches!(self.source, HistorySource::ExternalUnknown) - && self - .managed_entry - .as_ref() - .is_none_or(|_| self.replay_input.is_some()) + match self.source { + HistorySource::None | HistorySource::ConnectionLocal => true, + HistorySource::Managed => self.local_replay_input.is_some(), + HistorySource::ExternalUnknown => false, + } } pub(in crate::dispatch) fn is_external_unknown(&self) -> bool { self.source == HistorySource::ExternalUnknown } - pub(in crate::dispatch) fn completed_replay( - &self, - original: &CodexResponsesRequest, - output: &[Value], - ) -> Option { - let (parent_response_id, parent_depth, parent_bytes) = match self.source { - HistorySource::None => (None, 0, 0), - HistorySource::Managed => { - self.replay_input.as_ref()?; - let replay = self.managed_entry.as_ref()?.replay.as_ref()?; - ( - original.previous_response_id().map(ToString::to_string), - replay.depth, - replay.total_bytes, - ) - } - HistorySource::ExternalUnknown => return None, - }; - let depth = parent_depth.checked_add(1)?; - if depth > MAX_REPLAY_DEPTH { - tracing::warn!(depth, "response replay depth limit reached"); - return None; - } - - let mut turn_input = original.input().to_vec(); - let mut turn_output = output.to_vec(); - sanitize_replay_items(&mut turn_input); - sanitize_replay_items(&mut turn_output); - let Some(node_bytes) = limited_replay_json_bytes(&turn_input, &turn_output) else { - tracing::warn!("response replay snapshot byte limit reached"); - return None; - }; - let total_bytes = parent_bytes.checked_add(node_bytes)?; - if total_bytes > MAX_REPLAY_SESSION_BYTES { - tracing::warn!( - node_bytes, - total_bytes, - "response replay session byte limit reached" - ); - return None; - } - Some(ResponseReplaySnapshot { - parent_response_id, - turn_input, - turn_output, - depth, - total_bytes, - }) - } - pub(in crate::dispatch) fn conversation_id<'a>( &'a self, original: &'a CodexResponsesRequest, @@ -210,12 +147,21 @@ impl HistoryRecoveryPlan { } } +fn external_history_request(original: &CodexResponsesRequest) -> CodexResponsesRequest { + let mut request = original.clone(); + request.stream_commit_policy = StreamCommitPolicy::UntilOutputOrTerminal; + request.previous_response_scope = + Some(crate::upstream::openai::protocol::responses::PreviousResponseScope::ExternalUnknown); + request +} + fn full_replay_request( original: &CodexResponsesRequest, replay_input: &[Value], ) -> CodexResponsesRequest { let mut request = original.clone(); - let mut input = replay_input.to_vec(); + let mut input = Vec::with_capacity(replay_input.len() + original.input().len()); + input.extend_from_slice(replay_input); input.extend_from_slice(original.input()); sanitize_replay_items(&mut input); request.set_input(input); @@ -225,7 +171,7 @@ fn full_replay_request( request } -fn sanitize_replay_items(values: &mut [Value]) { +pub(crate) fn sanitize_replay_items(values: &mut [Value]) { for value in values { match value { Value::Array(values) => sanitize_replay_items(values), @@ -233,7 +179,7 @@ fn sanitize_replay_items(values: &mut [Value]) { object.remove("id"); object.remove("encrypted_content"); for value in object.values_mut() { - strip_encrypted_content(std::slice::from_mut(value)); + sanitize_replay_items(std::slice::from_mut(value)); } } Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => {} @@ -241,53 +187,6 @@ fn sanitize_replay_items(values: &mut [Value]) { } } -fn strip_encrypted_content(values: &mut [Value]) { - for value in values { - match value { - Value::Array(values) => strip_encrypted_content(values), - Value::Object(object) => { - object.remove("encrypted_content"); - for value in object.values_mut() { - strip_encrypted_content(std::slice::from_mut(value)); - } - } - Value::Null | Value::Bool(_) | Value::Number(_) | Value::String(_) => {} - } - } -} - -fn limited_replay_json_bytes(input: &[Value], output: &[Value]) -> Option { - let mut writer = ReplaySizeWriter::default(); - serde_json::to_writer(&mut writer, &(input, output)).ok()?; - Some(writer.written) -} - -#[derive(Default)] -struct ReplaySizeWriter { - written: u64, -} - -impl std::io::Write for ReplaySizeWriter { - fn write(&mut self, buffer: &[u8]) -> std::io::Result { - let next = self - .written - .checked_add(buffer.len() as u64) - .filter(|next| *next <= MAX_REPLAY_SNAPSHOT_BYTES) - .ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::FileTooLarge, - "response replay snapshot byte limit reached", - ) - })?; - self.written = next; - Ok(buffer.len()) - } - - fn flush(&mut self) -> std::io::Result<()> { - Ok(()) - } -} - fn non_empty(value: Option<&str>) -> Option<&str> { value.map(str::trim).filter(|value| !value.is_empty()) } diff --git a/backend/src/dispatch/service.rs b/backend/src/dispatch/service.rs index 559538265..cee820c70 100644 --- a/backend/src/dispatch/service.rs +++ b/backend/src/dispatch/service.rs @@ -32,7 +32,7 @@ use crate::{ }, upstream_call::{ AccountUpstreamContext, QUOTA_VERIFY_LIMIT_REACHED_MESSAGE, QuotaVerificationContext, - QuotaVerificationDecision, create_response_with_account_retrying_5xx, + QuotaVerificationDecision, create_response_with_account, verify_acquired_quota_if_required, }, }, @@ -266,7 +266,7 @@ impl ResponseDispatchService { let account = acquired.account.clone(); let release_account_id = account.id.clone(); let attempt = trace.start_attempt(&release_account_id); - let response_result = create_response_with_account_retrying_5xx( + let response_result = create_response_with_account( AccountUpstreamContext { codex: &self.codex, account_identity: &self.account_identity, diff --git a/backend/src/dispatch/stream/lifecycle.rs b/backend/src/dispatch/stream/lifecycle.rs index 5a1482169..2970ecc78 100644 --- a/backend/src/dispatch/stream/lifecycle.rs +++ b/backend/src/dispatch/stream/lifecycle.rs @@ -32,7 +32,7 @@ use crate::{ }, upstream_call::{ AccountUpstreamContext, QUOTA_VERIFY_LIMIT_REACHED_MESSAGE, QuotaVerificationContext, - QuotaVerificationDecision, create_response_stream_with_account_retrying_5xx, + QuotaVerificationDecision, create_response_stream_with_account, verify_acquired_quota_if_required, }, }, @@ -182,7 +182,7 @@ impl ResponseDispatchService { let account = acquired.account.clone(); let release_account_id = account.id.clone(); let attempt = trace.start_attempt(&release_account_id); - let response_result = create_response_stream_with_account_retrying_5xx( + let response_result = create_response_stream_with_account( AccountUpstreamContext { codex: &self.codex, account_identity: &self.account_identity, @@ -215,12 +215,6 @@ impl ResponseDispatchService { let response_metadata = response.response_metadata; self.models .observe_models_etag(response_metadata.models_etag.as_deref()); - self.cloudflare - .capture_set_cookie_headers(&release_account_id, &set_cookie_headers) - .await; - self.account_pool - .sync_passive_rate_limit_headers(&account, &rate_limit_headers) - .await; let (prefetched, body) = match prefetch_first_sse_chunk(response.body).await { Ok(prefetched) => prefetched, Err(ResponseDispatchError::Upstream(error)) => { @@ -395,6 +389,7 @@ impl ResponseDispatchService { request, tuple_schema, transport, + set_cookie_headers, rate_limit_headers, rate_limit_header_updates, turn_state_update, diff --git a/backend/src/dispatch/stream/live.rs b/backend/src/dispatch/stream/live.rs index e5138feda..1d8179513 100644 --- a/backend/src/dispatch/stream/live.rs +++ b/backend/src/dispatch/stream/live.rs @@ -29,8 +29,8 @@ use crate::{ events::extract_sse_usage, responses::{ CodexResponsesRequest, CollectedResponse, - reconvert_responses_sse_event_tuple_values, response_from_codex_sse, - response_sse_event_is_terminal, update_first_response_output_ms, + reconvert_responses_sse_event_tuple_values, response_body_has_first_output, + response_from_codex_sse, response_sse_event_is_terminal, }, sse::{ DONE_SSE_FRAME, encode_sse_event, parse_sse_events, @@ -45,6 +45,10 @@ use crate::{ }, }; +const MAX_LIVE_RESPONSE_CAPTURE_BYTES: usize = 16 * 1024 * 1024; +const LIVE_RESPONSE_CAPTURE_LIMIT_DETAIL: &str = + "upstream response exceeded the 16 MiB proxy capture limit"; + use super::{ sse_failure::{ STREAM_DISCONNECTED_CODE, STREAM_DISCONNECTED_MESSAGE, status_code_for_stream_failure, @@ -89,7 +93,9 @@ pub(in crate::dispatch) fn spawn_live_response_stream( .map(TupleSseEventTransformer::new); let mut body_bytes = Vec::new(); let mut first_token_ms = None; - if !send_live_response_stream_chunk( + let mut first_output_detector = FirstOutputDetector::default(); + first_output_detector.observe(context.started_at, &prefetched, &mut first_token_ms); + match send_live_response_stream_chunk( &sender, &mut body_bytes, tuple_transformer.as_mut(), @@ -97,10 +103,17 @@ pub(in crate::dispatch) fn spawn_live_response_stream( ) .await { - context.account_lease.complete().await; - return; + Ok(()) => {} + Err(LiveStreamWriteError::DownstreamClosed) => { + context.account_lease.complete().await; + return; + } + Err(LiveStreamWriteError::CaptureLimitExceeded) => { + terminate_oversized_live_stream(context, &sender, &mut body_bytes, first_token_ms) + .await; + return; + } } - update_first_response_output_ms(context.started_at, &body_bytes, &mut first_token_ms); loop { let next = tokio::select! { @@ -115,7 +128,8 @@ pub(in crate::dispatch) fn spawn_live_response_stream( }; match next { Ok(chunk) => { - if !send_live_response_stream_chunk( + first_output_detector.observe(context.started_at, &chunk, &mut first_token_ms); + match send_live_response_stream_chunk( &sender, &mut body_bytes, tuple_transformer.as_mut(), @@ -123,25 +137,46 @@ pub(in crate::dispatch) fn spawn_live_response_stream( ) .await { - context.account_lease.complete().await; - return; + Ok(()) => {} + Err(LiveStreamWriteError::DownstreamClosed) => { + context.account_lease.complete().await; + return; + } + Err(LiveStreamWriteError::CaptureLimitExceeded) => { + terminate_oversized_live_stream( + context, + &sender, + &mut body_bytes, + first_token_ms, + ) + .await; + return; + } } - update_first_response_output_ms( - context.started_at, - &body_bytes, - &mut first_token_ms, - ); } Err(error) => { - if !flush_live_response_stream_transformer( + match flush_live_response_stream_transformer( &sender, &mut body_bytes, tuple_transformer.as_mut(), ) .await { - context.account_lease.complete().await; - return; + Ok(()) => {} + Err(LiveStreamWriteError::DownstreamClosed) => { + context.account_lease.complete().await; + return; + } + Err(LiveStreamWriteError::CaptureLimitExceeded) => { + terminate_oversized_live_stream( + context, + &sender, + &mut body_bytes, + first_token_ms, + ) + .await; + return; + } } let detail = error.to_string(); let Some(body_text) = @@ -157,15 +192,23 @@ pub(in crate::dispatch) fn spawn_live_response_stream( } } - if !flush_live_response_stream_transformer( + match flush_live_response_stream_transformer( &sender, &mut body_bytes, tuple_transformer.as_mut(), ) .await { - context.account_lease.complete().await; - return; + Ok(()) => {} + Err(LiveStreamWriteError::DownstreamClosed) => { + context.account_lease.complete().await; + return; + } + Err(LiveStreamWriteError::CaptureLimitExceeded) => { + terminate_oversized_live_stream(context, &sender, &mut body_bytes, first_token_ms) + .await; + return; + } } let Some(body_text) = send_live_response_stream_tail(&sender, &mut body_bytes, None).await else { @@ -190,9 +233,9 @@ async fn send_live_response_stream_chunk( body_bytes: &mut Vec, transformer: Option<&mut TupleSseEventTransformer>, chunk: Bytes, -) -> bool { +) -> Result<(), LiveStreamWriteError> { let chunks = match transformer { - Some(transformer) => transformer.push(&chunk), + Some(transformer) => transformer.push(&chunk)?, None => vec![chunk], }; send_live_response_stream_chunks(sender, body_bytes, chunks).await @@ -202,25 +245,59 @@ async fn flush_live_response_stream_transformer( sender: &mpsc::Sender>, body_bytes: &mut Vec, transformer: Option<&mut TupleSseEventTransformer>, -) -> bool { +) -> Result<(), LiveStreamWriteError> { let Some(transformer) = transformer else { - return true; + return Ok(()); }; - send_live_response_stream_chunks(sender, body_bytes, transformer.finish()).await + send_live_response_stream_chunks(sender, body_bytes, transformer.finish()?).await } async fn send_live_response_stream_chunks( sender: &mpsc::Sender>, body_bytes: &mut Vec, chunks: Vec, -) -> bool { +) -> Result<(), LiveStreamWriteError> { for chunk in chunks { + if body_bytes.len().saturating_add(chunk.len()) > MAX_LIVE_RESPONSE_CAPTURE_BYTES { + return Err(LiveStreamWriteError::CaptureLimitExceeded); + } body_bytes.extend_from_slice(&chunk); if sender.send(Ok(chunk)).await.is_err() { - return false; + return Err(LiveStreamWriteError::DownstreamClosed); + } + } + Ok(()) +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum LiveStreamWriteError { + DownstreamClosed, + CaptureLimitExceeded, +} + +#[derive(Default)] +struct FirstOutputDetector { + pending: Vec, +} + +impl FirstOutputDetector { + fn observe(&mut self, started_at: Instant, chunk: &[u8], first_token_ms: &mut Option) { + if first_token_ms.is_some() { + return; + } + self.pending.extend_from_slice(chunk); + while let Some(frame_end) = sse_frame_end(&self.pending) { + let frame = self.pending.drain(..frame_end).collect::>(); + if response_body_has_first_output(&frame) { + *first_token_ms = Some(elapsed_millis_i64(started_at).max(1)); + self.pending.clear(); + return; + } + } + if self.pending.len() > MAX_LIVE_RESPONSE_CAPTURE_BYTES { + self.pending.clear(); } } - true } struct TupleSseEventTransformer { @@ -236,22 +313,25 @@ impl TupleSseEventTransformer { } } - fn push(&mut self, chunk: &[u8]) -> Vec { + fn push(&mut self, chunk: &[u8]) -> Result, LiveStreamWriteError> { self.pending.extend_from_slice(chunk); + if self.pending.len() > MAX_LIVE_RESPONSE_CAPTURE_BYTES { + return Err(LiveStreamWriteError::CaptureLimitExceeded); + } let mut chunks = Vec::new(); while let Some(frame_end) = sse_frame_end(&self.pending) { let frame = self.pending.drain(..frame_end).collect::>(); chunks.push(self.transform_frame(&frame)); } - chunks + Ok(chunks) } - fn finish(&mut self) -> Vec { + fn finish(&mut self) -> Result, LiveStreamWriteError> { if self.pending.is_empty() { - return Vec::new(); + return Ok(Vec::new()); } let frame = std::mem::take(&mut self.pending); - vec![self.transform_frame(&frame)] + Ok(vec![self.transform_frame(&frame)]) } fn transform_frame(&self, frame: &[u8]) -> Bytes { @@ -277,6 +357,32 @@ impl TupleSseEventTransformer { } } +async fn terminate_oversized_live_stream( + context: LiveResponseStreamContext, + sender: &mpsc::Sender>, + body_bytes: &mut Vec, + first_token_ms: Option, +) { + tracing::warn!( + request_id = %context.request_id, + account_id = %context.account_id, + captured_bytes = body_bytes.len(), + capture_limit_bytes = MAX_LIVE_RESPONSE_CAPTURE_BYTES, + "live response capture limit exceeded" + ); + let Some(body_text) = send_live_response_stream_tail( + sender, + body_bytes, + Some(LIVE_RESPONSE_CAPTURE_LIMIT_DETAIL), + ) + .await + else { + context.account_lease.complete().await; + return; + }; + finalize_live_response_stream(context, body_text, first_token_ms).await; +} + async fn send_live_response_stream_tail( sender: &mpsc::Sender>, body_bytes: &mut Vec, @@ -354,6 +460,7 @@ pub(in crate::dispatch) struct LiveResponseStreamContext { pub(in crate::dispatch) request: CodexResponsesRequest, pub(in crate::dispatch) tuple_schema: Option, pub(in crate::dispatch) transport: CodexBackendTransport, + pub(in crate::dispatch) set_cookie_headers: Vec, pub(in crate::dispatch) rate_limit_headers: Vec<(String, String)>, pub(in crate::dispatch) rate_limit_header_updates: Option, pub(in crate::dispatch) turn_state_update: Option, @@ -402,6 +509,10 @@ pub(in crate::dispatch) async fn finalize_live_response_stream( body: String, first_token_ms: Option, ) { + context + .cloudflare + .capture_set_cookie_headers(&context.account_id, &context.set_cookie_headers) + .await; let rate_limit_headers = live_response_rate_limit_headers(&context).await; context .account_pool diff --git a/backend/src/dispatch/upstream_call.rs b/backend/src/dispatch/upstream_call.rs index 1eace9faa..1e9cdea90 100644 --- a/backend/src/dispatch/upstream_call.rs +++ b/backend/src/dispatch/upstream_call.rs @@ -1,14 +1,11 @@ //! 调度层共享的账号级 Codex 上游调用。 -use std::{future::Future, time::Instant}; +use std::time::Instant; use serde_json::Value; use crate::{ - dispatch::{ - affinity::AccountIdentityService, errors::is_retryable_upstream_5xx_error, - recovery::cloudflare::CloudflareRecovery, - }, + dispatch::{affinity::AccountIdentityService, recovery::cloudflare::CloudflareRecovery}, fleet::{ account::Account, pool::{AccountLease, AccountPoolService}, @@ -184,7 +181,7 @@ pub(crate) async fn create_response_with_account( .await } -async fn create_response_stream_with_account( +pub(crate) async fn create_response_stream_with_account( context: AccountUpstreamContext<'_>, request: &CodexResponsesRequest, ) -> Result { @@ -222,97 +219,3 @@ async fn create_response_stream_with_account( ) .await } - -const MAX_UPSTREAM_5XX_RETRIES_PER_ACCOUNT: usize = 2; - -async fn retry_upstream_5xx( - request_id: &str, - account_id: &str, - endpoint: Option<&str>, - retry_message: &'static str, - mut operation: F, -) -> Result -where - F: FnMut() -> Fut, - Fut: Future>, -{ - let mut retries = 0; - loop { - let result = operation().await; - match result { - Err(error) - if is_retryable_upstream_5xx_error(&error) - && retries < MAX_UPSTREAM_5XX_RETRIES_PER_ACCOUNT => - { - log_upstream_retry( - request_id, - account_id, - endpoint, - retries + 1, - &error, - retry_message, - ); - retries += 1; - } - Ok(response) => return Ok(response), - Err(error) => return Err(error), - } - } -} - -fn log_upstream_retry( - request_id: &str, - account_id: &str, - endpoint: Option<&str>, - retry: usize, - error: &CodexClientError, - message: &'static str, -) { - if let Some(endpoint) = endpoint { - tracing::debug!( - request_id, - account_id = %account_id, - endpoint, - retry, - error = %error, - message - ); - } else { - tracing::debug!( - request_id, - account_id = %account_id, - retry, - error = %error, - message - ); - } -} - -pub(crate) async fn create_response_with_account_retrying_5xx( - context: AccountUpstreamContext<'_>, - request: &CodexResponsesRequest, - started_at: Instant, -) -> Result { - retry_upstream_5xx( - context.request_id, - &context.account.id, - None, - "upstream response request failed with retryable 5xx", - || create_response_with_account(context, request, started_at), - ) - .await -} - -pub(crate) async fn create_response_stream_with_account_retrying_5xx( - context: AccountUpstreamContext<'_>, - request: &CodexResponsesRequest, -) -> Result { - retry_upstream_5xx( - context.request_id, - &context.account.id, - None, - "upstream response stream request failed with retryable 5xx", - || create_response_stream_with_account(context, request), - ) - .await -} diff --git a/backend/src/fleet/cookies/store.rs b/backend/src/fleet/cookies/store.rs index ed5602da9..a87ae0c65 100644 --- a/backend/src/fleet/cookies/store.rs +++ b/backend/src/fleet/cookies/store.rs @@ -321,6 +321,8 @@ impl CloudflarePathBlockTracker { /// 记录一次 path-block 失败，并返回当前未过期次数。 pub async fn record_path_block(&self, account_id: &str, now: DateTime) -> u32 { let mut counts = self.counts.write().await; + counts + .retain(|_, state| now.signed_duration_since(state.last_at) <= PATH_BLOCK_STALE_AFTER); let count = counts .get(account_id) .filter(|state| now.signed_duration_since(state.last_at) <= PATH_BLOCK_STALE_AFTER) @@ -377,6 +379,9 @@ impl CloudflareChallengeCooldownTracker { ) -> CloudflareChallengeCooldown { let challenge_count = { let mut states = self.states.write().await; + states.retain(|_, state| { + now.signed_duration_since(state.updated_at) <= CHALLENGE_STALE_AFTER + }); let challenge_count = states .get(account_id) .filter(|state| { diff --git a/backend/src/fleet/manage/oauth.rs b/backend/src/fleet/manage/oauth.rs index a6f7e8b53..31c333cb2 100644 --- a/backend/src/fleet/manage/oauth.rs +++ b/backend/src/fleet/manage/oauth.rs @@ -17,6 +17,7 @@ const OPENAI_OAUTH_AUTHORIZE_ENDPOINT: &str = "https://auth.openai.com/oauth/aut const OPENAI_OAUTH_REDIRECT_URI: &str = "http://localhost:1455/auth/callback"; const OPENAI_OAUTH_SCOPE: &str = "openid profile email offline_access"; const OAUTH_SESSION_TTL_MINUTES: i64 = 30; +const MAX_PENDING_OAUTH_SESSIONS: usize = 1_024; #[derive(Clone)] pub struct AccountOAuthService { @@ -51,6 +52,14 @@ impl AccountOAuthService { let mut sessions = self.sessions.lock().await; sessions.retain(|_, session| session.expires_at > now); + if sessions.len() >= MAX_PENDING_OAUTH_SESSIONS + && let Some(oldest_id) = sessions + .iter() + .min_by_key(|(_, session)| session.expires_at) + .map(|(session_id, _)| session_id.clone()) + { + sessions.remove(&oldest_id); + } sessions.insert(session_id.clone(), session); drop(sessions); diff --git a/backend/src/fleet/quota/runtime.rs b/backend/src/fleet/quota/runtime.rs index c7855db3b..d59247e89 100644 --- a/backend/src/fleet/quota/runtime.rs +++ b/backend/src/fleet/quota/runtime.rs @@ -1,6 +1,10 @@ //! 运行时 quota refresh 业务服务。 -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + time::Duration, +}; use chrono::Utc; use thiserror::Error; @@ -8,7 +12,7 @@ use tokio::time::{Instant, sleep}; use tracing::warn; use crate::fleet::{ - account::AccountStatus, + account::{Account, AccountStatus}, pool::AccountPoolService, quota::{ quota_from_usage, quota_snapshot_limit_reached, quota_snapshot_limit_window_seconds, @@ -120,13 +124,16 @@ impl QuotaRefreshService { let now = Instant::now(); let now_utc = Utc::now(); + let candidate_ids = accounts + .iter() + .filter(|account| quota_refresh_candidate(account)) + .map(|account| account.id.as_str()) + .collect::>(); + last_refreshed.retain(|account_id, _| candidate_ids.contains(account_id.as_str())); + let mut candidates = accounts .into_iter() - .filter(|account| { - account.status == AccountStatus::QuotaExhausted - || (account.status == AccountStatus::Active - && (account.quota_limit_reached || account.quota_verify_required)) - }) + .filter(quota_refresh_candidate) .peekable(); while let Some(account) = candidates.next() { @@ -245,6 +252,12 @@ impl QuotaRefreshService { } } +fn quota_refresh_candidate(account: &Account) -> bool { + account.status == AccountStatus::QuotaExhausted + || (account.status == AccountStatus::Active + && (account.quota_limit_reached || account.quota_verify_required)) +} + fn quota_cooldown_ready(cooldown_until: chrono::DateTime, now: chrono::DateTime) -> bool { cooldown_until .checked_add_signed(chrono::Duration::seconds(COOLDOWN_REFRESH_GRACE_SECS)) diff --git a/backend/src/telemetry/recorder.rs b/backend/src/telemetry/recorder.rs index 43cafd4b3..c8c70fd6d 100644 --- a/backend/src/telemetry/recorder.rs +++ b/backend/src/telemetry/recorder.rs @@ -20,6 +20,8 @@ use crate::{ upstream::openai::protocol::responses::CodexResponsesRequest, }; +const MAX_CAPTURED_BODY_BYTES: usize = 4 * 1024; + /// 成功与失败事实的唯一写入入口。 #[derive(Clone)] pub struct Recorder { @@ -83,6 +85,10 @@ impl Recorder { self.ops_errors.append(&event).await?; Ok(()) } + + pub(crate) fn captures_body(&self) -> bool { + self.capture_body + } } pub(crate) struct DispatchErrorLogRecord<'a> { @@ -175,27 +181,55 @@ fn is_usage_fact(event: &UsageRecord) -> bool { } fn apply_success_capture_body_policy(event: &mut UsageRecord, capture_body: bool) { - if capture_body { - return; + if !capture_body { + remove_body_fields(&mut event.metadata); + } else { + limit_body_fields(&mut event.metadata); } - remove_body_fields(&mut event.metadata); } fn apply_error_capture_body_policy(event: &mut OpsErrorLog, capture_body: bool) { - let Some(metadata) = event.metadata.as_object_mut() else { + if !capture_body { + remove_body_fields(&mut event.metadata); + } else { + limit_body_fields(&mut event.metadata); + } +} + +fn limit_body_fields(metadata: &mut Value) { + let Some(metadata) = metadata.as_object_mut() else { return; }; for key in body_fields() { - if capture_body { - if let Some(Value::String(value)) = metadata.get_mut(key) { - value.truncate(4096); + let Some(value) = metadata.get_mut(key) else { + continue; + }; + match value { + Value::String(value) => truncate_utf8(value, MAX_CAPTURED_BODY_BYTES), + value => { + let Ok(mut encoded) = serde_json::to_string(value) else { + continue; + }; + if encoded.len() > MAX_CAPTURED_BODY_BYTES { + truncate_utf8(&mut encoded, MAX_CAPTURED_BODY_BYTES); + *value = Value::String(encoded); + } } - } else { - metadata.remove(key); } } } +fn truncate_utf8(value: &mut String, max_bytes: usize) { + if value.len() <= max_bytes { + return; + } + let mut end = max_bytes; + while !value.is_char_boundary(end) { + end -= 1; + } + value.truncate(end); +} + fn remove_body_fields(metadata: &mut Value) { let Some(metadata) = metadata.as_object_mut() else { return; diff --git a/backend/src/upstream/openai/protocol/responses.rs b/backend/src/upstream/openai/protocol/responses.rs index 9c26997e0..8c727235b 100644 --- a/backend/src/upstream/openai/protocol/responses.rs +++ b/backend/src/upstream/openai/protocol/responses.rs @@ -1,4 +1,4 @@ -use std::{collections::BTreeSet, time::Instant}; +use std::{collections::BTreeSet, sync::Arc, time::Instant}; use serde::Serialize; use serde_json::{Map, Value}; @@ -68,6 +68,8 @@ pub struct CodexResponsesRequest { pub parent_thread_id: Option, /// 已知 previous response 的持久化范围，仅用于本地 transport 校验。 pub previous_response_scope: Option, + /// 当前下游 WebSocket 持有的完整历史前缀，仅用于连接内换号恢复。 + pub local_replay_input: Option>>, /// 流式响应在交给下游前的本地提交策略。 pub stream_commit_policy: StreamCommitPolicy, } @@ -537,6 +539,7 @@ impl CodexResponsesRequest { codex_window_id: None, parent_thread_id: None, previous_response_scope: None, + local_replay_input: None, stream_commit_policy: StreamCommitPolicy::FirstForwardableEvent, } } diff --git a/backend/src/upstream/openai/transport/client.rs b/backend/src/upstream/openai/transport/client.rs index 9b1311468..f146f1dfc 100644 --- a/backend/src/upstream/openai/transport/client.rs +++ b/backend/src/upstream/openai/transport/client.rs @@ -9,7 +9,7 @@ use std::{ use async_trait::async_trait; use bytes::Bytes; -use futures::{Stream, TryStreamExt}; +use futures::{Stream, StreamExt, TryStreamExt}; use reqwest::{ Client, Response as ReqwestResponse, StatusCode, header::{ @@ -53,6 +53,8 @@ use super::websocket_pool::{ // --------------------------------------------------------------------------- const MAX_UPSTREAM_ERROR_BODY_BYTES: usize = 1024 * 1024; +const UPSTREAM_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); +const UPSTREAM_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); const X_CODEX_WS_STREAM_REQUEST_START_MS_CLIENT_METADATA_KEY: &str = "x-codex-ws-stream-request-start-ms"; type ReqwestClientCacheKey = (bool, Option); @@ -75,6 +77,7 @@ pub fn build_reqwest_client(force_http11: bool) -> Result .use_rustls_tls() .no_proxy() .pool_max_idle_per_host(4) + .connect_timeout(UPSTREAM_CONNECT_TIMEOUT) .tcp_keepalive(Duration::from_secs(30)) .gzip(true) .brotli(true) @@ -114,6 +117,12 @@ pub enum CodexClientError { /// SSE 响应解析失败。 #[error("invalid upstream SSE response: {0}")] InvalidSse(#[from] SseError), + /// HTTP/SSE 上游在空闲窗口内没有发送任何数据。 + #[error("upstream HTTP/SSE stream idle for {timeout:?}")] + StreamIdleTimeout { + /// 相邻数据块允许的最大空闲时间。 + timeout: Duration, + }, /// WebSocket 请求编码失败。 #[error("failed to encode websocket request: {0}")] WebSocketEncode(#[source] serde_json::Error), diff --git a/backend/src/upstream/openai/transport/client_sse.rs b/backend/src/upstream/openai/transport/client_sse.rs index 053381a98..881c6cf82 100644 --- a/backend/src/upstream/openai/transport/client_sse.rs +++ b/backend/src/upstream/openai/transport/client_sse.rs @@ -72,7 +72,7 @@ impl CodexBackendClient { let mut body_bytes = Vec::new(); let mut first_token_ms = None; - let mut stream = response.bytes_stream(); + let mut stream = http_sse_stream(response); while let Some(chunk) = stream.try_next().await? { body_bytes.extend_from_slice(&chunk); response_meta::update_first_token_ms(started_at, &body_bytes, &mut first_token_ms); @@ -128,7 +128,7 @@ impl CodexBackendClient { } Ok(CodexBackendStreamingResponse { - body: Box::pin(response.bytes_stream().map_err(CodexClientError::Http)), + body: http_sse_stream(response), transport: CodexBackendTransport::HttpSse, turn_state, set_cookie_headers, @@ -590,3 +590,21 @@ impl CodexBackendClient { Ok(headers) } } + +fn http_sse_stream(response: ReqwestResponse) -> CodexBackendSseStream { + let stream: CodexBackendSseStream = + Box::pin(response.bytes_stream().map_err(CodexClientError::Http)); + Box::pin(futures::stream::unfold(Some(stream), |stream| async move { + let mut stream = stream?; + match tokio::time::timeout(UPSTREAM_STREAM_IDLE_TIMEOUT, stream.next()).await { + Ok(Some(chunk)) => Some((chunk, Some(stream))), + Ok(None) => None, + Err(_) => Some(( + Err(CodexClientError::StreamIdleTimeout { + timeout: UPSTREAM_STREAM_IDLE_TIMEOUT, + }), + None, + )), + } + })) +} diff --git a/backend/src/upstream/openai/transport/websocket.rs b/backend/src/upstream/openai/transport/websocket.rs index b69d69161..30eaab2c6 100644 --- a/backend/src/upstream/openai/transport/websocket.rs +++ b/backend/src/upstream/openai/transport/websocket.rs @@ -53,8 +53,9 @@ const CODEX_RESPONSES_PATH: &str = "/codex/responses"; const WEBSOCKET_EXTENSIONS: &str = "permessage-deflate; client_max_window_bits"; const WEBSOCKET_RECEIVE_IDLE_TIMEOUT: Duration = Duration::from_secs(20); const WEBSOCKET_ACTIVE_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(5 * 60); +const WEBSOCKET_CONNECT_TIMEOUT: Duration = Duration::from_secs(15); +const WEBSOCKET_SEND_TIMEOUT: Duration = WEBSOCKET_ACTIVE_STREAM_IDLE_TIMEOUT; const WEBSOCKET_STREAM_BUFFER: usize = 16; -const WEBSOCKET_FIRST_TOKEN_FRESH_RETRY_ATTEMPTS: usize = 2; /// WebSocket audit artifact 输出目录环境变量。 pub const WS_AUDIT_DIR_ENV: &str = "CODEX_PROXY_WS_AUDIT_DIR"; @@ -188,10 +189,9 @@ pub use frames::{ }; use frames::{ WebSocketStreamPoolReturn, WebSocketTerminalKind, audit_header_value, - collect_websocket_response, is_initial_event_timeout, - prefetch_stream_frames_until_output_or_terminal, reusable_websocket_metadata, - reused_stream_prefetch_error, stream_websocket_response, websocket_audit_file_name, - websocket_connection_metadata, + collect_websocket_response, prefetch_stream_frames_until_output_or_terminal, + reusable_websocket_metadata, reused_stream_prefetch_error, stream_websocket_response, + websocket_audit_file_name, websocket_connection_metadata, }; impl CodexWebSocketRequest { @@ -327,13 +327,36 @@ async fn connect_websocket( let connector = super::tls::maybe_build_rustls_client_config_with_custom_ca() .map_err(|error| tungstenite::Error::Io(std::io::Error::other(error)))? .map(Connector::Rustls); - match connect_async_tls_with_config(request, Some(websocket_config()), false, connector).await { + let result = timeout( + WEBSOCKET_CONNECT_TIMEOUT, + connect_async_tls_with_config(request, Some(websocket_config()), false, connector), + ) + .await + .map_err(|_| CodexWebSocketExchangeError::ConnectTimeout { + timeout: WEBSOCKET_CONNECT_TIMEOUT, + })?; + match result { Ok((websocket, response)) => Ok((websocket, response)), Err(tungstenite::Error::Http(response)) => Err(websocket_opening_error(response.as_ref())), Err(error) => Err(error.into()), } } +async fn send_websocket_request( + websocket: &PumpedWebSocket, + payload_text: &str, +) -> Result<(), CodexWebSocketExchangeError> { + timeout( + WEBSOCKET_SEND_TIMEOUT, + websocket.send(Message::Text(payload_text.to_string().into())), + ) + .await + .map_err(|_| CodexWebSocketExchangeError::SendTimeout { + timeout: WEBSOCKET_SEND_TIMEOUT, + })??; + Ok(()) +} + /// 建立连接并交给后台 pump 托管；`keepalive` 决定该连接是否做主动保活。 /// /// 非池化（即用即弃）连接用 [`PumpKeepalive::disabled`]；池化连接用连接池派生的保活策略， @@ -420,9 +443,7 @@ async fn execute_fresh_response_create_request( ) -> Result { let (websocket, response) = connect_pumped_websocket(request.connection(), PumpKeepalive::disabled()).await?; - websocket - .send(Message::Text(request.payload_text().to_string().into())) - .await?; + send_websocket_request(&websocket, request.payload_text()).await?; let metadata = websocket_connection_metadata(&response); let (exchange, websocket, _metadata, _continuation, _terminal) = collect_websocket_response( @@ -444,9 +465,7 @@ async fn execute_fresh_response_create_request_stream( ) -> Result { let (mut websocket, response) = connect_pumped_websocket(request.connection(), PumpKeepalive::disabled()).await?; - websocket - .send(Message::Text(request.payload_text().to_string().into())) - .await?; + send_websocket_request(&websocket, request.payload_text()).await?; let mut metadata = websocket_connection_metadata(&response); let prefetched_frames = if should_prefetch_until_output_or_terminal(request) { @@ -475,41 +494,6 @@ async fn execute_fresh_response_create_request_stream( )) } -async fn execute_fresh_response_create_request_with_retries( - request: &CodexWebSocketRequest, - started_at: Instant, - initial_event_timeout: Option, - attempts: usize, -) -> Result { - let attempts = attempts.max(1); - for attempt in 1..=attempts { - match execute_fresh_response_create_request(request, started_at, initial_event_timeout) - .await - { - Ok(exchange) => return Ok(exchange), - Err(error) if attempt < attempts && is_initial_event_timeout(&error) => continue, - Err(error) => return Err(error), - } - } - unreachable!("fresh websocket retry loop always returns"); -} - -async fn execute_fresh_response_create_request_stream_with_retries( - request: &CodexWebSocketRequest, - initial_event_timeout: Option, - attempts: usize, -) -> Result { - let attempts = attempts.max(1); - for attempt in 1..=attempts { - match execute_fresh_response_create_request_stream(request, initial_event_timeout).await { - Ok(exchange) => return Ok(exchange), - Err(error) if attempt < attempts && is_initial_event_timeout(&error) => continue, - Err(error) => return Err(error), - } - } - unreachable!("fresh websocket stream retry loop always returns"); -} - // --------------------------------------------------------------------------- // Pool-aware execution // --------------------------------------------------------------------------- @@ -526,11 +510,10 @@ pub(crate) async fn execute_response_create_request_with_pool( PreviousResponseUnavailableReason::PoolUnavailable, )); } - return execute_fresh_response_create_request_with_retries( + return execute_fresh_response_create_request( request, started_at, fallback_initial_event_timeout, - WEBSOCKET_FIRST_TOKEN_FRESH_RETRY_ATTEMPTS, ) .await; }; @@ -564,11 +547,10 @@ pub(crate) async fn execute_response_create_request_with_pool( PreviousResponseUnavailableReason::ReusedConnectionLost, )); } - let mut exchange = execute_fresh_response_create_request_with_retries( + let mut exchange = execute_fresh_response_create_request( request, started_at, pool.initial_event_timeout(), - WEBSOCKET_FIRST_TOKEN_FRESH_RETRY_ATTEMPTS, ) .await?; exchange.pool_decision = Some(WebSocketPoolDecision::retry_after_stale_reuse()); @@ -591,24 +573,10 @@ pub(crate) async fn execute_response_create_request_with_pool( let result = execute_fresh_pooled_response_create_request(request, pool, lease, started_at) .await; - match result { - Ok(mut exchange) => { - exchange.pool_decision = Some(WebSocketPoolDecision::new()); - Ok(exchange) - } - Err(error) if is_initial_event_timeout(&error) => { - let mut exchange = execute_fresh_response_create_request_with_retries( - request, - started_at, - pool.initial_event_timeout(), - 1, - ) - .await?; - exchange.pool_decision = Some(WebSocketPoolDecision::new()); - Ok(exchange) - } - Err(error) => Err(error), - } + result.map(|mut exchange| { + exchange.pool_decision = Some(WebSocketPoolDecision::new()); + exchange + }) } WebSocketPoolAcquire::Bypass(reason) => { if !request.continuation().permits_fresh_connection() { @@ -622,11 +590,10 @@ pub(crate) async fn execute_response_create_request_with_pool( } })); } - let mut exchange = execute_fresh_response_create_request_with_retries( + let mut exchange = execute_fresh_response_create_request( request, started_at, pool.initial_event_timeout(), - WEBSOCKET_FIRST_TOKEN_FRESH_RETRY_ATTEMPTS, ) .await?; exchange.pool_decision = Some(WebSocketPoolDecision::bypass(reason)); @@ -646,10 +613,9 @@ pub(crate) async fn execute_response_create_request_stream_with_pool( PreviousResponseUnavailableReason::PoolUnavailable, )); } - return execute_fresh_response_create_request_stream_with_retries( + return execute_fresh_response_create_request_stream( request, fallback_initial_event_timeout, - WEBSOCKET_FIRST_TOKEN_FRESH_RETRY_ATTEMPTS, ) .await; }; @@ -678,10 +644,9 @@ pub(crate) async fn execute_response_create_request_stream_with_pool( PreviousResponseUnavailableReason::ReusedConnectionLost, )); } - let mut exchange = execute_fresh_response_create_request_stream_with_retries( + let mut exchange = execute_fresh_response_create_request_stream( request, pool.initial_event_timeout(), - WEBSOCKET_FIRST_TOKEN_FRESH_RETRY_ATTEMPTS, ) .await?; exchange.pool_decision = Some(WebSocketPoolDecision::retry_after_stale_reuse()); @@ -703,23 +668,10 @@ pub(crate) async fn execute_response_create_request_stream_with_pool( } let result = execute_fresh_pooled_response_create_request_stream(request, pool, lease).await; - match result { - Ok(mut exchange) => { - exchange.pool_decision = Some(WebSocketPoolDecision::new()); - Ok(exchange) - } - Err(error) if is_initial_event_timeout(&error) => { - let mut exchange = execute_fresh_response_create_request_stream_with_retries( - request, - pool.initial_event_timeout(), - 1, - ) - .await?; - exchange.pool_decision = Some(WebSocketPoolDecision::new()); - Ok(exchange) - } - Err(error) => Err(error), - } + result.map(|mut exchange| { + exchange.pool_decision = Some(WebSocketPoolDecision::new()); + exchange + }) } WebSocketPoolAcquire::Bypass(reason) => { if !request.continuation().permits_fresh_connection() { @@ -733,12 +685,9 @@ pub(crate) async fn execute_response_create_request_stream_with_pool( } })); } - let mut exchange = execute_fresh_response_create_request_stream_with_retries( - request, - pool.initial_event_timeout(), - WEBSOCKET_FIRST_TOKEN_FRESH_RETRY_ATTEMPTS, - ) - .await?; + let mut exchange = + execute_fresh_response_create_request_stream(request, pool.initial_event_timeout()) + .await?; exchange.pool_decision = Some(WebSocketPoolDecision::bypass(reason)); Ok(exchange) } @@ -765,12 +714,9 @@ async fn execute_fresh_pooled_response_create_request( return Err(error); } }; - if let Err(error) = websocket - .send(Message::Text(request.payload_text().to_string().into())) - .await - { + if let Err(error) = send_websocket_request(&websocket, request.payload_text()).await { lease.discard().await; - return Err(error.into()); + return Err(error); } let now = tokio::time::Instant::now(); @@ -823,12 +769,9 @@ async fn execute_fresh_pooled_response_create_request_stream( return Err(error); } }; - if let Err(error) = websocket - .send(Message::Text(request.payload_text().to_string().into())) - .await - { + if let Err(error) = send_websocket_request(&websocket, request.payload_text()).await { lease.discard().await; - return Err(error.into()); + return Err(error); } let now = tokio::time::Instant::now(); @@ -871,10 +814,7 @@ async fn execute_pooled_response_create_request( connection: PooledWebSocketConnection, started_at: Instant, ) -> Result { - if let Err(error) = connection - .websocket - .send(Message::Text(request.payload_text().to_string().into())) - .await + if let Err(error) = send_websocket_request(&connection.websocket, request.payload_text()).await { lease.discard().await; return Err( @@ -931,10 +871,7 @@ async fn execute_pooled_response_create_request_stream( let mut metadata = reusable_websocket_metadata(connection.metadata); let created_at = connection.created_at; let continuation = connection.continuation; - if let Err(error) = websocket - .send(Message::Text(request.payload_text().to_string().into())) - .await - { + if let Err(error) = send_websocket_request(&websocket, request.payload_text()).await { lease.discard().await; return Err( CodexWebSocketExchangeError::ReusedConnectionDiedBeforeFirstOutput { diff --git a/backend/src/upstream/openai/transport/websocket_frames.rs b/backend/src/upstream/openai/transport/websocket_frames.rs index f9221149e..5ec180737 100644 --- a/backend/src/upstream/openai/transport/websocket_frames.rs +++ b/backend/src/upstream/openai/transport/websocket_frames.rs @@ -73,6 +73,18 @@ pub enum CodexWebSocketExchangeError { /// WebSocket 传输失败。 #[error("websocket transport error: {0}")] Transport(#[from] tungstenite::Error), + /// DNS、TCP、TLS 或 WebSocket upgrade 未在限定时间内完成。 + #[error("websocket connect timed out after {timeout:?}")] + ConnectTimeout { + /// 建连超时时长。 + timeout: Duration, + }, + /// 请求帧未在限定时间内写入上游连接。 + #[error("websocket request send timed out after {timeout:?}")] + SendTimeout { + /// 发送超时时长。 + timeout: Duration, + }, /// SSE 聚合结果无法解析。 #[error("invalid websocket SSE response: {0}")] InvalidSse(#[from] SseError), @@ -102,14 +114,14 @@ pub enum CodexWebSocketExchangeError { /// 上游返回非文本事件帧。 #[error("unexpected binary websocket event")] UnexpectedBinaryEvent, - /// 复用的池连接在首个真实输出前失效。 - #[error("reused websocket connection died before first output: {message}")] + /// 复用的池连接在收到首个上游事件前失效。 + #[error("reused websocket connection died before first upstream event: {message}")] ReusedConnectionDiedBeforeFirstOutput { /// 底层失效原因。 message: String, }, - /// 首个真实输出在配置的绝对超时内未到达（连接落到病态上游后端）。 - #[error("websocket first output not received within {timeout:?}")] + /// 建连并发送后，上游在配置时间内没有产生任何事件。 + #[error("websocket first upstream event not received within {timeout:?}")] InitialEventTimeout { /// 首个上游事件超时时长。 timeout: Duration, @@ -748,13 +760,6 @@ fn receive_idle_timeout( } } -pub(super) fn is_initial_event_timeout(error: &CodexWebSocketExchangeError) -> bool { - matches!( - error, - CodexWebSocketExchangeError::InitialEventTimeout { .. } - ) -} - async fn next_websocket_message( websocket: &mut PumpedWebSocket, receive_timeout: Duration, diff --git a/backend/src/upstream/openai/transport/websocket_pool.rs b/backend/src/upstream/openai/transport/websocket_pool.rs index 32e010041..8aecac68e 100644 --- a/backend/src/upstream/openai/transport/websocket_pool.rs +++ b/backend/src/upstream/openai/transport/websocket_pool.rs @@ -85,7 +85,7 @@ pub struct CodexWebSocketPoolConfig { pub ping_timeout: Duration, /// idle socket 无活动多久后视为失活。 pub liveness_timeout: Option, - /// 首个上游事件到达前的超时；`None` 表示禁用首事件超时。 + /// 建连并发送后首个上游事件到达前的超时；`None` 表示禁用。 pub initial_event_timeout: Option, } @@ -144,7 +144,7 @@ impl CodexWebSocketPool { self.config.keepalive() } - /// 首个上游事件到达前的超时；`None` 表示禁用首事件超时。 + /// 建连并发送后首个上游事件到达前的超时；`None` 表示禁用。 pub(crate) fn initial_event_timeout(&self) -> Option { self.config.initial_event_timeout } diff --git a/backend/tests/api/admin/accounts_routes/lifecycle_routes.rs b/backend/tests/api/admin/accounts_routes/lifecycle_routes.rs index e543174d8..dfc5e53a9 100644 --- a/backend/tests/api/admin/accounts_routes/lifecycle_routes.rs +++ b/backend/tests/api/admin/accounts_routes/lifecycle_routes.rs @@ -32,7 +32,6 @@ async fn admin_accounts_lifecycle_should_update_and_delete_accounts() { function_call_ids: Vec::new(), variant_hash: None, continuation_scope: codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::ConnectionLocal, - replay: None, created_at: chrono::Utc::now(), }, chrono::Duration::hours(1), diff --git a/backend/tests/api/client/responses/websocket.rs b/backend/tests/api/client/responses/websocket.rs index 770f952f3..95dea7360 100644 --- a/backend/tests/api/client/responses/websocket.rs +++ b/backend/tests/api/client/responses/websocket.rs @@ -121,6 +121,82 @@ async fn responses_websocket_should_forward_multiple_response_create_requests_on assert_eq!(upstream_requests[1]["type"], "response.create"); } +#[tokio::test] +async fn responses_websocket_should_replay_connection_history_after_previous_response_not_found() { + let upstream_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let upstream_base_url = format!("http://{}", upstream_listener.local_addr().unwrap()); + let upstream = tokio::spawn(async move { + let (first_stream, _) = upstream_listener.accept().await.unwrap(); + let mut first = accept_upstream_websocket(first_stream).await.unwrap(); + let initial = receive_upstream_request(&mut first).await; + send_upstream_response(&mut first, "resp_first", "first answer").await; + + let continued = receive_upstream_request(&mut first).await; + first + .send(Message::Text( + json!({ + "type": "response.failed", + "response": { + "id": "resp_missing", + "status": "failed", + "error": { + "code": "previous_response_not_found", + "message": "Previous response with id resp_first was not found", + }, + }, + }) + .to_string() + .into(), + )) + .await + .unwrap(); + first.close(None).await.unwrap(); + + let (replay_stream, _) = upstream_listener.accept().await.unwrap(); + let mut replay = accept_upstream_websocket(replay_stream).await.unwrap(); + let replayed = receive_upstream_request(&mut replay).await; + send_upstream_response(&mut replay, "resp_recovered", "recovered").await; + (initial, continued, replayed) + }); + let (app, api_key, _dir) = + crate::dispatch::service::test_app_with_account(upstream_base_url).await; + let server = spawn_app(app).await; + let mut websocket = connect_responses_websocket(server.address, &api_key).await; + + let mut first_payload: Value = serde_json::from_str(&response_create_payload("first")).unwrap(); + first_payload["input"][0]["id"] = json!("client_input_id"); + first_payload["input"][0]["encrypted_content"] = json!("client_secret"); + websocket + .send(Message::Text(first_payload.to_string().into())) + .await + .unwrap(); + let first = receive_response(&mut websocket).await; + assert_eq!(first.last().unwrap()["response"]["id"], "resp_first"); + + let mut second_payload: Value = + serde_json::from_str(&response_create_payload("second")).unwrap(); + second_payload["previous_response_id"] = json!("resp_first"); + websocket + .send(Message::Text(second_payload.to_string().into())) + .await + .unwrap(); + let second = receive_response(&mut websocket).await; + let (initial, continued, replayed) = upstream.await.unwrap(); + + assert_eq!(second.last().unwrap()["response"]["id"], "resp_recovered"); + assert_eq!(initial["input"][0]["content"], "first"); + assert_eq!(continued["previous_response_id"], "resp_first"); + assert_eq!(continued["input"][0]["content"], "second"); + assert!(replayed.get("previous_response_id").is_none()); + assert_eq!(replayed["input"][0]["content"], "first"); + assert_eq!(replayed["input"][1]["role"], "assistant"); + assert_eq!(replayed["input"][2]["content"], "second"); + let replayed_json = replayed.to_string(); + assert!(!replayed_json.contains("client_input_id")); + assert!(!replayed_json.contains("client_secret")); + assert!(!replayed_json.contains("assistant_output_resp_first")); +} + #[tokio::test] async fn responses_websocket_should_allow_official_retry_after_midstream_upstream_disconnect() { let upstream_listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -237,6 +313,12 @@ async fn send_upstream_response( "id": response_id, "object": "response", "status": "completed", + "output": [{ + "type": "message", + "id": format!("assistant_output_{response_id}"), + "role": "assistant", + "content": [{"type": "output_text", "text": text}], + }], "usage": { "input_tokens": 3, "output_tokens": 1, @@ -252,6 +334,11 @@ async fn send_upstream_response( } } +async fn receive_upstream_request(websocket: &mut WebSocketStream) -> Value { + let request = websocket.next().await.unwrap().unwrap(); + serde_json::from_str(&request.into_text().unwrap()).unwrap() +} + async fn receive_response(websocket: &mut WebSocketStream) -> Vec where S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, diff --git a/backend/tests/dispatch/affinity/mod.rs b/backend/tests/dispatch/affinity/mod.rs index 299945b86..64f10b4b9 100644 --- a/backend/tests/dispatch/affinity/mod.rs +++ b/backend/tests/dispatch/affinity/mod.rs @@ -1,10 +1,8 @@ use chrono::{Duration, Utc}; use codex_proxy_rs::dispatch::affinity::{ - MAX_REPLAY_DEPTH, MAX_REPLAY_SESSION_BYTES, RedisSessionAffinityStore, ResponseReplaySnapshot, - SessionAffinityEntry, + MAX_CONVERSATION_AFFINITIES, RedisSessionAffinityStore, SessionAffinityEntry, }; use redis::AsyncCommands; -use serde_json::json; use crate::support::storage::create_test_redis; @@ -18,9 +16,10 @@ async fn redis_affinity_store_upserts_reads_and_forgets_entries() { .upsert("resp_1", &entry, Duration::hours(4)) .await .unwrap(); + let mut connection = redis.manager(); let ttl_seconds: i64 = connection - .ttl(redis.key("affinity:v2:resp:resp_1")) + .ttl(redis.key("affinity:v3:resp:resp_1")) .await .unwrap(); assert!((3 * 60 * 60..=4 * 60 * 60).contains(&ttl_seconds)); @@ -96,8 +95,9 @@ async fn redis_affinity_store_returns_latest_matching_conversation_variant() { } #[tokio::test] -async fn redis_affinity_store_forgets_all_entries_for_account() { - let store = RedisSessionAffinityStore::new(create_test_redis("affinity-account").await); +async fn redis_affinity_store_forgets_account_entries_in_one_operation() { + let redis = create_test_redis("affinity-account").await; + let store = RedisSessionAffinityStore::new(redis.clone()); let now = Utc::now(); for response_id in ["resp_1", "resp_2"] { store @@ -110,7 +110,14 @@ async fn redis_affinity_store_forgets_all_entries_for_account() { .unwrap(); } + let bytes_before: u64 = redis::cmd("GET") + .arg(redis.key("affinity:v3:global:bytes")) + .query_async(&mut redis.manager()) + .await + .unwrap(); + assert!(bytes_before > 0); assert_eq!(store.forget_account("acct_delete").await.unwrap(), 2); + for response_id in ["resp_1", "resp_2"] { assert!( store @@ -120,99 +127,44 @@ async fn redis_affinity_store_forgets_all_entries_for_account() { .is_none() ); } + let bytes_after: Option = redis::cmd("GET") + .arg(redis.key("affinity:v3:global:bytes")) + .query_async(&mut redis.manager()) + .await + .unwrap(); + assert!(bytes_after.is_none()); } #[tokio::test] -async fn redis_affinity_store_should_keep_replay_growth_linear_across_turns() { - let redis = create_test_redis("affinity-replay-linear").await; +async fn redis_affinity_store_caps_active_conversation_index() { + let redis = create_test_redis("affinity-conversation-cap").await; let store = RedisSessionAffinityStore::new(redis.clone()); let now = Utc::now(); - let mut parent_response_id = None; - let mut total_bytes = 0; - for turn in 0..16u16 { - let response_id = format!("resp_{turn}"); - let turn_input = vec![json!({"role": "user", "content": format!("input-{turn}")})]; - let turn_output = vec![json!({"type": "message", "content": format!("output-{turn}")})]; - let node_bytes = replay_node_bytes(&turn_input, &turn_output); - total_bytes += node_bytes; - let mut entry = affinity( - "acct_linear", - "conv_linear", - None, - now + Duration::milliseconds(i64::from(turn)), - ); - entry.replay = Some(ResponseReplaySnapshot { - parent_response_id: parent_response_id.clone(), - turn_input, - turn_output, - depth: turn + 1, - total_bytes, - }); + for index in 0..=MAX_CONVERSATION_AFFINITIES { store - .upsert(&response_id, &entry, Duration::hours(4)) + .upsert( + &format!("resp_cap_{index}"), + &affinity( + "acct_cap", + "conv_cap", + None, + now + Duration::milliseconds(index as i64), + ), + Duration::hours(4), + ) .await .unwrap(); - parent_response_id = Some(response_id); } - let head = store - .get("resp_15", now, Duration::hours(4)) - .await - .unwrap() - .unwrap(); - let replay = store - .replay_input( - "resp_15", - &head, - now, - Duration::hours(4), - MAX_REPLAY_DEPTH, - MAX_REPLAY_SESSION_BYTES, - ) - .await - .unwrap() - .unwrap(); - assert_eq!(replay.len(), 32); - assert_eq!(replay[0]["content"], "input-0"); - assert_eq!(replay[31]["content"], "output-15"); - - let mut connection = redis.manager(); - let head_json: String = connection - .get(redis.key("affinity:v2:resp:resp_15")) + let count: usize = redis::cmd("ZCARD") + .arg(redis.key("affinity:v3:conv:conv_cap")) + .query_async(&mut redis.manager()) .await .unwrap(); - assert!(head_json.contains("input-15")); - assert!(!head_json.contains("input-0")); - assert!(head_json.len() < 1_024); -} - -#[tokio::test] -async fn redis_affinity_store_should_reject_replay_metadata_over_capacity() { - let store = RedisSessionAffinityStore::new(create_test_redis("affinity-replay-limit").await); - let now = Utc::now(); - let mut entry = affinity("acct_limit", "conv_limit", None, now); - entry.replay = Some(ResponseReplaySnapshot { - parent_response_id: None, - turn_input: vec![json!({"role": "user", "content": "small"})], - turn_output: Vec::new(), - depth: 1, - total_bytes: MAX_REPLAY_SESSION_BYTES + 1, - }); - store - .upsert("resp_limit", &entry, Duration::hours(4)) - .await - .unwrap(); - + assert_eq!(count, MAX_CONVERSATION_AFFINITIES); assert!( store - .replay_input( - "resp_limit", - &entry, - now, - Duration::hours(4), - MAX_REPLAY_DEPTH, - MAX_REPLAY_SESSION_BYTES, - ) + .get("resp_cap_0", now, Duration::hours(4)) .await .unwrap() .is_none() @@ -220,11 +172,11 @@ async fn redis_affinity_store_should_reject_replay_metadata_over_capacity() { } #[tokio::test] -async fn redis_affinity_store_should_use_pruned_account_zset_index() { +async fn redis_affinity_store_prunes_expired_account_index_members_lazily() { let redis = create_test_redis("affinity-account-zset").await; let store = RedisSessionAffinityStore::new(redis.clone()); let now = Utc::now(); - let account_key = redis.key("affinity:v2:account:acct_zset"); + let account_key = redis.key("affinity:v3:account:acct_zset"); let mut connection = redis.manager(); let _: usize = connection .zadd( @@ -255,7 +207,7 @@ async fn redis_affinity_store_should_use_pruned_account_zset_index() { } #[tokio::test] -async fn redis_affinity_store_should_not_persist_already_expired_entries() { +async fn redis_affinity_store_does_not_persist_expired_entries() { let store = RedisSessionAffinityStore::new(create_test_redis("affinity-expired-write").await); let now = Utc::now(); store @@ -280,6 +232,20 @@ async fn redis_affinity_store_should_not_persist_already_expired_entries() { ); } +#[tokio::test] +async fn redis_affinity_store_rejects_oversized_metadata() { + let store = RedisSessionAffinityStore::new(create_test_redis("affinity-metadata-limit").await); + let now = Utc::now(); + let mut entry = affinity("acct_limit", "conv_limit", None, now); + entry.function_call_ids = vec!["x".repeat(64 * 1024)]; + + let error = store + .upsert("resp_limit", &entry, Duration::hours(4)) + .await + .unwrap_err(); + assert!(error.to_string().contains("metadata is too large")); +} + fn affinity( account_id: &str, conversation_id: &str, @@ -295,11 +261,6 @@ fn affinity( function_call_ids: vec!["call_1".to_string()], variant_hash: variant_hash.map(ToString::to_string), continuation_scope: codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::ConnectionLocal, - replay: None, created_at, } } - -fn replay_node_bytes(input: &[serde_json::Value], output: &[serde_json::Value]) -> u64 { - serde_json::to_vec(&(input, output)).unwrap().len() as u64 -} diff --git a/backend/tests/dispatch/service.rs b/backend/tests/dispatch/service.rs index 35f43de6d..b1254c5f8 100644 --- a/backend/tests/dispatch/service.rs +++ b/backend/tests/dispatch/service.rs @@ -16,7 +16,7 @@ use codex_proxy_rs::{ api::router, bootstrap::config::AppConfig, bootstrap::services::{Services, UsageRecordOptions}, - dispatch::affinity::{ResponseReplaySnapshot, SessionAffinityEntry}, + dispatch::affinity::SessionAffinityEntry, fleet::{account::AccountStatus, cookies::PgCookieStore}, infra::identity::hash_credential, telemetry::usage::store::{PgUsageRecordStore, UsageRecordFilter}, @@ -63,19 +63,6 @@ fn websocket_accept_config() -> WebSocketConfig { config } -fn replay_snapshot(turn_input: Vec, turn_output: Vec) -> ResponseReplaySnapshot { - let total_bytes = serde_json::to_vec(&(&turn_input, &turn_output)) - .unwrap() - .len() as u64; - ResponseReplaySnapshot { - parent_response_id: None, - turn_input, - turn_output, - depth: 1, - total_bytes, - } -} - pub(crate) async fn accept_async(stream: TcpStream) -> Result, WsError> { accept_hdr_async(stream, |_request, _response| {}).await } @@ -287,32 +274,6 @@ pub(crate) async fn test_app_with_account( (router::router().with_state(state), api_key, dir) } -async fn test_app_with_account_and_state( - base_url: String, -) -> ( - axum::Router, - AppState, - String, - crate::support::storage::TestDatabaseGuard, -) { - let (pool, dir) = init_test_db("openai-chat-upstream-state").await; - let api_key = insert_client_api_key(&pool).await; - insert_account(&pool).await; - let state = test_app_state_with_pool(&test_config(test_database_url(), base_url), pool).await; - state - .services - .account_pool - .restore_from_store() - .await - .expect("account pool should restore"); - ( - router::router().with_state(state.clone()), - state, - api_key, - dir, - ) -} - async fn test_app_with_account_and_pool( base_url: String, ) -> ( @@ -492,13 +453,6 @@ async fn test_app_with_two_accounts_and_affinity( function_call_ids: Vec::new(), variant_hash: None, continuation_scope: codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::Persisted, - replay: Some(replay_snapshot( - vec![json!({ - "role": "user", - "content": "Original managed history" - })], - Vec::new(), - )), created_at: now, }, ) @@ -513,79 +467,6 @@ async fn test_app_with_two_accounts_and_affinity( (router::router().with_state(state), api_key, dir) } -async fn test_app_with_two_accounts_and_affinity_status( - base_url: String, - affinity_account_status: &str, -) -> ( - axum::Router, - AppState, - String, - PgPool, - crate::support::storage::TestDatabaseGuard, -) { - let (pool, dir) = init_test_db(&format!( - "openai-responses-affinity-{affinity_account_status}" - )) - .await; - let api_key = insert_client_api_key(&pool).await; - insert_named_account(&pool, "acct_primary", "access-primary", "chatgpt-primary").await; - insert_named_account( - &pool, - "acct_affinity", - "access-affinity", - "chatgpt-affinity", - ) - .await; - sqlx::query("update accounts set status = $1 where id = $2") - .bind(affinity_account_status) - .bind("acct_affinity") - .execute(&pool) - .await - .unwrap(); - let now = Utc::now(); - let state = - test_app_state_with_pool(&test_config(test_database_url(), base_url), pool.clone()).await; - state - .services - .session_affinity - .record( - "resp_affinity_risk".to_string(), - SessionAffinityEntry { - account_id: "acct_affinity".to_string(), - conversation_id: "conv_affinity_risk".to_string(), - turn_state: Some("turn_affinity_risk".to_string()), - instructions_hash: None, - input_tokens: None, - function_call_ids: Vec::new(), - variant_hash: None, - continuation_scope: codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::Persisted, - replay: Some(replay_snapshot( - vec![json!({ - "role": "user", - "content": "Original affinity history" - })], - Vec::new(), - )), - created_at: now, - }, - ) - .await - .unwrap(); - state - .services - .account_pool - .restore_from_store() - .await - .expect("account pool should restore"); - ( - router::router().with_state(state.clone()), - state, - api_key, - pool, - dir, - ) -} - async fn test_app_with_two_accounts( base_url: String, ) -> ( @@ -807,7 +688,6 @@ async fn insert_session_affinity( function_call_ids: Vec::new(), variant_hash: None, continuation_scope: codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::Persisted, - replay: None, created_at: Utc::now(), }, ) diff --git a/backend/tests/dispatch/service/responses_http.rs b/backend/tests/dispatch/service/responses_http.rs index 2b2f3b5f6..46da6047d 100644 --- a/backend/tests/dispatch/service/responses_http.rs +++ b/backend/tests/dispatch/service/responses_http.rs @@ -1226,109 +1226,3 @@ async fn responses_should_prefer_session_affinity_account_for_previous_response( ); assert_eq!(captured.payload["previous_response_id"], "resp_previous"); } - -#[tokio::test] -async fn responses_should_replay_full_history_when_banned_affinity_switches_account() { - let (base_url, upstream) = - spawn_single_websocket_completed_upstream("resp_after_banned_affinity").await; - let (app, state, api_key, _pool, _dir) = - test_app_with_two_accounts_and_affinity_status(base_url, "banned").await; - let response = app - .oneshot( - Request::builder() - .method("POST") - .uri("/v1/responses") - .header("authorization", format!("Bearer {api_key}")) - .header("content-type", "application/json") - .body(Body::from( - json!({ - "model": "gpt-5.5", - "previous_response_id": "resp_affinity_risk", - "input": [{"role": "user", "content": "Continue after ban"}], - "stream": false - }) - .to_string(), - )) - .unwrap(), - ) - .await - .unwrap(); - let captured = upstream.await.unwrap(); - let affinity = state - .services - .session_affinity - .lookup("resp_affinity_risk", Utc::now()) - .await; - - assert_eq!(response.status(), StatusCode::OK); - assert_eq!( - captured_header(&captured.headers, "authorization"), - Some("Bearer access-primary") - ); - assert_eq!( - captured_header(&captured.headers, "chatgpt-account-id"), - Some("chatgpt-primary") - ); - assert!(captured_header(&captured.headers, "x-codex-turn-state").is_none()); - assert!(captured.payload.get("previous_response_id").is_none()); - assert_eq!( - captured.payload["input"][0]["content"], - "Original affinity history" - ); - assert_eq!( - captured.payload["input"][1]["content"], - "Continue after ban" - ); - assert!(affinity.is_some()); -} - -#[tokio::test] -async fn responses_should_replay_full_history_when_quota_affinity_switches_account() { - let (base_url, upstream) = - spawn_single_websocket_completed_upstream("resp_after_quota_affinity").await; - let (app, state, api_key, _pool, _dir) = - test_app_with_two_accounts_and_affinity_status(base_url, "quota_exhausted").await; - let response = app - .oneshot( - Request::builder() - .method("POST") - .uri("/v1/responses") - .header("authorization", format!("Bearer {api_key}")) - .header("content-type", "application/json") - .body(Body::from( - json!({ - "model": "gpt-5.5", - "previous_response_id": "resp_affinity_risk", - "input": [{"role": "user", "content": "Continue after quota"}], - "stream": false - }) - .to_string(), - )) - .unwrap(), - ) - .await - .unwrap(); - let captured = upstream.await.unwrap(); - let affinity = state - .services - .session_affinity - .lookup("resp_affinity_risk", Utc::now()) - .await; - - assert_eq!(response.status(), StatusCode::OK); - assert_eq!( - captured_header(&captured.headers, "authorization"), - Some("Bearer access-primary") - ); - assert!(captured_header(&captured.headers, "x-codex-turn-state").is_none()); - assert!(captured.payload.get("previous_response_id").is_none()); - assert_eq!( - captured.payload["input"][0]["content"], - "Original affinity history" - ); - assert_eq!( - captured.payload["input"][1]["content"], - "Continue after quota" - ); - assert!(affinity.is_some()); -} diff --git a/backend/tests/dispatch/service/responses_recovery.rs b/backend/tests/dispatch/service/responses_recovery.rs index 376cfaaa1..2e2c7a77b 100644 --- a/backend/tests/dispatch/service/responses_recovery.rs +++ b/backend/tests/dispatch/service/responses_recovery.rs @@ -1194,7 +1194,7 @@ async fn responses_external_previous_response_not_found_should_return_exact_upst } #[tokio::test] -async fn responses_managed_history_should_rebuild_full_replay_for_next_configured_candidate() { +async fn responses_managed_history_without_local_transcript_should_not_fan_out() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/codex/responses")) @@ -1205,17 +1205,6 @@ async fn responses_managed_history_should_rebuild_full_replay_for_next_configure .expect(1) .mount(&server) .await; - Mock::given(method("POST")) - .and(path("/codex/responses")) - .and(header("authorization", "Bearer access-secondary")) - .respond_with( - ResponseTemplate::new(200) - .insert_header("content-type", "text/event-stream") - .set_body_string(RESPONSES_SUCCESS_SSE), - ) - .expect(1) - .mount(&server) - .await; let (app, state, api_key, _pool, _dir) = test_app_with_two_accounts_and_state(server.uri()).await; state @@ -1232,18 +1221,6 @@ async fn responses_managed_history_should_rebuild_full_replay_for_next_configure function_call_ids: Vec::new(), variant_hash: None, continuation_scope: codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::Persisted, - replay: Some(replay_snapshot( - vec![ - json!({"role": "user", "content": "Earlier question"}), - json!({ - "type": "reasoning", - "id": "reasoning-old", - "encrypted_content": "account-bound-secret", - "summary": [] - }), - ], - Vec::new(), - )), created_at: Utc::now(), }, ) @@ -1272,26 +1249,19 @@ async fn responses_managed_history_should_rebuild_full_replay_for_next_configure ) .await .unwrap(); - assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); let requests = server.received_requests().await.unwrap(); - let secondary = requests - .iter() - .find(|request| { - request - .headers - .get("authorization") - .and_then(|value| value.to_str().ok()) - == Some("Bearer access-secondary") - }) - .expect("secondary request"); - let body: Value = serde_json::from_slice(&secondary.body).unwrap(); - assert!(body.get("previous_response_id").is_none()); - assert_eq!(body["input"][0]["content"], "Earlier question"); - assert!(body["input"][1].get("id").is_none()); - assert!(body["input"][1].get("encrypted_content").is_none()); - assert_eq!(body["input"][2]["content"], "Current question"); - assert!(secondary.headers.get("x-codex-turn-state").is_none()); + assert_eq!(requests.len(), 1); + assert_eq!( + requests[0] + .headers + .get("authorization") + .and_then(|value| value.to_str().ok()), + Some("Bearer access-primary") + ); + let body: Value = serde_json::from_slice(&requests[0].body).unwrap(); + assert_eq!(body["previous_response_id"], "resp_managed"); } #[tokio::test] @@ -1956,7 +1926,7 @@ async fn responses_stream_should_return_cloudflare_path_block_error_when_404_fal } #[tokio::test] -async fn responses_should_retry_same_account_after_5xx_before_fallback() { +async fn responses_should_rotate_account_after_5xx_without_same_account_retry() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/codex/responses")) @@ -1964,12 +1934,12 @@ async fn responses_should_retry_same_account_after_5xx_before_fallback() { .respond_with(ResponseTemplate::new(500).set_body_json(json!({ "error": {"message": "temporary upstream failure"} }))) - .up_to_n_times(2) + .expect(1) .mount(&server) .await; Mock::given(method("POST")) .and(path("/codex/responses")) - .and(header("authorization", "Bearer access-primary")) + .and(header("authorization", "Bearer access-secondary")) .respond_with( ResponseTemplate::new(200) .insert_header("content-type", "text/event-stream") @@ -2014,8 +1984,8 @@ async fn responses_should_retry_same_account_after_5xx_before_fallback() { assert_eq!(status, StatusCode::OK, "requests: {authorizations:?}"); assert_eq!(body["id"], "resp_after_5xx_retry"); - assert_eq!(primary_requests, 3, "requests: {authorizations:?}"); - assert_eq!(secondary_requests, 0, "requests: {authorizations:?}"); + assert_eq!(primary_requests, 1, "requests: {authorizations:?}"); + assert_eq!(secondary_requests, 1, "requests: {authorizations:?}"); } #[tokio::test] @@ -2075,7 +2045,7 @@ async fn responses_should_traverse_all_configured_strategy_candidates_without_fi } #[tokio::test] -async fn responses_stream_should_retry_same_account_after_5xx_before_fallback() { +async fn responses_stream_should_rotate_account_after_5xx_without_same_account_retry() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/codex/responses")) @@ -2083,12 +2053,12 @@ async fn responses_stream_should_retry_same_account_after_5xx_before_fallback() .respond_with(ResponseTemplate::new(500).set_body_json(json!({ "error": {"message": "temporary upstream failure"} }))) - .up_to_n_times(2) + .expect(1) .mount(&server) .await; Mock::given(method("POST")) .and(path("/codex/responses")) - .and(header("authorization", "Bearer access-primary")) + .and(header("authorization", "Bearer access-secondary")) .respond_with( ResponseTemplate::new(200) .insert_header("content-type", "text/event-stream") @@ -2140,8 +2110,8 @@ async fn responses_stream_should_retry_same_account_after_5xx_before_fallback() assert_eq!(status, StatusCode::OK, "requests: {authorizations:?}"); assert!(content_type.starts_with("text/event-stream")); assert!(body.contains("resp_stream_after_5xx_retry")); - assert_eq!(primary_requests, 3, "requests: {authorizations:?}"); - assert_eq!(secondary_requests, 0, "requests: {authorizations:?}"); + assert_eq!(primary_requests, 1, "requests: {authorizations:?}"); + assert_eq!(secondary_requests, 1, "requests: {authorizations:?}"); } #[tokio::test] diff --git a/backend/tests/dispatch/service/responses_websocket.rs b/backend/tests/dispatch/service/responses_websocket.rs index aad054de3..92c0b6b67 100644 --- a/backend/tests/dispatch/service/responses_websocket.rs +++ b/backend/tests/dispatch/service/responses_websocket.rs @@ -354,305 +354,6 @@ async fn responses_websocket_should_reuse_connection_for_recorded_conversation() assert_eq!(second_payload["previous_response_id"], "resp_pool_first"); } -#[tokio::test] -async fn responses_websocket_should_replay_history_when_reused_connection_dies_before_first_output() -{ - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let base_url = format!("http://{}", listener.local_addr().unwrap()); - let upstream = tokio::spawn(async move { - let (stream, _) = listener.accept().await.unwrap(); - let mut websocket = accept_async(stream).await.unwrap(); - let first_message = websocket.next().await.unwrap().unwrap(); - let first_payload = serde_json::from_str::(&first_message.into_text().unwrap()) - .expect("first websocket payload should be json"); - websocket - .send(Message::Text( - response_completed_websocket_message("resp_stale_reuse_first").into(), - )) - .await - .unwrap(); - - let stale_message = websocket.next().await.unwrap().unwrap(); - let stale_payload = serde_json::from_str::(&stale_message.into_text().unwrap()) - .expect("stale reused websocket payload should be json"); - websocket.close(None).await.unwrap(); - - let fresh_payload = - accept_successful_websocket_response(&listener, "resp_after_stale_reuse").await; - - (first_payload, stale_payload, fresh_payload) - }); - let (app, api_key, _dir) = test_app_with_account(base_url).await; - - let first_response = app - .clone() - .oneshot(responses_json_request( - &api_key, - &json!({ - "model": "gpt-5.5", - "input": [{ - "role": "user", - "content": "prime stale websocket reuse" - }] - }), - )) - .await - .unwrap(); - assert_eq!(first_response.status(), StatusCode::OK); - assert!( - response_text(first_response) - .await - .contains("\"id\":\"resp_stale_reuse_first\"") - ); - - let second_response = app - .oneshot(responses_json_request( - &api_key, - &json!({ - "model": "gpt-5.5", - "input": [], - "previous_response_id": "resp_stale_reuse_first" - }), - )) - .await - .unwrap(); - assert_eq!(second_response.status(), StatusCode::OK); - let second_body = response_text(second_response).await; - assert!( - second_body.contains("\"id\":\"resp_after_stale_reuse\""), - "{second_body}" - ); - let (first_payload, stale_payload, fresh_payload) = upstream.await.unwrap(); - - assert_eq!( - stale_payload["previous_response_id"], - "resp_stale_reuse_first" - ); - assert!(fresh_payload.get("previous_response_id").is_none()); - assert_eq!( - fresh_payload["input"][0]["content"], - "prime stale websocket reuse" - ); - assert!(first_payload.get("prompt_cache_key").is_none()); - assert!(fresh_payload.get("prompt_cache_key").is_none()); -} - -#[tokio::test] -async fn responses_websocket_stream_should_replay_managed_history_after_previous_response_not_found() - { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let base_url = format!("http://{}", listener.local_addr().unwrap()); - let upstream = tokio::spawn(async move { - let (first_stream, _) = listener.accept().await.unwrap(); - let mut first_websocket = accept_async(first_stream).await.unwrap(); - let first_message = first_websocket.next().await.unwrap().unwrap(); - let first_payload = serde_json::from_str::(&first_message.into_text().unwrap()) - .expect("first websocket payload should be json"); - first_websocket - .send(Message::Text( - response_failed_websocket_message( - "resp_previous_missing", - "previous_response_not_found", - "Previous response with id resp_managed_stale was not found", - ) - .into(), - )) - .await - .unwrap(); - first_websocket.close(None).await.unwrap(); - - let replay_payload = - accept_successful_websocket_response(&listener, "resp_replayed_success").await; - (first_payload, replay_payload) - }); - let (app, state, api_key, _dir) = test_app_with_account_and_state(base_url).await; - state - .services - .session_affinity - .record( - "resp_managed_stale".to_string(), - SessionAffinityEntry { - account_id: "acct_chat".to_string(), - conversation_id: "conversation-managed-stale".to_string(), - turn_state: Some("account-bound-turn-state".to_string()), - instructions_hash: None, - input_tokens: Some(7), - function_call_ids: Vec::new(), - variant_hash: None, - continuation_scope: codex_proxy_rs::upstream::openai::protocol::responses::PreviousResponseScope::Persisted, - replay: Some(replay_snapshot( - vec![json!({ - "role": "user", - "content": "Earlier managed question" - })], - Vec::new(), - )), - created_at: Utc::now(), - }, - ) - .await - .unwrap(); - - let response = app - .oneshot(responses_json_request( - &api_key, - &json!({ - "model": "gpt-5.5", - "previous_response_id": "resp_managed_stale", - "input": [{"role": "user", "content": "Current managed question"}], - "stream": true - }), - )) - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::OK); - let body = response_text(response).await; - let (first_payload, replay_payload) = upstream.await.unwrap(); - - assert!(body.contains("resp_replayed_success"), "{body}"); - assert!(!body.contains("resp_previous_missing"), "{body}"); - assert_eq!(first_payload["previous_response_id"], "resp_managed_stale"); - assert!(replay_payload.get("previous_response_id").is_none()); - assert_eq!( - replay_payload["input"][0]["content"], - "Earlier managed question" - ); - assert_eq!( - replay_payload["input"][1]["content"], - "Current managed question" - ); -} - -#[tokio::test] -async fn responses_websocket_should_store_sanitized_incremental_replay_snapshot() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let base_url = format!("http://{}", listener.local_addr().unwrap()); - let upstream = tokio::spawn(async move { - let (stream, _) = listener.accept().await.unwrap(); - let mut websocket = accept_async(stream).await.unwrap(); - let _payload = websocket.next().await.unwrap().unwrap(); - let mut completed: Value = serde_json::from_str(include_str!( - "../../fixtures/responses/websocket/completed_with_reasoning_replay.json" - )) - .unwrap(); - completed["response"]["status"] = Value::String("completed".to_string()); - completed["response"]["output"] - .as_array_mut() - .unwrap() - .push(json!({ - "type": "message", - "id": "msg_sensitive_output", - "role": "assistant", - "content": [{"type": "output_text", "text": "done"}] - })); - websocket - .send(Message::Text(completed.to_string().into())) - .await - .unwrap(); - websocket.close(None).await.unwrap(); - }); - let (app, state, api_key, _dir) = test_app_with_account_and_state(base_url).await; - - let response = app - .oneshot(responses_json_request( - &api_key, - &json!({ - "model": "gpt-5.5", - "stream": false, - "input": [ - { - "type": "message", - "id": "account-bound-input-id", - "role": "user", - "content": [{ - "type": "input_text", - "text": "remember this", - "encrypted_content": "nested-secret" - }] - }, - { - "type": "function_call_output", - "call_id": "call_must_survive", - "output": "ok" - } - ] - }), - )) - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::OK); - upstream.await.unwrap(); - - let stored = state - .services - .session_affinity - .lookup("resp_implicit_resume_first", Utc::now()) - .await - .unwrap(); - let replay = stored.replay.unwrap(); - assert_eq!(replay.depth, 1); - assert!(replay.parent_response_id.is_none()); - assert!(replay.turn_input[0].get("id").is_none()); - assert_eq!(replay.turn_input[1]["call_id"], "call_must_survive"); - assert!(replay.turn_output[0].get("id").is_none()); - let serialized = serde_json::to_string(&replay).unwrap(); - assert!(!serialized.contains("encrypted_content")); - assert!(!serialized.contains("nested-secret")); - assert!(!serialized.contains("enc_reasoning_replay")); -} - -#[tokio::test] -async fn responses_should_skip_oversized_replay_snapshot_without_failing_response() { - let server = MockServer::start().await; - let oversized = "x".repeat(2 * 1024 * 1024 + 1024); - let completed = json!({ - "response": { - "id": "resp_oversized_replay", - "object": "response", - "status": "completed", - "output": [{ - "type": "message", - "id": "msg_oversized", - "role": "assistant", - "content": [{"type": "output_text", "text": oversized}] - }], - "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2} - } - }); - Mock::given(method("POST")) - .and(path("/codex/responses")) - .respond_with( - ResponseTemplate::new(200) - .insert_header("content-type", "text/event-stream") - .set_body_string(format!("event: response.completed\ndata: {completed}\n\n")), - ) - .expect(1) - .mount(&server) - .await; - let (app, state, api_key, _dir) = test_app_with_account_and_state(server.uri()).await; - - let response = app - .oneshot(responses_json_request( - &api_key, - &json!({ - "model": "gpt-5.5", - "input": [{"role": "user", "content": "small input"}], - "stream": false, - "use_websocket": false - }), - )) - .await - .unwrap(); - assert_eq!(response.status(), StatusCode::OK); - let stored = state - .services - .session_affinity - .lookup("resp_oversized_replay", Utc::now()) - .await - .unwrap(); - assert!(stored.replay.is_none()); -} - #[tokio::test] async fn responses_websocket_should_not_reuse_connection_when_pool_is_disabled() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -767,8 +468,10 @@ async fn responses_websocket_should_not_reuse_connection_when_pool_is_disabled() .body(Body::from( json!({ "model": "gpt-5.5", - "input": [], - "previous_response_id": "resp_disabled_pool_first" + "input": [{ + "role": "user", + "content": "open a second independent websocket" + }] }) .to_string(), )) @@ -793,124 +496,10 @@ async fn responses_websocket_should_not_reuse_connection_when_pool_is_disabled() assert!(second_payload.get("previous_response_id").is_none()); assert_eq!( second_payload["input"][0]["content"], - "do not reuse this upstream websocket" + "open a second independent websocket" ); } -#[tokio::test] -async fn responses_websocket_stream_should_strip_turn_state_when_replaying_on_new_connection() { - let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); - let base_url = format!("http://{}", listener.local_addr().unwrap()); - let upstream = tokio::spawn(async move { - let (first_stream, _) = listener.accept().await.unwrap(); - let mut first_websocket = accept_async(first_stream).await.unwrap(); - let first_message = first_websocket.next().await.unwrap().unwrap(); - let first_payload = serde_json::from_str::(&first_message.into_text().unwrap()) - .expect("first websocket payload should be json"); - first_websocket - .send(Message::Text( - json!({ - "type": "response.metadata", - "headers": { - "x-codex-turn-state": "turn-from-metadata" - } - }) - .to_string() - .into(), - )) - .await - .unwrap(); - first_websocket - .send(Message::Text( - response_completed_websocket_message("resp_metadata_turn_state").into(), - )) - .await - .unwrap(); - first_websocket.close(None).await.unwrap(); - - let (second_stream, _) = listener.accept().await.unwrap(); - let second_headers = Arc::new(Mutex::new(Vec::new())); - let second_headers_for_callback = Arc::clone(&second_headers); - let mut second_websocket = accept_hdr_async(second_stream, move |request, _response| { - *second_headers_for_callback.lock().unwrap() = request_headers(request); - }) - .await - .unwrap(); - let second_message = second_websocket.next().await.unwrap().unwrap(); - let second_payload = serde_json::from_str::(&second_message.into_text().unwrap()) - .expect("second websocket payload should be json"); - second_websocket - .send(Message::Text( - response_completed_websocket_message("resp_metadata_turn_state_next").into(), - )) - .await - .unwrap(); - second_websocket.close(None).await.unwrap(); - let second_headers = second_headers.lock().unwrap().clone(); - (first_payload, second_payload, second_headers) - }); - let (app, api_key, _pool, _dir) = test_app_with_account_pool_config(base_url, |config| { - config.ws_pool.enabled = false; - }) - .await; - - let first_response = app - .clone() - .oneshot( - Request::builder() - .method("POST") - .uri("/v1/responses") - .header("authorization", format!("Bearer {api_key}")) - .header("content-type", "application/json") - .body(Body::from( - json!({ - "model": "gpt-5.5", - "input": [], - "stream": true, - "use_websocket": true - }) - .to_string(), - )) - .unwrap(), - ) - .await - .unwrap(); - assert_eq!(first_response.status(), StatusCode::OK); - let first_body = response_text(first_response).await; - assert!(first_body.contains("\"id\":\"resp_metadata_turn_state\"")); - assert!(!first_body.contains("response.metadata")); - let second_response = app - .oneshot( - Request::builder() - .method("POST") - .uri("/v1/responses") - .header("authorization", format!("Bearer {api_key}")) - .header("content-type", "application/json") - .body(Body::from( - json!({ - "model": "gpt-5.5", - "input": [], - "previous_response_id": "resp_metadata_turn_state" - }) - .to_string(), - )) - .unwrap(), - ) - .await - .unwrap(); - assert_eq!(second_response.status(), StatusCode::OK); - assert!( - response_text(second_response) - .await - .contains("\"id\":\"resp_metadata_turn_state_next\"") - ); - let (first_payload, second_payload, second_headers) = upstream.await.unwrap(); - - assert!(first_payload.get("previous_response_id").is_none()); - assert!(second_payload.get("previous_response_id").is_none()); - assert!(captured_header(&second_headers, "x-codex-turn-state").is_none()); -} - #[tokio::test] async fn responses_websocket_pool_should_be_evicted_after_admin_account_status_cycle() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -991,8 +580,10 @@ async fn responses_websocket_pool_should_be_evicted_after_admin_account_status_c &api_key, &json!({ "model": "gpt-5.5", - "input": [], - "previous_response_id": "resp_pool_status_first" + "input": [{ + "role": "user", + "content": "start again after the account status cycle" + }] }), )) .await diff --git a/backend/tests/dispatch/service/usage_logging.rs b/backend/tests/dispatch/service/usage_logging.rs index 1083a5157..9adac3454 100644 --- a/backend/tests/dispatch/service/usage_logging.rs +++ b/backend/tests/dispatch/service/usage_logging.rs @@ -73,7 +73,7 @@ async fn responses_should_succeed_when_redis_affinity_write_fails() { .unwrap(); let mut connection = redis.manager(); let _: () = redis::cmd("SET") - .arg(redis.key("affinity:v2:account:acct_chat")) + .arg(redis.key("affinity:v3:account:acct_chat")) .arg("wrong-type") .query_async(&mut connection) .await @@ -908,7 +908,7 @@ async fn responses_stream_should_not_record_first_token_when_metadata_only_prefi } #[tokio::test] -async fn responses_should_record_request_count_when_5xx_retries_are_exhausted() { +async fn responses_should_record_one_request_when_single_account_returns_5xx() { let server = MockServer::start().await; Mock::given(method("POST")) .and(path("/codex/responses")) @@ -950,7 +950,7 @@ async fn responses_should_record_request_count_when_5xx_retries_are_exhausted() .unwrap(); assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); - assert_eq!(authorizations.len(), 3, "requests: {authorizations:?}"); + assert_eq!(authorizations.len(), 1, "requests: {authorizations:?}"); assert_eq!(failed_usage.map(|row| row.0).unwrap_or_default(), 1); let event = latest_response_upstream_ops_error_log(&pool).await; let metadata: Value = serde_json::from_str(&event.metadata_json).unwrap(); @@ -976,7 +976,7 @@ async fn responses_should_record_attempt_trace_after_retrying_another_account() .respond_with(ResponseTemplate::new(500).set_body_json(json!({ "error": {"message": "secondary failed"} }))) - .expect(3) + .expect(1) .mount(&server) .await; diff --git a/backend/tests/upstream/openai/transport/http_client.rs b/backend/tests/upstream/openai/transport/http_client.rs index 3463520a1..79d47123c 100644 --- a/backend/tests/upstream/openai/transport/http_client.rs +++ b/backend/tests/upstream/openai/transport/http_client.rs @@ -212,6 +212,49 @@ async fn codex_backend_client_should_parse_retry_after_from_rate_limit_error_bod assert_eq!(retry_after_seconds, Some(12)); } +#[tokio::test(start_paused = true)] +async fn codex_backend_http_sse_should_fail_after_five_minutes_without_stream_data() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + let _request = read_http_request(&mut stream).await; + stream + .write_all( + b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\n\r\n", + ) + .await + .unwrap(); + std::future::pending::<()>().await; + }); + let client = CodexBackendClient::new( + reqwest::Client::builder().no_proxy().build().unwrap(), + base_url, + crate::support::fingerprint::runtime_test_fingerprint(), + ); + let mut request = + codex_proxy_rs::upstream::openai::protocol::responses::CodexResponsesRequest::new_http_sse( + "gpt-5.5", + "", + Vec::new(), + ); + request.force_http_sse = true; + + let mut response = client + .create_response_stream(&request, request_context("req_http_idle", Some("acct"))) + .await + .unwrap(); + let next = tokio::spawn(async move { response.body.next().await }); + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(5 * 60)).await; + + let Some(Err(CodexClientError::StreamIdleTimeout { timeout })) = next.await.unwrap() else { + panic!("expected HTTP/SSE idle timeout"); + }; + assert_eq!(timeout, Duration::from_secs(5 * 60)); + server.abort(); +} + #[tokio::test] async fn codex_backend_client_should_capture_only_safe_response_metadata() { let server = wiremock::MockServer::start().await; diff --git a/docs/architecture.md b/docs/architecture.md index 6eaeafed7..f83f86e59 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -244,11 +244,11 @@ dispatch/ - `service.rs`：非流式 Responses 主循环。 - `stream/lifecycle.rs`：流式 Responses 主循环。 -- `upstream_call.rs`：把账号凭据、账号作用域身份和 Cookie 交给上游客户端，处理单账号有限重试。 +- `upstream_call.rs`：把账号凭据、账号作用域身份和 Cookie 交给上游客户端。 - `attempts.rs`：在请求开始时冻结完整候选顺序，记录已尝试、忙碌和状态排除的账号。 - `errors.rs`：错误分类与客户端错误映射。 - `recording.rs`：把调度轨迹和上游诊断写入 `telemetry::Recorder`。 -- `affinity`：会话亲和、账号作用域身份、Redis 索引和响应历史快照。 +- `affinity`：会话亲和、账号作用域身份和 Redis 响应归属索引。 - `recovery`：账号风险隔离、Cloudflare 冷却、候选耗尽和 previous response 恢复。 - `stream`：首段预取、SSE 失败识别、下游流转发和结束时结算。 @@ -378,7 +378,7 @@ bootstrap/ 5. `AccountAttemptLedger` 按当前调度策略冻结完整候选顺序。会话亲和账号可以排在前面，但不会绕过可用性检查。 6. 账号池租用并发槽位，必要时先刷新 quota；请求间隔在发送上游前执行。 7. `upstream_call` 注入账号 token、`chatgpt-account-id`、账号作用域身份、指纹请求头和 Cookie。 -8. 上游通过 HTTP/SSE 或 WebSocket 返回结果。单账号可重试的 5xx 最多额外重试两次。 +8. 上游通过 HTTP/SSE 或 WebSocket 返回结果。建连最长等待 15 秒；WebSocket 发送后 20 秒没有任何上游事件时放弃异常连接，收到首个事件后与 HTTP/SSE 一样使用 5 分钟流空闲窗口。账号级 5xx 和传输故障回到候选账本，不在同一账号内叠加重试。 9. 在结果尚未提交给客户端时，账号级失败可继续尝试候选账本中的下一个账号。 10. 完成后更新账号用量、会话亲和、调度反馈和遥测事实，并释放账号槽位。 @@ -391,7 +391,7 @@ bootstrap/ - `scheduler::candidates` 过滤非 active、配额不可用、模型不匹配、处于 Cloudflare 冷却或明确排除的账号。 - 当前 `rotation_strategy` 对剩余账号排序。请求级快照保留所有符合条件的层级，层级只影响顺序，不缩小 failover 边界。 -单个请求的候选顺序在开始时冻结。瞬时并发槽位占满的账号进入 busy 队列，等待槽位变化后再判断；状态已经变化的账号被排除。候选账本不会重复取出同一个账号。单账号内部的 5xx 重试，以及 previous response 失效后在原账号执行的完整重放，是账本之外的明确重试。 +单个请求的候选顺序在开始时冻结。瞬时并发槽位占满的账号进入 busy 队列，最多等待 10 秒；状态已经变化的账号被排除。候选账本不会重复取出同一个账号。复用的 WebSocket 已失效时允许新建一次连接；当前下游 WebSocket 持有完整历史时，previous response 恢复可以在同账号或后续候选账号上重放。 会触发账号隔离或换号的典型情况： @@ -408,21 +408,17 @@ bootstrap/ ## previous response 与流式恢复 -成功响应写入 Redis 亲和记录，保存响应 ID、账号 ID、会话 ID、turn state、变体哈希和受限的 replay 节点。 +成功响应写入 Redis 亲和记录，保存响应 ID、账号 ID、会话 ID、turn state 和变体哈希。Redis 不保存请求正文、响应正文或完整对话历史。 -previous response 分两类处理： +完整历史只保存在当前下游 Responses WebSocket 的连接状态中。每次 `response.completed` 后，将本轮输入和上游输出追加到该连接的 transcript；写入前递归删除对象 `id` 和 `encrypted_content`，保留工具续接所需的 `call_id`。连接关闭时 transcript 随处理任务一并释放，不写入 Redis，也没有进程级共享 transcript 缓存。 -- 受管历史：Redis 能找到响应归属。先使用原账号和原 previous response；若上游返回 `previous_response_not_found`、连接续接忙碌或原连接不可复用，先在原账号执行完整历史重放。需要换号且 replay 完整时，其他账号也使用完整历史重放。 -- 外部未知历史：Redis 没有该响应。只能在第一个账号原样尝试一次，因为服务端没有足够历史安全地转移到其他账号；失败时保留上游错误。 +previous response 按历史来源处理： -replay 按轮保存增量输入和输出，不保存每一轮累计的完整输入。写入前递归删除对象 ID 和 `encrypted_content`。容量边界： +- 当前下游 WebSocket 有完整 transcript：先使用亲和账号和原 previous response。若上游返回 `previous_response_not_found`、连接忙碌或原连接失效，去掉 previous ID 后用完整 transcript 重放。账号级失败需要换号时，后续账号仍由当前调度策略选出，并接收同一份完整 transcript。 +- Redis 能找到归属但当前连接没有 transcript：只允许原账号按原请求续接；失败后不能把不完整上下文发给其他账号。 +- Redis 没有归属且当前连接也没有 transcript：在首个候选账号原样尝试一次，失败时保留上游错误。 -- 单轮 snapshot：2 MiB -- 单会话累计 replay：16 MiB -- 最大深度：128 轮 -- Redis TTL：4 小时 - -超过任一边界时不再生成新的 replay 节点，正常代理响应不因此失败。 +新的下游 WebSocket 不继承上一条连接的 transcript。官方 Codex 客户端收到可重试断流后，从本地 Session 重新组装完整历史并建立新请求；该请求重新进入账号池调度。客户端单条 HTTP 请求或 WebSocket 文本帧上限为 16 MiB。 流式请求只有在响应尚未提交给客户端时才能透明恢复或换号。首段预取会识别终止事件和账号级失败；一旦真实输出已经发给客户端，代理会用 `response.failed` 结束当前 SSE 或 WebSocket response，不能在另一账号上重放后拼接到已有输出。 @@ -483,11 +479,14 @@ Redis 使用统一的 `cpr:` 前缀。 | `cpr:admin:session:` | 管理员登录会话 | `admin.session_ttl_minutes` | | `cpr:lease:refresh:` | token 刷新租约 | 租约 PX TTL | | `cpr:models:plan_snapshots` | 各订阅计划模型快照 HASH | 下一次完整刷新替换 | -| `cpr:affinity:v2:resp:` | 响应归属和 replay 节点 | 4 小时 | -| `cpr:affinity:v2:conv:` | 会话响应 ZSET 索引 | 4 小时 | -| `cpr:affinity:v2:account:` | 账号响应 ZSET 索引 | 4 小时 | +| `cpr:affinity:v3:resp:` | 响应归属元数据 | 4 小时 | +| `cpr:affinity:v3:conv:` | 会话响应 ZSET 索引，最多 128 项 | 4 小时 | +| `cpr:affinity:v3:account:` | 账号响应 ZSET 索引，最多 16384 项 | 4 小时 | +| `cpr:affinity:v3:global:index` | 全局响应 ZSET 索引，最多 65536 项 | 4 小时 | +| `cpr:affinity:v3:global:bytes` | 亲和元数据总字节数 | 4 小时 | +| `cpr:affinity:v3:global:info` | 亲和索引清理所需的归属与字节信息 | 4 小时 | -账号删除时会批量删除该账号的亲和响应并清理会话索引。Redis 写入失败不应改变已经得到的上游响应语义，但会失去对应的会话恢复、模型缓存或分布式协调能力。 +亲和元数据单条不超过 64 KiB，全局最多 128 MiB。写入时惰性清理过期成员并执行会话、账号和全局裁剪。账号删除使用一段 Redis Lua 操作批量删除该账号的响应归属并清理会话索引。Redis 写入失败不改变已经得到的上游响应语义，但会失去对应的账号亲和、模型缓存或分布式协调能力。 ### 本地文件