refactor(dispatch): enforce single-owner failure decisions

Move Cyber classification and retry decisions behind its controller-owned typed result.

Share feature failure owner ordering across attempt and committed stream paths, with architecture guards and documentation.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
zyycn
2026-07-14 18:35:06 +08:00
co-authored by Codex
parent 0384e39289
commit 83f2f55e4a
5 changed files with 257 additions and 139 deletions
@@ -11,8 +11,11 @@ mod types;
use crate::{
dispatch::{
affinity::SessionAffinityService, controllers::ControllerFailureFact,
errors::ClientFailure, transport::observation::UpstreamFailureFacts,
affinity::SessionAffinityService,
controllers::ControllerFailureFact,
errors::ClientFailure,
lifecycle::contract::{AttemptDecision, AttemptObservation, AttemptReturnKind},
transport::observation::UpstreamFailureFacts,
},
upstream::openai::protocol::responses::{CodexResponsesRequest, ResponsesSseFailure},
};
@@ -48,6 +51,11 @@ pub(in crate::dispatch) struct CyberPolicyController {
store: Store,
}
/// 已由 Cyber owner 认领的失败;调用方只能请求 owner 产生 effect 或 decision。
pub(super) struct ClassifiedCyberPolicyFailure<'a> {
fact: ControllerFailureFact<'a>,
}
impl CyberPolicyController {
pub(in crate::dispatch) fn new(session_affinity: Arc<SessionAffinityService>) -> Self {
Self {
@@ -55,18 +63,46 @@ impl CyberPolicyController {
}
}
pub(super) fn client_failure(failure: ResponsesSseFailure) -> ClientFailure {
ClientFailure::new(failure, 400, true)
pub(super) fn classify(
fact: ControllerFailureFact<'_>,
) -> Option<ClassifiedCyberPolicyFailure<'_>> {
let owned = match fact {
ControllerFailureFact::Upstream(facts) => {
facts
.status_code
.is_some_and(|status| (400..=499).contains(&status))
&& is_cyber_policy_code(facts.code.as_deref())
}
ControllerFailureFact::Response(failure) => is_cyber_policy_failure(failure),
};
owned.then_some(ClassifiedCyberPolicyFailure { fact })
}
pub(super) fn attempt_client_failure(failure: ControllerFailureFact<'_>) -> ClientFailure {
match failure {
pub(super) fn client_failure(classified: &ClassifiedCyberPolicyFailure<'_>) -> ClientFailure {
match classified.fact {
ControllerFailureFact::Upstream(facts) => ClientFailure::new(
failure_from_upstream_facts(facts),
facts.status_code.unwrap_or(400),
true,
),
ControllerFailureFact::Response(failure) => Self::client_failure(failure.clone()),
ControllerFailureFact::Response(failure) => {
ClientFailure::new(failure.clone(), 400, true)
}
}
}
pub(super) fn decision(
observation: &AttemptObservation,
classified: ClassifiedCyberPolicyFailure<'_>,
) -> AttemptDecision {
let client_failure = Self::client_failure(&classified);
if observation.routing.can_retry_next_candidate {
AttemptDecision::RetryNextCandidate {
exhaustion: None,
on_exhaustion: Some(client_failure),
}
} else {
AttemptDecision::Return(AttemptReturnKind::Failed(client_failure))
}
}
@@ -122,18 +158,6 @@ impl CyberPolicyController {
}
}
pub(super) fn owns_failure(&self, failure: ControllerFailureFact<'_>) -> bool {
match failure {
ControllerFailureFact::Upstream(facts) => {
facts
.status_code
.is_some_and(|status| (400..=499).contains(&status))
&& is_cyber_policy_code(facts.code.as_deref())
}
ControllerFailureFact::Response(failure) => is_cyber_policy_failure(failure),
}
}
pub(in crate::dispatch) async fn observe_success(&self, plan: &CyberPolicyScope) {
if !plan.has_failures() {
return;
+144 -116
View File
@@ -32,7 +32,9 @@ use crate::{
use crate::{
dispatch::{
controllers::cloudflare::CloudflareRecovery,
controllers::cyber_policy::{CyberPolicyController, CyberPolicyScope},
controllers::cyber_policy::{
ClassifiedCyberPolicyFailure, CyberPolicyController, CyberPolicyScope,
},
controllers::{
account_failure::AccountFailureController, history::HistoryController,
history::HistoryState, quota::QuotaController,
@@ -168,6 +170,30 @@ enum ControllerFailureFact<'a> {
Response(&'a ResponsesSseFailure),
}
#[derive(Clone, Copy)]
enum FailureObservation<'a> {
Attempt(&'a AttemptObservation),
Stream {
account_id: &'a str,
failure: &'a ResponsesSseFailure,
},
}
impl<'a> FailureObservation<'a> {
fn fact(self) -> Option<ControllerFailureFact<'a>> {
match self {
Self::Attempt(observation) => ControllerFailureFact::from_attempt(observation),
Self::Stream { failure, .. } => Some(ControllerFailureFact::Response(failure)),
}
}
}
enum SharedFailureClassification<'a> {
CyberPolicy(ClassifiedCyberPolicyFailure<'a>),
AccountFailure(account_failure::ClassifiedFailure),
Quota(quota::ClassifiedQuotaFailure),
}
impl<'a> ControllerFailureFact<'a> {
fn from_attempt(observation: &'a AttemptObservation) -> Option<Self> {
match &observation.kind {
@@ -183,40 +209,30 @@ impl<'a> ControllerFailureFact<'a> {
/// 单次 attempt 的唯一分类结果;副作用与 decision 必须共同消费这个值。
enum AttemptClassification<'a> {
CyberPolicy(ControllerFailureFact<'a>),
Shared(SharedFailureClassification<'a>),
Cloudflare(cloudflare::CloudflareFailure),
History(AttemptDecision),
AccountFailure(account_failure::ClassifiedFailure),
Quota(quota::ClassifiedQuotaFailure),
Default(AttemptDecision),
}
enum StreamFailureClassification {
CyberPolicy(ClientFailure),
AccountFailure {
classified: account_failure::ClassifiedFailure,
client_failure: ClientFailure,
},
Quota {
classified: quota::ClassifiedQuotaFailure,
client_failure: ClientFailure,
},
Default(ClientFailure),
struct StreamFailureClassification<'a> {
failure: &'a ResponsesSseFailure,
owner: Option<SharedFailureClassification<'a>>,
}
impl StreamFailureClassification {
fn client_failure(&self) -> &ClientFailure {
match self {
Self::CyberPolicy(failure)
| Self::AccountFailure {
client_failure: failure,
..
impl StreamFailureClassification<'_> {
fn client_failure(&self) -> ClientFailure {
match &self.owner {
Some(SharedFailureClassification::CyberPolicy(classified)) => {
CyberPolicyController::client_failure(classified)
}
| Self::Quota {
client_failure: failure,
..
Some(SharedFailureClassification::AccountFailure(classified)) => {
AccountFailureController::client_failure(classified, self.failure)
}
| Self::Default(failure) => failure,
Some(SharedFailureClassification::Quota(_)) => {
QuotaController::client_failure(self.failure)
}
None => default_response_client_failure(self.failure),
}
}
}
@@ -523,60 +539,47 @@ impl ControllerSet {
}
}
fn classify_stream_failure(
fn classify_stream_failure<'a>(
&self,
account_id: &str,
failure: &ResponsesSseFailure,
) -> StreamFailureClassification {
let fact = ControllerFailureFact::Response(failure);
if self.cyber_policy.owns_failure(fact) {
return StreamFailureClassification::CyberPolicy(
CyberPolicyController::client_failure(failure.clone()),
);
account_id: &'a str,
failure: &'a ResponsesSseFailure,
) -> StreamFailureClassification<'a> {
StreamFailureClassification {
failure,
owner: Self::classify_shared_failure(FailureObservation::Stream {
account_id,
failure,
}),
}
if let Some(classified) = AccountFailureController::classify_failure(account_id, fact) {
let client_failure = AccountFailureController::client_failure(&classified, failure);
return StreamFailureClassification::AccountFailure {
classified,
client_failure,
};
}
if let Some(classified) = QuotaController::classify_failure(account_id, fact) {
return StreamFailureClassification::Quota {
classified,
client_failure: QuotaController::client_failure(failure),
};
}
StreamFailureClassification::Default(default_response_client_failure(failure))
}
async fn apply_stream_failure(
&self,
scope: &ControllerRequestScope,
account_id: &str,
classification: &StreamFailureClassification,
classification: &StreamFailureClassification<'_>,
) {
match classification {
StreamFailureClassification::CyberPolicy(_) => {
match classification.owner.as_ref() {
Some(SharedFailureClassification::CyberPolicy(_)) => {
self.cyber_policy
.exclude_account(&scope.cyber_policy, account_id)
.await;
}
StreamFailureClassification::AccountFailure { classified, .. } => {
Some(SharedFailureClassification::AccountFailure(classified)) => {
let _ = best_effort_controller_io(
"account_failure.stream",
AccountFailureController::apply_effect(&self.account_pool, classified),
)
.await;
}
StreamFailureClassification::Quota { classified, .. } => {
Some(SharedFailureClassification::Quota(classified)) => {
let _ = best_effort_controller_io(
"quota.stream",
QuotaController::apply_effect(&self.account_pool, classified),
)
.await;
}
StreamFailureClassification::Default(_) => {}
None => {}
}
}
@@ -714,17 +717,33 @@ impl ControllerSet {
}
match &classification {
AttemptClassification::CyberPolicy(_) => {
if let Some(account_id) = observation
.account
.as_ref()
.map(|account| account.id.as_str())
{
self.cyber_policy
.exclude_account(&scope.cyber_policy, account_id)
.await;
AttemptClassification::Shared(shared) => match shared {
SharedFailureClassification::CyberPolicy(_) => {
if let Some(account_id) = observation
.account
.as_ref()
.map(|account| account.id.as_str())
{
self.cyber_policy
.exclude_account(&scope.cyber_policy, account_id)
.await;
}
}
}
SharedFailureClassification::AccountFailure(classified) => {
let _ = best_effort_controller_io(
"account_failure.attempt",
AccountFailureController::apply_effect(&self.account_pool, classified),
)
.await;
}
SharedFailureClassification::Quota(classified) => {
let _ = best_effort_controller_io(
"quota.attempt",
QuotaController::apply_effect(&self.account_pool, classified),
)
.await;
}
},
AttemptClassification::Cloudflare(failure) => {
let _ = best_effort_controller_io(
"cloudflare.attempt",
@@ -736,20 +755,6 @@ impl ControllerSet {
)
.await;
}
AttemptClassification::AccountFailure(classified) => {
let _ = best_effort_controller_io(
"account_failure.attempt",
AccountFailureController::apply_effect(&self.account_pool, classified),
)
.await;
}
AttemptClassification::Quota(classified) => {
let _ = best_effort_controller_io(
"quota.attempt",
QuotaController::apply_effect(&self.account_pool, classified),
)
.await;
}
AttemptClassification::History(_) | AttemptClassification::Default(_) => {}
}
@@ -760,44 +765,83 @@ impl ControllerSet {
&self,
observation: &'a AttemptObservation,
) -> AttemptClassification<'a> {
let failure = ControllerFailureFact::from_attempt(observation);
if let Some(failure) = failure
&& self.cyber_policy.owns_failure(failure)
{
return AttemptClassification::CyberPolicy(failure);
}
let shared = Self::classify_shared_failure(FailureObservation::Attempt(observation));
let shared = match shared {
Some(shared @ SharedFailureClassification::CyberPolicy(_)) => {
return AttemptClassification::Shared(shared);
}
shared => shared,
};
if let Some(classified) = cloudflare::CloudflareController::classify(observation) {
return AttemptClassification::Cloudflare(classified);
}
if let Some(decision) = HistoryController::decide(observation) {
return AttemptClassification::History(decision);
}
if let Some(classified) = AccountFailureController::classify(observation) {
return AttemptClassification::AccountFailure(classified);
}
if let Some(classified) = QuotaController::classify(observation) {
return AttemptClassification::Quota(classified);
if let Some(shared) = shared {
return AttemptClassification::Shared(shared);
}
AttemptClassification::Default(Self::default_attempt_decision(observation))
}
/// complete、prefetch 与 committed stream 共用这一份 feature owner 优先级。
fn classify_shared_failure<'a>(
observation: FailureObservation<'a>,
) -> Option<SharedFailureClassification<'a>> {
let fact = observation.fact();
if let Some(classified) = fact.and_then(CyberPolicyController::classify) {
return Some(SharedFailureClassification::CyberPolicy(classified));
}
let account_failure = match observation {
FailureObservation::Attempt(observation) => {
AccountFailureController::classify(observation)
}
FailureObservation::Stream { account_id, .. } => {
fact.and_then(|fact| AccountFailureController::classify_failure(account_id, fact))
}
};
if let Some(classified) = account_failure {
return Some(SharedFailureClassification::AccountFailure(classified));
}
let quota = match observation {
FailureObservation::Attempt(observation) => QuotaController::classify(observation),
FailureObservation::Stream { account_id, .. } => {
fact.and_then(|fact| QuotaController::classify_failure(account_id, fact))
}
};
quota.map(SharedFailureClassification::Quota)
}
fn attempt_decision(
&self,
observation: &AttemptObservation,
classification: AttemptClassification<'_>,
) -> AttemptDecision {
match classification {
AttemptClassification::CyberPolicy(failure) => {
let client_failure = CyberPolicyController::attempt_client_failure(failure);
if observation.routing.can_retry_next_candidate {
AttemptDecision::RetryNextCandidate {
exhaustion: None,
on_exhaustion: Some(client_failure),
}
} else {
AttemptDecision::Return(AttemptReturnKind::Failed(client_failure))
AttemptClassification::Shared(shared) => match shared {
SharedFailureClassification::CyberPolicy(classified) => {
CyberPolicyController::decision(observation, classified)
}
}
SharedFailureClassification::AccountFailure(classified) => {
let client_failure = response_failure(observation).map(|failure| {
AccountFailureController::client_failure(&classified, failure)
});
with_response_client_failure(
AccountFailureController::decision(observation, classified),
client_failure,
)
}
SharedFailureClassification::Quota(classified) => {
let client_failure =
response_failure(observation).map(QuotaController::client_failure);
with_response_client_failure(
QuotaController::decision(observation, classified),
client_failure,
)
}
},
AttemptClassification::Cloudflare(classified) => {
cloudflare::CloudflareController::decision(observation, classified)
}
@@ -807,22 +851,6 @@ impl ControllerSet {
.map(|failure| ClientFailure::new(failure.clone(), 400, false)),
),
AttemptClassification::Default(decision) => decision,
AttemptClassification::AccountFailure(classified) => {
let client_failure = response_failure(observation)
.map(|failure| AccountFailureController::client_failure(&classified, failure));
with_response_client_failure(
AccountFailureController::decision(observation, classified),
client_failure,
)
}
AttemptClassification::Quota(classified) => {
let client_failure =
response_failure(observation).map(QuotaController::client_failure);
with_response_client_failure(
QuotaController::decision(observation, classified),
client_failure,
)
}
}
}
+64
View File
@@ -42,6 +42,51 @@ fn cyber_policy_semantics_must_stay_inside_its_owner() {
);
}
#[test]
fn cyber_policy_owner_should_construct_its_own_attempt_decision() {
let manifest = Path::new(env!("CARGO_MANIFEST_DIR"));
let composition = read(&manifest.join("src/dispatch/controllers/mod.rs"));
let cyber_policy = read(&manifest.join("src/dispatch/controllers/cyber_policy/mod.rs"));
let decision = between(
&composition,
"fn attempt_decision(",
"fn default_attempt_decision(",
);
assert!(cyber_policy.contains("pub(super) fn decision("));
assert!(cyber_policy.contains("AttemptDecision::RetryNextCandidate"));
assert!(decision.contains("CyberPolicyController::decision(observation, classified)"));
assert!(!decision.contains("can_retry_next_candidate"));
assert!(!decision.contains("AttemptDecision::RetryNextCandidate"));
}
#[test]
fn attempt_and_stream_should_share_one_feature_failure_owner_order() {
let manifest = Path::new(env!("CARGO_MANIFEST_DIR"));
let composition = read(&manifest.join("src/dispatch/controllers/mod.rs"));
let shared = between(
&composition,
"fn classify_shared_failure",
"fn attempt_decision(",
);
assert_eq!(composition.matches("fn classify_shared_failure").count(), 1);
assert!(
composition
.contains("Self::classify_shared_failure(FailureObservation::Attempt(observation))")
);
assert!(composition.contains("Self::classify_shared_failure(FailureObservation::Stream {"));
assert_appears_in_order(
shared,
&[
"fact.and_then(CyberPolicyController::classify)",
"AccountFailureController::classify(observation)",
"QuotaController::classify(observation)",
],
);
assert!(!composition.contains("self.cyber_policy.owns_failure"));
}
#[test]
fn request_enter_io_should_be_parallel_and_bounded() {
let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("src/dispatch/controllers");
@@ -272,6 +317,25 @@ fn collect_live_executor_dependencies(root: &Path, path: &Path, violations: &mut
}
}
fn between<'a>(source: &'a str, start: &str, end: &str) -> &'a str {
let start = source
.find(start)
.expect("source start marker should exist");
let tail = &source[start..];
let end = tail.find(end).expect("source end marker should exist");
&tail[..end]
}
fn assert_appears_in_order(source: &str, required: &[&str]) {
let mut cursor = 0;
for marker in required {
let offset = source[cursor..]
.find(marker)
.unwrap_or_else(|| panic!("ordered source marker should exist: {marker}"));
cursor += offset + marker.len();
}
}
fn read(path: &Path) -> String {
fs::read_to_string(path).unwrap_or_else(|error| panic!("read {}: {error}", path.display()))
}
+1 -1
View File
@@ -218,7 +218,7 @@ fleet/
`v1/*` 使用 Request、Attempt、Stream 三层洋葱生命周期。核心代码位于 `lifecycle/`、`controllers/`、`routing/`、`transport/` 和 `failure/`;`service.rs` 与 `stream/lifecycle.rs` 只是 complete/stream 入口适配器,`stream/live.rs` 是响应提交后的流执行器,不承载功能策略。
架构硬规则是“单功能单归属”:功能阈值、状态机、优先级和 decision 只能由对应 controller 拥有;`lifecycle` 只管固定顺序、retry、commit、逆序退出和 exactly-once finalize。迁移不保留 compatibility re-export、feature flag 切换或新旧双路径,新路径接管后必须在同一次变更中删除旧入口和重复判断。
架构硬规则是“单功能单归属”:功能阈值、状态机和 decision 只能由对应 controller 拥有;`ControllerSet` 只声明 controller 的固定优先级并分发其 typed classification、effect 与 decision,不能替 owner 构造功能专用 decision;`lifecycle` 只管 retry、commit、逆序退出和 exactly-once finalize。complete、prefetch 与 committed stream 必须共用一份 feature failure owner 顺序。迁移不保留 compatibility re-export、feature flag 切换或新旧双路径,新路径接管后必须在同一次变更中删除旧入口和重复判断。
```text
dispatch/
+5 -3
View File
@@ -44,14 +44,14 @@
边界必须满足以下规则:
- `controller` 唯一拥有功能规则、状态机、优先级、阈值、typed facts、decision 和 scope;不能选择下一个账号、编码 HTTP/SSE/WebSocket 或直接调用其他 controller。
- `controller` 唯一拥有本功能的规则、状态机、阈值、typed classification、decision 和 scope;不能选择具体的下一个账号、编码 HTTP/SSE/WebSocket 或直接调用其他 controller。跨 controller 的唯一固定优先级由 `ControllerSet` 声明。
- `lifecycle` 只拥有固定顺序、enter/leave、短路、逆序 unwind、retry 循环、commit 边界和 exactly-once finalize;不得出现 `if cyber_policy` 或功能阈值。
- `transport` / `upstream` 只把原始 HTTP、SSE、WebSocket 输入规范化为通用事实,例如状态码、错误码、terminal、usage;不得解释事实的业务后果。
- `api` 只解析请求,并编码 `ResponseDispatchResponse`、`ResponseDispatchStream` 或包含 `ClientFailure` 的 `ResponseDispatchError`;不得重新识别 cyber、quota、history 等功能。
- `store` 只提供原子读写、CAS、TTL 和数据映射;业务专用 store trait 可以由 controller 定义,基础设施实现不能拥有业务阈值和状态转换。
- controller 之间不得横向调用;共享内容只能是没有业务含义的协议事实和值对象,不能把共享业务规则藏进 `utils`、`common` 或 helper。
允许的唯一挂载点是静态 composition root `ControllerSet`:它负责构造 controller、注入依赖,并通过类型化方法分发各阶段 hook。业务代码不在各个入口或尝试循环中手工挂载功能。
允许的唯一挂载点是静态 composition root `ControllerSet`:它负责构造 controller、注入依赖、声明固定优先级,并通过类型化方法分发各阶段 hook。它只能调用 owner 暴露的分类、effect 与 decision,不能替 owner 构造功能专用 decision。业务代码不在各个入口或尝试循环中手工挂载功能。
## 已实现目录与边界
@@ -218,7 +218,7 @@ controller 分为两类:
8. 按唯一优先级分类 history、session policy、账号隔离和 transport retry。
9. 产生 `AttemptDecision`,由唯一 pipeline 应用为 `AttemptApplyOutcome` 并记录 attempt trace。
classifier 优先级只能存在一处。例如 `cyber_policy` 必须先于通用 403、quota 或 5xx 分类;其他模块消费分类后的类型,不能再次调用 `is_cyber_policy_*`。
classifier 优先级只能存在一处。complete、prefetch 与 committed stream 共同经过同一个 feature failure owner classifier;`cyber_policy` 必须先于通用 403、quota 或 5xx 分类。composition root 只能把不透明的分类结果交还 owner 产生 decision,其他模块不能再次调用 `is_cyber_policy_*` 或自行拼装 Cyber retry。
### Stream enter/leave
@@ -248,6 +248,8 @@ classifier 优先级只能存在一处。例如 `cyber_policy` 必须先于通
- complete 与 stream 入口共享 `enter_request` 和 `AttemptRunner`,不存在各自维护的账号尝试主循环。
- `ControllerSet` 是静态 composition root;controller 顺序和 typed hook 在编译期可见,不使用动态 middleware 链。
- complete、prefetch 与 committed stream 的公共失败 owner 优先级只在 `ControllerSet::classify_shared_failure` 声明一次;composition root 不实现 Cyber、quota 或账号失败规则。
- Cyber failure 由 `CyberPolicyController` 返回不透明分类,并由该 owner 唯一构造换号或返回 decision;`ControllerSet` 只分发 effect 与 decision。
- `ControllerRequestScope` 只保存 cyber、history、usage 的真实状态;不存在通用 entered-controller marker、容量错误或空 unwind hook。
- complete 的 accepted response 归一为 `FinalOutcome` 后由 `finalize_complete` 消费 scope;retry 继续进入下一 attempt,rejection 在 controller effect 已完成后直接返回,不经过空 finalize。
- complete 的真实退出顺序固定为 Cloudflare → Usage → Affinity → Telemetry;live stream 固定为 Cloudflare → Usage → Affinity → Telemetry → Cyber,无副作用的阶段不制造空分支。