perf(openai,xai): trim passive quota round trips and stream chunk copies

Drop the account re-read from passive rate-limit quota sync: the quota CAS is fenced on the account revision and the follow-up state write carries the same fence, so the availability transition can use the selection-time snapshot and converge on the next response. Stream Grok inference chunks as Bytes end to end instead of copying every chunk into a fresh Vec.

Co-authored-by: Claude <noreply@anthropic.com>

backend/Cargo.lock

@@ -2158,6 +2158,7 @@ dependencies = [
  "async-stream",
  "async-trait",
  "base64",
+ "bytes",
  "chrono",
  "futures",
  "gateway-admin",

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(
-                    &current,
+                    account,
                     availability,
                     fact.exhausted().then_some("quota_exhausted".to_owned()),
                     None,

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" }

backend/crates/providers/xai/src/provider.rs

@@ -989,7 +989,7 @@ async fn next_grok_chunk(
     selector: &dyn GrokSessionSelector,
     session: &SelectedGrokSession,
     context: &AttemptContext,
-) -> Result<Option<Vec<u8>>, ProviderError> {
+) -> Result<Option<bytes::Bytes>, ProviderError> {
     let Some(stream_deadline) = remaining(context.deadline()) else {
         return Err(provider_error(
             ProviderErrorKind::Timeout,

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,

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<Box<dyn Stream<Item = Result<Vec<u8>, GrokInferenceTransportError>> + Send + 'static>>;
+    Pin<Box<dyn Stream<Item = Result<bytes::Bytes, GrokInferenceTransportError>> + Send + 'static>>;
 
 /// Whether the account-isolated inference client was already cached.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]

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,