diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 2b56b6f96..ffd79f502 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -2158,6 +2158,7 @@ dependencies = [ "async-stream", "async-trait", "base64", + "bytes", "chrono", "futures", "gateway-admin", diff --git a/backend/crates/providers/openai/src/credential/quota.rs b/backend/crates/providers/openai/src/credential/quota.rs index e8255147b..425f8379e 100644 --- a/backend/crates/providers/openai/src/credential/quota.rs +++ b/backend/crates/providers/openai/src/credential/quota.rs @@ -658,23 +658,20 @@ impl CodexCredentialQuotaService { if outcome == QuotaWriteOutcome::Conflict { return Ok(false); } - let Some(current) = self.store.get_account(account.id()).await? else { - return Ok(false); - }; - if current.revision() != account.revision() { - return Ok(false); - } + // quota CAS 以 account.revision 为 fence,成功即证明该 revision 在写入时 + // 仍是当前凭据;后续状态写回带同一 fence,轮换竞态下安全失败。 + // 可用性判断使用选择时的账号快照,偏差在下一次响应或 quota 轮询收敛。 self.scheduling - .observe(current.id().clone(), current.revision(), observed_at, fact); + .observe(account.id().clone(), account.revision(), observed_at, fact); if let Some(availability) = quota_success_availability( - current.availability(), - current.cooldown_until(), + account.availability(), + account.cooldown_until(), fact.exhausted(), observed_at, ) { self.repository .apply_state( - ¤t, + account, availability, fact.exhausted().then_some("quota_exhausted".to_owned()), None, diff --git a/backend/crates/providers/xai/Cargo.toml b/backend/crates/providers/xai/Cargo.toml index 850c2d09b..077c53bea 100644 --- a/backend/crates/providers/xai/Cargo.toml +++ b/backend/crates/providers/xai/Cargo.toml @@ -11,6 +11,7 @@ autotests = false async-stream = "0.3.6" async-trait = "0.1.89" base64 = "0.22.1" +bytes = "1.9.0" chrono = { version = "0.4.45", features = ["clock", "serde"] } futures = "0.3.32" gateway-admin = { path = "../../gateway-admin" } diff --git a/backend/crates/providers/xai/src/provider.rs b/backend/crates/providers/xai/src/provider.rs index c0e912481..dac144992 100644 --- a/backend/crates/providers/xai/src/provider.rs +++ b/backend/crates/providers/xai/src/provider.rs @@ -989,7 +989,7 @@ async fn next_grok_chunk( selector: &dyn GrokSessionSelector, session: &SelectedGrokSession, context: &AttemptContext, -) -> Result>, ProviderError> { +) -> Result, ProviderError> { let Some(stream_deadline) = remaining(context.deadline()) else { return Err(provider_error( ProviderErrorKind::Timeout, diff --git a/backend/crates/providers/xai/src/transport/network.rs b/backend/crates/providers/xai/src/transport/network.rs index f8bdd054c..ff5fc0455 100644 --- a/backend/crates/providers/xai/src/transport/network.rs +++ b/backend/crates/providers/xai/src/transport/network.rs @@ -409,7 +409,8 @@ impl GrokInferenceTransport for ReqwestGrokInferenceTransport { .is_some_and(|total| total <= MAX_INFERENCE_BODY_BYTES) => { *observed += chunk.len(); - Ok(chunk.to_vec()) + // Bytes 直传,长流不再逐 chunk 复制到 Vec。 + Ok(chunk) } Ok(_) => Err(GrokInferenceTransportError::new( GrokInferenceTransportErrorKind::Protocol, diff --git a/backend/crates/providers/xai/src/transport/transport.rs b/backend/crates/providers/xai/src/transport/transport.rs index 523b6acdc..b4be4ce7c 100644 --- a/backend/crates/providers/xai/src/transport/transport.rs +++ b/backend/crates/providers/xai/src/transport/transport.rs @@ -77,7 +77,7 @@ impl fmt::Debug for GrokInferenceRequest { /// Stream of raw SSE byte chunks returned after the POST is accepted. pub type GrokInferenceChunkStream = - Pin, GrokInferenceTransportError>> + Send + 'static>>; + Pin> + Send + 'static>>; /// Whether the account-isolated inference client was already cached. #[derive(Debug, Clone, Copy, PartialEq, Eq)] diff --git a/backend/crates/providers/xai/tests/provider/contract.rs b/backend/crates/providers/xai/tests/provider/contract.rs index 66ddbf1fb..ec15e2ead 100644 --- a/backend/crates/providers/xai/tests/provider/contract.rs +++ b/backend/crates/providers/xai/tests/provider/contract.rs @@ -325,20 +325,20 @@ impl GrokInferenceTransport for StubInferenceTransport { Box::pin(async move { match mode { InferenceMode::Success => Ok(GrokInferenceResponse::new( - Box::pin(stream::iter([Ok(SUCCESS_SSE.to_vec())])), + Box::pin(stream::iter([Ok(bytes::Bytes::from_static(SUCCESS_SSE))])), UpstreamHttpVersion::Http2, 200, None, )), InferenceMode::SuccessWithMetrics(metrics) => Ok(GrokInferenceResponse::new( - Box::pin(stream::iter([Ok(SUCCESS_SSE.to_vec())])), + Box::pin(stream::iter([Ok(bytes::Bytes::from_static(SUCCESS_SSE))])), UpstreamHttpVersion::Http2, 200, None, ) .with_transport_metrics(metrics)), InferenceMode::SuccessBody(body) => Ok(GrokInferenceResponse::new( - Box::pin(stream::iter([Ok(body)])), + Box::pin(stream::iter([Ok(bytes::Bytes::from(body))])), UpstreamHttpVersion::Http2, 200, None,