starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-06 08:06:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): remove precommit byte limit

Browse Source 
Keep large structural events buffered until semantic output, terminal completion, EOF, or the replay grace deadline. Retain byte counts for diagnostics and preserve historical byte_limit exports.

Refs #369

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-09 18:05:01 +08:00
co-authored by Codex
parent 53d4086cc0
commit d1febdd249
5 changed files with 55 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/providers/openai/src/provider/failure.rs
+2 -8
View File
@@ -84,10 +84,8 @@ pub(super) async fn wait_for_replay_grace(deadline: Option<Instant>) {
/// 提交边界前的上游事件预取
///
/// 原始 chunk 计数而不是重编码后的 event 大小
/// 时间与字节阈值共同限定无感换号
/// 窗口；任一边界到达都会提交已缓存 wire，不能因网关私有资源规则伪造上游协议
/// 失败
/// 结构事件保留到宽限期结束、语义输出、终态或 EOF，再提交已缓存 wire
/// 大段配置回显不应提前结束无感换号窗口，原始 chunk 字节数只用于诊断
/// 一旦提交，后续事件不再具备无痕重放资格
pub(super) struct PreCommitClientEvents {
pending: Vec<ProviderEvent>,
@@ -102,7 +100,6 @@ pub(super) struct PreCommitClientEvents {
pub(super) enum PreCommitReleaseReason {
SemanticOutput,
Terminal,
ByteLimit,
GraceTimeout,
Eof,
}
@@ -141,9 +138,6 @@ impl PreCommitClientEvents {
if completed {
return self.commit_pending(PreCommitReleaseReason::Terminal);
}
if self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES {
return self.commit_pending(PreCommitReleaseReason::ByteLimit);
}
if starts_replay_grace && self.replay_grace_started_at.is_none() {
self.replay_grace_started_at = Some(Instant::now());
}
backend/crates/providers/openai/src/provider/mod.rs
-4
View File
@@ -125,10 +125,6 @@ const WEBSOCKET_TRANSPORT: &str = "websocket";
// 在已观测到的 Codex OAuth 上游 16 MiB 附近消息边界前留出传输 metadata 余量
const WEBSOCKET_HTTP_FALLBACK_THRESHOLD_BYTES: usize = 15 * 1024 * 1024;
const MAX_COOKIE_HEADER_BYTES: usize = 16 * 1024;
/// 提交边界前预取 128 KiB 原始上游 chunk；容纳携带配置回显的前导事件，
/// 超过阈值后结束无感换号窗口（最后一个 chunk 可越过阈值），
/// 但不会把上游数据改写成协议失败
const MAX_STREAM_PREFETCH_BYTES: usize = 128 * 1024;
/// 短暂保留 response.created 等结构事件，让随后到达的明确拒绝可以无感换号；
/// 到期即放行，避免模型长时间思考时让客户端一直收不到首事件
const STREAM_REPLAY_GRACE: Duration = Duration::from_millis(2_500);
backend/crates/providers/openai/tests/provider/contract/mod.rs
+38 -11
View File
@@ -9578,7 +9578,7 @@ async fn exact_websocket_busy_then_replay_scope_relaxation_is_rejected_before_se
}
#[tokio::test]
async fn continuation_prefetch_over_128_kib_should_commit_wire_without_protocol_failure() {
async fn continuation_prefetch_over_128_kib_should_wait_for_grace_without_protocol_failure() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_prefetch_limit").await;
let padding = "x".repeat(128 * 1024);
@@ -9596,7 +9596,7 @@ async fn continuation_prefetch_over_128_kib_should_commit_wire_without_protocol_
);
assert!(body.len() > 128 * 1024);
let (base_url, release, _first_chunk_sent, server) =
paused_chunked_sse_server(body, String::new()).await;
paused_chunked_sse_server(body.clone(), String::new()).await;
let mut stream = provider_with_base_url(&store, base_url)
.execute(
planned_request("openai", http_generate_operation()),
@@ -9605,25 +9605,52 @@ async fn continuation_prefetch_over_128_kib_should_commit_wire_without_protocol_
)
.await
.expect("prepare provider stream");
let visible = loop {
let event = timeout(Duration::from_secs(1), stream.next())
let visible = {
let next_visible = async {
loop {
let event = stream
.next()
.await
.expect("provider stream remains open")
.expect("large prefetch cannot create a protocol failure");
if event.has_client_event() {
break event;
}
}
};
tokio::pin!(next_visible);
assert!(
timeout(Duration::from_secs(1), next_visible.as_mut())
.await
.is_err(),
"large structural events must remain buffered during the grace period"
);
timeout(Duration::from_secs(3), next_visible)
.await
.expect("prefetch threshold must release buffered wire")
.expect("provider stream remains open")
.expect("threshold cannot create a protocol failure");
if event.has_client_event() {
break event;
}
.expect("grace expiry must release buffered wire")
};
assert_eq!(
visible.wire_event().and_then(|wire| wire.event_type()),
Some("response.created")
);
let mut wire = visible
.wire_event()
.unwrap()
.raw_sse_frame()
.unwrap()
.to_vec();
release.send(()).expect("finish upstream response");
while let Some(event) = stream.next().await {
event.expect("clean upstream EOF cannot become a protocol failure");
if let Some(frame) = event
.expect("clean upstream EOF cannot become a protocol failure")
.wire_event()
.and_then(|wire| wire.raw_sse_frame())
{
wire.extend_from_slice(frame);
}
}
assert_eq!(wire, body.as_bytes());
server.await.expect("chunked SSE server");
}
backend/crates/providers/openai/tests/provider/contract/precommit.rs
+13 -19
View File
@@ -79,14 +79,13 @@ async fn stream_failure(stream: &mut ProviderStream) -> (Vec<ProviderEvent>, Pro
}
#[tokio::test]
async fn large_structural_events_preserve_overload_replay_past_the_old_grace_on_http_and_websocket()
{
async fn large_structural_events_preserve_overload_replay_on_http_and_websocket() {
for websocket in [false, true] {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
// #259 仅保留了事件长度；用合成配置回显复现尺寸，不依赖现场私有正文
let created = structural_event("response.created", 38_781);
let progress = structural_event("response.in_progress", 38_785);
// 大段配置回显不应提前结束重试窗口；使用合成正文覆盖单帧与累计的大体积前导事件
let created = structural_event("response.created", 300 * 1024);
let progress = structural_event("response.in_progress", 300 * 1024);
let failure = overload();
let expected = vec![created.clone(), progress.clone(), failure.clone()];
let (base_url, release, server) = if websocket {
@@ -160,8 +159,8 @@ async fn large_structural_events_preserve_overload_replay_past_the_old_grace_on_
async fn later_structural_events_do_not_extend_grace_and_late_overload_is_not_replayable() {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
let created = sse(&structural_event("response.created", 38_781));
let progress = sse(&structural_event("response.in_progress", 38_785));
let created = sse(&structural_event("response.created", 96 * 1024));
let progress = sse(&structural_event("response.in_progress", 96 * 1024));
let expected_bytes = created.len() + progress.len();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
@@ -220,21 +219,20 @@ async fn immediate_release_preserves_wire_and_records_the_boundary_once() {
for (reason, body) in [
(
"semantic_output",
sse(&structural_event("response.created", 512)) + &sse(&semantic),
sse(&structural_event("response.created", 300 * 1024)) + &sse(&semantic),
),
(
"semantic_output",
sse(&structural_event("response.created", 512)) + &sse(&tool),
sse(&structural_event("response.created", 300 * 1024)) + &sse(&tool),
),
(
"terminal",
sse(&structural_event("response.created", 512)) + &sse(&terminal),
sse(&structural_event("response.created", 300 * 1024)) + &sse(&terminal),
),
(
"byte_limit",
sse(&structural_event("response.created", 129 * 1024)),
"eof",
sse(&structural_event("response.created", 300 * 1024)),
),
("eof", sse(&structural_event("response.created", 512))),
] {
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
@@ -274,11 +272,7 @@ async fn immediate_release_preserves_wire_and_records_the_boundary_once() {
let releases = releases(&trace);
assert_eq!(releases.len(), 1);
assert_eq!(releases[0]["data"]["reason"], reason);
if reason == "byte_limit" {
assert!(releases[0]["data"]["prefetchedBytes"].as_u64().unwrap() > 128 * 1024);
} else {
assert_eq!(releases[0]["data"]["prefetchedBytes"], body.len());
}
assert_eq!(releases[0]["data"]["prefetchedBytes"], body.len());
server.await.unwrap();
}
}
@@ -288,7 +282,7 @@ async fn cancelling_buffered_structural_events_does_not_release_or_offer_replay(
let store = Arc::new(MemoryAccountStore::default());
create_account(&store, "acct_provider_contract").await;
let (base_url, release, _, server) = paused_chunked_sse_server(
sse(&structural_event("response.created", 38_781)),
sse(&structural_event("response.created", 300 * 1024)),
String::new(),
)
.await;
docs/api.md
+2 -1
View File
@@ -1855,7 +1855,8 @@ WebSocket 握手默认请求专项计时，客户端显式提供的开关保留
请求与错误事件各自的状态、attempt、时间线阶段、计时和上述失败分类。
`provider.precommit.released` 事件的 `precommitRelease` 保留 `reason / prefetchedBytes / waitMs`，
分别表示 Provider 释放缓存的原因、累计预取字节数和等待毫秒数，不等同于下游已提交。
原因只允许 `semantic_output / terminal / byte_limit / grace_timeout / eof`，数值只允许非负安全整数，
当前释放原因为 `semantic_output / terminal / grace_timeout / eof`，导出也接受历史记录中的 `byte_limit`。
累计字节数只用于诊断，不触发缓存释放；数值只允许非负安全整数，
缺失或未通过校验的字段为 `null`；其他事件的 `precommitRelease` 为 `null`
诊断包不自动导出 message/raw error、任意 metadata、其他 trace event data、请求响应正文和头部。