From 5877b0cb2ffd745ebb58bdd2883120779429c1a6 Mon Sep 17 00:00:00 2001 From: Zyy Date: Thu, 29 Oct 2026 12:04:09 +0800 Subject: [PATCH] fix(openai): tolerate session affinity storage jitter Raise the session affinity storage budget to 500ms while keeping cyber policy operations on a separate 100ms budget. Preserve fail-closed binding behavior and cover delayed admission and timeout rejection before upstream sending. Refs #549 Co-authored-by: Codex --- .../openai/src/credential/selector.rs | 16 +++-- .../provider/contract/session_binding.rs | 65 +++++++++++++++++++ .../crates/providers/openai/tests/support.rs | 18 +++++ 3 files changed, 92 insertions(+), 7 deletions(-) diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index e33899acf..503ffa355 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -45,7 +45,9 @@ const CLOUDFLARE_CHALLENGE_BACKOFF: [Duration; 4] = [ Duration::from_secs(120), ]; const CLOUDFLARE_PATH_BLOCK_THRESHOLD: u32 = 3; -const SESSION_AFFINITY_TIMEOUT: Duration = Duration::from_millis(100); +// 绑定失败会阻止发送，预算容纳短暂调度抖动；可丢失的策略排除记录保持独立预算 +const SESSION_AFFINITY_TIMEOUT: Duration = Duration::from_millis(500); +const CYBER_POLICY_TIMEOUT: Duration = Duration::from_millis(100); const CYBER_POLICY_SESSION_TTL: Duration = Duration::from_secs(60 * 60); const MAX_ACCOUNT_SNAPSHOT_RETRIES: u32 = 3; const COOLDOWN_READ_CONCURRENCY: usize = 128; @@ -1141,7 +1143,7 @@ impl CodexCredentialSelector { ) -> Option { let key = key?.clone(); let state = match tokio::time::timeout( - SESSION_AFFINITY_TIMEOUT, + CYBER_POLICY_TIMEOUT, self.session_exclusions.load(&self.provider_kind, &key), ) .await @@ -1153,7 +1155,7 @@ impl CodexCredentialSelector { } Err(_) => { tracing::warn!( - timeout_ms = SESSION_AFFINITY_TIMEOUT.as_millis(), + timeout_ms = CYBER_POLICY_TIMEOUT.as_millis(), "OpenAI cyber policy state read timed out" ); None @@ -1171,7 +1173,7 @@ impl CodexCredentialSelector { return; }; match tokio::time::timeout( - SESSION_AFFINITY_TIMEOUT, + CYBER_POLICY_TIMEOUT, self.session_exclusions.record_failure( &self.provider_kind, &scope.key, @@ -1192,7 +1194,7 @@ impl CodexCredentialSelector { Err(_) => { tracing::warn!( account_id = %account.id(), - timeout_ms = SESSION_AFFINITY_TIMEOUT.as_millis(), + timeout_ms = CYBER_POLICY_TIMEOUT.as_millis(), "OpenAI cyber policy exclusion write timed out" ); } @@ -1207,7 +1209,7 @@ impl CodexCredentialSelector { return; }; match tokio::time::timeout( - SESSION_AFFINITY_TIMEOUT, + CYBER_POLICY_TIMEOUT, self.session_exclusions .clear(&self.provider_kind, &scope.key, state.revision()), ) @@ -1219,7 +1221,7 @@ impl CodexCredentialSelector { } Err(_) => { tracing::warn!( - timeout_ms = SESSION_AFFINITY_TIMEOUT.as_millis(), + timeout_ms = CYBER_POLICY_TIMEOUT.as_millis(), "OpenAI cyber policy exclusion clear timed out" ); } diff --git a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs index 0106310a5..2e01551f6 100644 --- a/backend/crates/providers/openai/tests/provider/contract/session_binding.rs +++ b/backend/crates/providers/openai/tests/provider/contract/session_binding.rs @@ -2,6 +2,71 @@ use super::*; +#[tokio::test] +async fn session_affinity_storage_tolerates_delays_but_rejects_expired_requests_before_sending() { + for (delay_ms, should_succeed) in [(250, true), (750, false)] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_subagent_a").await; + let affinity = Arc::new(MemorySessionAffinity::with_operation_delay( + Duration::from_millis(delay_ms), + )); + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/codex/responses")) + .respond_with( + ResponseTemplate::new(200).set_body_raw( + format!("event: response.created\ndata: {{\"type\":\"response.created\",\"response\":{{\"id\":\"resp_scope_capture\",\"status\":\"in_progress\"}}}}\n\n{CAPTURE_COMPLETED_SSE}"), + "text/event-stream", + ), + ) + .expect(if should_succeed { 1 } else { 0 }) + .mount(&server) + .await; + let provider = provider_with_affinity_and_base_url(&store, affinity.clone(), server.uri()); + + // 只推进存储等待的虚拟时间，网络交付恢复真实时钟，避免依赖宿主调度速度 + tokio::time::pause(); + let result = provider + .execute( + planned_request("openai", turn_request("delayed-session", "delayed-turn")), + context("req_affinity_delay", CancellationToken::new()), + ) + .await; + tokio::time::resume(); + + if should_succeed { + let mut stream = result.expect("delayed binding and turn association must succeed"); + assert_eq!( + stream.metadata().provider_account_id().as_str(), + "acct_subagent_a" + ); + assert_eq!(affinity.binding_count(), 1); + assert_eq!(affinity.alias_ttls().len(), 1); + let mut completed = false; + while let Some(event) = stream.next().await { + completed |= event + .unwrap() + .canonical_facts() + .iter() + .any(|event| matches!(event, GatewayEvent::Completed(_))); + } + assert!(completed, "admitted request must complete through upstream"); + } else { + let error = result + .err() + .expect("expired storage wait must reject the request"); + assert_eq!( + error.kind(), + ProviderErrorKind::ProviderInfrastructureUnavailable + ); + assert_eq!(error.send_state(), UpstreamSendState::NotSent); + assert!(error.retry_is_prohibited()); + assert_eq!(affinity.binding_count(), 0); + } + server.verify().await; + } +} + #[tokio::test] async fn cyber_policy_session_exclusion_has_actionable_non_retryable_error() { let store = Arc::new(MemoryAccountStore::default()); diff --git a/backend/crates/providers/openai/tests/support.rs b/backend/crates/providers/openai/tests/support.rs index 6bdb9ff9f..62d6985e0 100644 --- a/backend/crates/providers/openai/tests/support.rs +++ b/backend/crates/providers/openai/tests/support.rs @@ -785,6 +785,7 @@ impl ProviderLeasePort for TestLeaseCoordinator { #[derive(Default)] pub(crate) struct MemorySessionAffinity { + operation_delay: Duration, initial_claim_barrier: Option>, aliases: Mutex>, bindings: @@ -795,6 +796,19 @@ pub(crate) struct MemorySessionAffinity { } impl MemorySessionAffinity { + pub(crate) fn with_operation_delay(operation_delay: Duration) -> Self { + Self { + operation_delay, + ..Self::default() + } + } + + async fn delay_operation(&self) { + if !self.operation_delay.is_zero() { + tokio::time::sleep(self.operation_delay).await; + } + } + pub(crate) fn with_initial_claim_barrier(participants: usize) -> Self { Self { initial_claim_barrier: Some(Arc::new(tokio::sync::Barrier::new(participants))), @@ -865,6 +879,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity { Result, ProviderStoreError>, > { Box::pin(async move { + self.delay_operation().await; self.lookups .lock() .expect("lookups") @@ -893,6 +908,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity { Result, ProviderStoreError>, > { Box::pin(async move { + self.delay_operation().await; // 让并发首请求都读到空绑定，再验证发送前的原子认领冲突 if expected.is_none() && let Some(barrier) = &self.initial_claim_barrier @@ -931,6 +947,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity { Result, ProviderStoreError>, > { Box::pin(async move { + self.delay_operation().await; Ok(self .aliases .lock() @@ -950,6 +967,7 @@ impl ProviderSessionAffinityPort for MemorySessionAffinity { ttl: Duration, ) -> BoxFuture<'a, Result> { Box::pin(async move { + self.delay_operation().await; let mut aliases = self.aliases.lock().unwrap(); let current = aliases .entry((