mirror of
https://github.com/zyycn/codex-proxy-rs.git
synced 2026-11-13 08:06:41 +00:00
fix(scheduling): log queue fallback binding migrations
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
@@ -480,17 +480,18 @@ impl CodexCredentialSelector {
|
||||
&& request
|
||||
.session_affinity_observation
|
||||
.is_some_and(CodexSessionAffinity::follow_only);
|
||||
let select = |queue_policy| {
|
||||
let select = |queue_policy, queue_rejection| {
|
||||
self.select_inner_loop(
|
||||
request,
|
||||
cyber_policy_session_key,
|
||||
model,
|
||||
follow_only,
|
||||
queue_policy,
|
||||
queue_rejection,
|
||||
)
|
||||
};
|
||||
let selection = async {
|
||||
let rejection = match select(policy.queue_policy()).await {
|
||||
let rejection = match select(policy.queue_policy(), None).await {
|
||||
Err(CredentialSelectionError::QueueRejected(reason))
|
||||
if !follow_only
|
||||
&& request.attempt.required_account().is_none()
|
||||
@@ -512,7 +513,7 @@ impl CodexCredentialSelector {
|
||||
"account.queue.reselect",
|
||||
serde_json::json!({"reason": rejection.to_string()}),
|
||||
);
|
||||
select(queue)
|
||||
select(queue, Some(rejection))
|
||||
},
|
||||
|err| {
|
||||
matches!(
|
||||
@@ -555,6 +556,7 @@ impl CodexCredentialSelector {
|
||||
model: CodexSelectionModel<'_>,
|
||||
follow_only: bool,
|
||||
mut queue_policy: ConcurrencyQueuePolicy,
|
||||
queue_rejection: Option<QueueRejection>,
|
||||
) -> Result<CodexCredentialLease, CredentialSelectionError> {
|
||||
let adapter_model = match model {
|
||||
CodexSelectionModel::Requested(model) if request.managed_upstream => Some(
|
||||
@@ -1351,7 +1353,30 @@ impl CodexCredentialSelector {
|
||||
tracing::warn!(%error, "OpenAI cache preference update unavailable");
|
||||
}
|
||||
Err(error) => return Err(error),
|
||||
Ok(true) => {}
|
||||
Ok(true) => {
|
||||
if let Some(reason) = queue_rejection
|
||||
&& let Some(previous) = binding.as_ref().filter(|binding| {
|
||||
binding.account_id() != binding_account
|
||||
})
|
||||
{
|
||||
tracing::info!(
|
||||
request_id = %request.attempt.request_id(),
|
||||
attempt_index = request.attempt.attempt_index().get(),
|
||||
reason = %reason,
|
||||
from_account_id = %previous.account_id(),
|
||||
to_account_id = %binding_account,
|
||||
"OpenAI 账号排队失败后已迁移绑定"
|
||||
);
|
||||
request.attempt.trace().record_evidence(
|
||||
"account.binding.migrated",
|
||||
serde_json::json!({
|
||||
"reason": reason.to_string(),
|
||||
"fromAccountId": previous.account_id().as_str(),
|
||||
"toAccountId": binding_account.as_str(),
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let affinity_telemetry = affinity.telemetry(account.id(), cache_preference);
|
||||
|
||||
@@ -18,7 +18,10 @@ fn projection_attempt(
|
||||
RequestAttemptContext::new(
|
||||
ModelRequestId::new("req_projection").unwrap(),
|
||||
ClientApiKeyId::new(client_key).unwrap(),
|
||||
),
|
||||
)
|
||||
.with_trace(gateway_core::diagnostics::TraceContext::new(
|
||||
"req_projection",
|
||||
)),
|
||||
NonZeroU32::new(1).unwrap(),
|
||||
SystemTime::now() + Duration::from_secs(30),
|
||||
policy,
|
||||
@@ -420,15 +423,35 @@ async fn strict_queue_failure_reselects_once_and_only_rebinds_after_admission()
|
||||
assert!(pending.as_mut().now_or_never().is_none());
|
||||
}
|
||||
let result = pending.await;
|
||||
let reason = if full {
|
||||
QueueRejection::Full
|
||||
} else {
|
||||
QueueRejection::Timeout
|
||||
};
|
||||
let snapshot = ctx.trace().snapshot().unwrap();
|
||||
let migrations = snapshot["events"]
|
||||
.as_array()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter(|event| event["stage"] == "account.binding.migrated")
|
||||
.map(|event| &event["data"])
|
||||
.collect::<Vec<_>>();
|
||||
if spare && enabled {
|
||||
assert_eq!(
|
||||
migrations,
|
||||
vec![&json!({
|
||||
"reason": reason.to_string(),
|
||||
"fromAccountId": first.as_str(),
|
||||
"toAccountId": second.as_str(),
|
||||
})]
|
||||
);
|
||||
} else {
|
||||
assert!(migrations.is_empty());
|
||||
}
|
||||
let expected = if spare && enabled {
|
||||
assert_eq!(result.unwrap().account_id(), &second);
|
||||
&second
|
||||
} else {
|
||||
let reason = if full {
|
||||
QueueRejection::Full
|
||||
} else {
|
||||
QueueRejection::Timeout
|
||||
};
|
||||
assert!(
|
||||
matches!(result, Err(CredentialSelectionError::QueueRejected(actual)) if actual == reason)
|
||||
);
|
||||
|
||||
@@ -2762,6 +2762,10 @@ WebSocket 握手不主动添加专项计时请求头,客户端显式提供的
|
||||
`environment.scope = export_time` 明确版本、构建与部署信息是导出时状态,不能代替事发环境
|
||||
|
||||
`request.routed.data.accountPolicy` 保存执行时冻结的账号策略与预算。
|
||||
OpenAI 账号队满或超时后的重选记录为 `account.queue.reselect`;绑定迁移成功后,
|
||||
`account.binding.migrated` 保存拒绝原因 `reason`、原账号 `fromAccountId` 和新账号 `toAccountId`。
|
||||
首次认领、临时分流和失败重选不记录为绑定迁移
|
||||
|
||||
`attempt.failed.data.recovery` 保存 Provider 的重放安全性和恢复意图;`retry.decided` 保存当时
|
||||
实际参与决策的门槛、剩余预算和发送状态,`retry.policy` 保存宿主及插件策略的停止事实。
|
||||
已有的 `provider.precommit.released` 数据保留释放原因、预取字节数与等待时间,不等同于下游已提交
|
||||
|
||||
Reference in New Issue
Block a user