From 3ef4743e1a26a46bede91d48f64cd65e93aa34ad Mon Sep 17 00:00:00 2001 From: Zyy Date: Mon, 12 Oct 2026 17:06:03 +0800 Subject: [PATCH] fix(store): persist retries after transport commit Let Core decide replay eligibility after WebSocket preludes or HTTP keepalives. Persist subsequent attempts while still enforcing the running lifecycle and consecutive attempt numbers, preserving delivery facts and the upstream send watermark. Cover retry metadata, duplicate and skipped attempts, and terminal request immutability with a PostgreSQL regression test. Co-authored-by: Codex --- .../src/postgres/execution/mod.rs | 6 +- backend/crates/gateway-store/src/value.rs | 1 - .../tests/postgres/execution/mod.rs | 113 ++++++++++++++++++ 3 files changed, 117 insertions(+), 3 deletions(-) diff --git a/backend/crates/gateway-store/src/postgres/execution/mod.rs b/backend/crates/gateway-store/src/postgres/execution/mod.rs index 6e054e1e9..242213910 100644 --- a/backend/crates/gateway-store/src/postgres/execution/mod.rs +++ b/backend/crates/gateway-store/src/postgres/execution/mod.rs @@ -513,6 +513,8 @@ impl ModelRequestRepository for PgExecutionStore { attempt: ModelRequestAttemptStart, ) -> StoreResult { attempt.validate()?; + // 重试资格由 Core 的重放屏障裁决，已交付前导事件或心跳不代表禁止新 attempt + // 观测存储只校验运行态与连续序号，保留首次交付时间和客户端状态 // upstream_send_state 是请求级单调水位（sent > ambiguous > not_sent）， // 由 mark/finalize 抬升；开启新 attempt 不得把已持久化的 sent 重置回 // not_sent——崩溃恢复的终态写回会原样继承本列 @@ -539,7 +541,7 @@ impl ModelRequestRepository for PgExecutionStore { 'capacityUsedSlots', coalesce($10::bigint, observed.used_slots), 'capacityTotalSlots', coalesce($11::bigint, observed.total_slots)))) from observed - where model_requests.id = observed.id and outcome = 'running' and downstream_committed_at is null + where model_requests.id = observed.id and outcome = 'running' and $8 = attempt_count + 1 returning attempt_count", ) @@ -573,7 +575,7 @@ impl ModelRequestRepository for PgExecutionStore { source: None, entity: ENTITY, id: attempt.model_request_id, - kind: ConflictKind::DownstreamAlreadyCommitted, + kind: ConflictKind::InvalidTransition, })?; u32::try_from(count) .map_err(|source| invalid("attempt_count is invalid").with_source(source)) diff --git a/backend/crates/gateway-store/src/value.rs b/backend/crates/gateway-store/src/value.rs index 03ff2816f..088af8893 100644 --- a/backend/crates/gateway-store/src/value.rs +++ b/backend/crates/gateway-store/src/value.rs @@ -15,7 +15,6 @@ pub enum ConflictKind { StaleRevision, DuplicateName, AlreadyFinalized, - DownstreamAlreadyCommitted, RequestNotRunning, InvalidTransition, LeaseLost, diff --git a/backend/crates/gateway-store/tests/postgres/execution/mod.rs b/backend/crates/gateway-store/tests/postgres/execution/mod.rs index a08de1d18..f5ff64494 100644 --- a/backend/crates/gateway-store/tests/postgres/execution/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/execution/mod.rs @@ -2241,6 +2241,119 @@ async fn shared_core_usage_and_timings_preserve_fields_and_reject_invalid_phases database.close().await; } +#[tokio::test] +async fn retry_after_transport_commit_preserves_delivery_facts_and_attempt_order() { + let Some(database) = TestDatabase::create("retry_after_commit").await else { + return; + }; + let store = PgExecutionStore::new(database.pool.clone()); + for (transport, status) in [("websocket", None), ("http_sse", Some(200))] { + let id = format!("req_retry_after_{transport}_commit"); + let mut request = accepted_request(&id); + request.client_transport = transport.to_owned(); + store.create_model_request(request).await.unwrap(); + let mut attempt = ModelRequestAttemptStart { + model_request_id: id.clone(), + attempt_count: 1, + provider_kind: "openai".to_owned(), + account_snapshot: None, + provider_account_ref: Some("account_before_retry".to_owned()), + upstream_model_id: Some("model".to_owned()), + upstream_transport: "websocket".to_owned(), + http_version: Some("HTTP/1.1".to_owned()), + account_selection_wait_ms: Some(5), + capacity_used_slots: Some(1), + capacity_total_slots: Some(10), + }; + store + .begin_model_request_attempt(attempt.clone()) + .await + .unwrap(); + store + .mark_upstream_send_state(&id, gateway_store::postgres::UpstreamSendState::Sent) + .await + .unwrap(); + // 可重放的 WS 前导事件或 HTTP 心跳先交付，不能替代 Core 的业务重放屏障 + let committed_at = Utc::now(); + ModelRequestRepository::mark_downstream_committed(&store, &id, committed_at, status) + .await + .unwrap(); + let committed = stored_row(&database.pool, &id).await; + attempt.attempt_count = 2; + attempt.provider_account_ref = Some("account_after_retry".to_owned()); + attempt.upstream_transport = "http_sse".to_owned(); + attempt.http_version = Some("HTTP/2".to_owned()); + attempt.account_selection_wait_ms = Some(7); + attempt.capacity_used_slots = Some(2); + assert_eq!( + store + .begin_model_request_attempt(attempt.clone()) + .await + .unwrap(), + 2 + ); + let retried = stored_row(&database.pool, &id).await; + assert_eq!( + retried["downstream_committed_at"], + committed["downstream_committed_at"] + ); + assert_eq!( + retried["client_status_code"], + committed["client_status_code"] + ); + assert_eq!(retried["upstream_send_state"], "sent"); + assert_eq!(retried["provider_account_ref"], "account_after_retry"); + assert_eq!(retried["upstream_transport"], "http_sse"); + assert_eq!( + retried["request_observation_json"]["scheduling"]["accountSelectionWaitMs"], + 12 + ); + assert_eq!( + retried["request_observation_json"]["scheduling"]["capacityUsedSlots"], + 2 + ); + assert_eq!( + retried["request_observation_json"]["transport"]["httpVersion"], + "HTTP/2" + ); + // 已接收的序号与跳号都不能覆盖新 attempt 的账号和计时 + for count in [2, 4] { + attempt.attempt_count = count; + let error = store + .begin_model_request_attempt(attempt.clone()) + .await + .unwrap_err(); + assert!(matches!( + error, + gateway_store::StoreError::Conflict { + kind: gateway_store::ConflictKind::InvalidTransition, + .. + } + )); + assert_eq!(stored_row(&database.pool, &id).await, retried); + } + let mut finalization = successful_core_finalization(&id); + finalization.attempt_count = 2; + finalization.downstream_committed_at = Some(committed_at.into()); + finalization.client_status_code = status; + finalization.upstream_transport = Some("http_sse".to_owned()); + ExecutionStore::finalize_model_request(&store, finalization) + .await + .unwrap(); + let terminal = stored_row(&database.pool, &id).await; + assert_eq!(terminal["outcome"], "succeeded"); + assert_eq!(terminal["attempt_count"], 2); + assert_eq!( + terminal["request_observation_json"]["scheduling"], + retried["request_observation_json"]["scheduling"] + ); + attempt.attempt_count = 3; + assert!(store.begin_model_request_attempt(attempt).await.is_err()); + assert_eq!(stored_row(&database.pool, &id).await, terminal); + } + database.close().await; +} + #[tokio::test] async fn concurrent_retry_updates_keep_request_context_and_clear_previous_transport() { let Some(database) = TestDatabase::create("observation_retry_cas").await else {