diff --git a/backend/crates/gateway-store/src/postgres/execution.rs b/backend/crates/gateway-store/src/postgres/execution.rs index d489fc3b3..8853d671f 100644 --- a/backend/crates/gateway-store/src/postgres/execution.rs +++ b/backend/crates/gateway-store/src/postgres/execution.rs @@ -728,7 +728,8 @@ impl ModelRequestRepository for PgExecutionStore { upstream_connection_idle_ms = $47 where id = $1 and outcome = 'running' returning id, client_api_key_ref, continuation_affinity_hash, - continuation_requested, outcome, started_at, completed_at + continuation_requested, provider_kind, upstream_transport, + outcome, started_at, completed_at ), recovery_target as ( select prior.id from model_requests prior @@ -775,9 +776,47 @@ impl ModelRequestRepository for PgExecutionStore { where prior.id = target.id and prior.recovered_at is null returning prior.id + ), session_transport_recovery_target as ( + select prior.id + from model_requests prior + cross join finalized current + where current.outcome = 'succeeded' + and current.provider_kind = 'openai' + and current.upstream_transport = 'http_sse' + and current.continuation_affinity_hash is not null + and prior.client_api_key_ref = current.client_api_key_ref + and prior.continuation_affinity_hash = current.continuation_affinity_hash + and prior.provider_kind = current.provider_kind + and prior.outcome = 'failed' + and prior.error_kind = 'upstream_unavailable' + and prior.upstream_transport = 'websocket' + and prior.upstream_send_state = 'ambiguous' + and prior.downstream_committed_at is null + and prior.recovered_at is null + and current.started_at >= prior.completed_at + and current.started_at <= prior.completed_at + interval '30 seconds' + ), session_transport_recovery_update as ( + update model_requests prior + set recovery_attempt_count = prior.recovery_attempt_count + 1, + recovery_request_id = current.id, + recovered_at = current.completed_at, + recovery_retry_delay_ms = greatest( + 0, + floor(extract(epoch from (current.started_at - prior.completed_at)) * 1000) + )::bigint, + recovery_total_latency_ms = greatest( + 0, + floor(extract(epoch from (current.completed_at - prior.completed_at)) * 1000) + )::bigint + from finalized current + join session_transport_recovery_target target on true + where prior.id = target.id + and prior.recovered_at is null + returning prior.id ) select (select count(*) from finalized)::bigint - + (select count(*) * 0 from recovery_update)::bigint", + + (select count(*) * 0 from recovery_update)::bigint + + (select count(*) * 0 from session_transport_recovery_update)::bigint", ) .bind(&finalization.model_request_id) .bind(finalization.outcome.as_str()) diff --git a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs index 844acbf9b..70d7246bd 100644 --- a/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs +++ b/backend/crates/gateway-store/src/postgres/observability/queries/ops.rs @@ -128,7 +128,6 @@ fn push_request_error_predicates( range: ObservabilityRange, filter: &OpsErrorFilter, ) { - statement.push(" and mr.recovered_at is null"); push_range(statement, "mr.completed_at", range); for (column, value) in [ ("mr.client_api_key_ref", &filter.client_api_key_ref), @@ -171,7 +170,6 @@ fn push_ops_event_predicates( range: ObservabilityRange, filter: &OpsErrorFilter, ) { - statement.push(" and (mr.id is null or mr.recovered_at is null)"); push_range(statement, "oe.created_at", range); for (column, value) in [ ("mr.client_api_key_ref", &filter.client_api_key_ref), diff --git a/backend/crates/gateway-store/tests/postgres/execution.rs b/backend/crates/gateway-store/tests/postgres/execution.rs index 8df813fa4..5a5f2854e 100644 --- a/backend/crates/gateway-store/tests/postgres/execution.rs +++ b/backend/crates/gateway-store/tests/postgres/execution.rs @@ -719,6 +719,165 @@ async fn successful_new_chain_should_mark_recent_continuation_failure_recovered( database.close().await; } +#[tokio::test] +async fn successful_http_downgrade_should_mark_pending_websocket_failures_recovered() { + let Some(database) = TestDatabase::create("execution_transport_downgrade_recovered").await + else { + return; + }; + let affinity_hash = "d".repeat(64); + let first_failure_at = Utc::now() - Duration::seconds(8); + let second_failure_at = first_failure_at + Duration::seconds(1); + seed_websocket_transport_failure( + &database.pool, + "req_websocket_failed_first", + &affinity_hash, + first_failure_at, + ) + .await; + seed_websocket_transport_failure( + &database.pool, + "req_websocket_failed_second", + &affinity_hash, + second_failure_at, + ) + .await; + seed_websocket_transport_failure( + &database.pool, + "req_websocket_failed_after_delivery", + &affinity_hash, + second_failure_at, + ) + .await; + sqlx::query( + "update model_requests + set downstream_committed_at = completed_at + where id = 'req_websocket_failed_after_delivery'", + ) + .execute(&database.pool) + .await + .expect("mark control failure delivered downstream"); + + let recovery_started_at = first_failure_at + Duration::seconds(4); + seed_transport_recovery_request( + &database.pool, + "req_http_fallback_succeeded", + &affinity_hash, + recovery_started_at, + ) + .await; + let store = PgExecutionStore::new(database.pool.clone()); + let mut recovery = successful_core_finalization("req_http_fallback_succeeded"); + recovery.upstream_transport = Some("http_sse".to_owned()); + recovery.downstream_committed_at = Some(std::time::SystemTime::now()); + ExecutionStore::finalize_model_request(&store, recovery) + .await + .expect("finalize successful HTTP fallback"); + + type PersistedRecovery = ( + String, + Option, + Option>, + i32, + Option, + Option, + ); + let recovered: Vec = sqlx::query_as( + "select id, recovery_request_id, recovered_at, recovery_attempt_count, + recovery_retry_delay_ms, recovery_total_latency_ms + from model_requests + where id like 'req_websocket_failed_%' + order by id", + ) + .fetch_all(&database.pool) + .await + .expect("load WebSocket recovery observations"); + + assert_eq!(recovered.len(), 3); + let delivered = &recovered[0]; + assert_eq!(delivered.0, "req_websocket_failed_after_delivery"); + assert_eq!((&delivered.1, delivered.2, delivered.3), (&None, None, 0)); + for (row, expected_delay_ms) in recovered[1..].iter().zip([4_000, 3_000]) { + assert_eq!(row.1.as_deref(), Some("req_http_fallback_succeeded")); + assert!(row.2.is_some()); + assert_eq!(row.3, 1); + assert_eq!(row.4, Some(expected_delay_ms)); + assert!(row.5.is_some_and(|latency| latency >= expected_delay_ms)); + } + database.close().await; +} + +#[tokio::test] +async fn websocket_success_or_late_http_success_should_not_hide_transport_failure() { + let Some(database) = TestDatabase::create("execution_transport_downgrade_unresolved").await + else { + return; + }; + let failure_at = Utc::now() - Duration::seconds(60); + let store = PgExecutionStore::new(database.pool.clone()); + + let websocket_affinity = "e".repeat(64); + seed_websocket_transport_failure( + &database.pool, + "req_websocket_control_failure", + &websocket_affinity, + failure_at, + ) + .await; + seed_transport_recovery_request( + &database.pool, + "req_websocket_control_success", + &websocket_affinity, + failure_at + Duration::seconds(4), + ) + .await; + ExecutionStore::finalize_model_request( + &store, + successful_core_finalization("req_websocket_control_success"), + ) + .await + .expect("finalize successful WebSocket control request"); + + let late_affinity = "f".repeat(64); + seed_websocket_transport_failure( + &database.pool, + "req_websocket_late_failure", + &late_affinity, + failure_at, + ) + .await; + seed_transport_recovery_request( + &database.pool, + "req_http_late_success", + &late_affinity, + failure_at + Duration::seconds(31), + ) + .await; + let mut late_recovery = successful_core_finalization("req_http_late_success"); + late_recovery.upstream_transport = Some("http_sse".to_owned()); + ExecutionStore::finalize_model_request(&store, late_recovery) + .await + .expect("finalize late HTTP request"); + + type UnresolvedTransportFailure = (String, Option, Option>, i32); + let unresolved: Vec = sqlx::query_as( + "select id, recovery_request_id, recovered_at, recovery_attempt_count + from model_requests + where id in ('req_websocket_control_failure', 'req_websocket_late_failure') + order by id", + ) + .fetch_all(&database.pool) + .await + .expect("load unresolved transport failures"); + assert_eq!(unresolved.len(), 2); + assert!( + unresolved + .iter() + .all(|row| row.1.is_none() && row.2.is_none() && row.3 == 0) + ); + database.close().await; +} + #[tokio::test] async fn failed_or_late_new_chain_should_leave_continuation_failure_unresolved() { let Some(database) = TestDatabase::create("execution_continuation_unresolved").await else { @@ -842,6 +1001,61 @@ async fn seed_recovery_request( .expect("prepare seeded recovery request"); } +async fn seed_websocket_transport_failure( + pool: &sqlx::PgPool, + id: &str, + affinity_hash: &str, + completed_at: chrono::DateTime, +) { + seed_running_request(pool, id) + .await + .expect("seed WebSocket transport failure request"); + sqlx::query( + "update model_requests + set client_api_key_ref = 'key_transport_recovery', + client_transport = 'http_sse', + continuation_affinity_hash = $2, + attempt_count = 1, upstream_send_state = 'ambiguous', + upstream_transport = 'websocket', outcome = 'failed', + error_kind = 'upstream_unavailable', + provider_error_code = 'websocket_close_1000', + started_at = $3 - interval '1 second', deadline_at = $3, + completed_at = $3 + where id = $1", + ) + .bind(id) + .bind(affinity_hash) + .bind(completed_at) + .execute(pool) + .await + .expect("finalize seeded WebSocket transport failure"); +} + +async fn seed_transport_recovery_request( + pool: &sqlx::PgPool, + id: &str, + affinity_hash: &str, + started_at: chrono::DateTime, +) { + seed_running_request(pool, id) + .await + .expect("seed transport recovery request"); + sqlx::query( + "update model_requests + set client_api_key_ref = 'key_transport_recovery', + client_transport = 'http_sse', + continuation_affinity_hash = $2, + started_at = $3, deadline_at = $3 + interval '1 minute' + where id = $1", + ) + .bind(id) + .bind(affinity_hash) + .bind(started_at) + .execute(pool) + .await + .expect("prepare seeded transport recovery request"); +} + async fn seed_running_request(pool: &sqlx::PgPool, id: &str) -> Result<(), sqlx::Error> { sqlx::query( "insert into model_requests ( diff --git a/backend/crates/gateway-store/tests/postgres/observability/mod.rs b/backend/crates/gateway-store/tests/postgres/observability/mod.rs index 14daeedce..4c203beb9 100644 --- a/backend/crates/gateway-store/tests/postgres/observability/mod.rs +++ b/backend/crates/gateway-store/tests/postgres/observability/mod.rs @@ -258,7 +258,8 @@ async fn ops_search_should_treat_sql_wildcards_as_literals() { } #[tokio::test] -async fn recovered_continuation_failure_should_be_hidden_from_default_business_metrics() { +async fn recovered_continuation_failure_should_be_visible_in_ops_but_hidden_from_business_metrics() +{ let Some(database) = TestDatabase::create("observability_recovered_continuation").await else { return; }; @@ -318,8 +319,13 @@ async fn recovered_continuation_failure_should_be_hidden_from_default_business_m page_size: ObservabilityPageSize::new(10).expect("page size"), }) .await - .expect("ops errors without recovered intermediates"); - assert_eq!(errors.total, 0); + .expect("ops errors retain recovered failures"); + assert_eq!(errors.total, 2); + assert!(errors.items.iter().all(|error| { + error.recovery_request_id.as_deref() == Some("req_observe_success") + && error.recovered_at.is_some() + && error.recovery_attempt_count == 1 + })); let retained: (String, Option) = sqlx::query_as( "select outcome, recovery_request_id diff --git a/backend/migrations/.frozen-sha256 b/backend/migrations/.frozen-sha256 index 015355dad..8cee48305 100644 --- a/backend/migrations/.frozen-sha256 +++ b/backend/migrations/.frozen-sha256 @@ -3,3 +3,4 @@ 1b6ffab434c754814aebb0eee3b650bd9770010059472bccf5bc856b319880ca 0003_client_min_versions.sql 22f912ab3e196de7983ef50a1dba5aee340bbea82bdd15d05863e87012000c83 0004_ops_error_send_state.sql 194d7b4bedd83f0d76428a21699b0e946449bf2ba61a0fdf8710ebfd421af8b2 0005_openai_continuation_recovery_observability.sql +3b4bfd5e3f5135fa2fad1088b0c2a287861e1250d66af31dbd40dc890751b567 0006_openai_session_transport_recovery.sql diff --git a/backend/migrations/0006_openai_session_transport_recovery.sql b/backend/migrations/0006_openai_session_transport_recovery.sql new file mode 100644 index 000000000..759372e77 --- /dev/null +++ b/backend/migrations/0006_openai_session_transport_recovery.sql @@ -0,0 +1,61 @@ +create index model_requests_pending_ws_transport_recovery_idx + on model_requests ( + client_api_key_ref, + continuation_affinity_hash, + completed_at desc, + id desc + ) + where provider_kind = 'openai' + and outcome = 'failed' + and error_kind = 'upstream_unavailable' + and upstream_transport = 'websocket' + and upstream_send_state = 'ambiguous' + and downstream_committed_at is null + and recovered_at is null + and continuation_affinity_hash is not null; + +with session_transport_recoveries as ( + select + prior.id as failed_request_id, + recovered.id as recovery_request_id, + recovered.started_at as recovery_started_at, + recovered.completed_at as recovered_at + from model_requests prior + cross join lateral ( + select current.id, current.started_at, current.completed_at + from model_requests current + where current.client_api_key_ref = prior.client_api_key_ref + and current.continuation_affinity_hash = prior.continuation_affinity_hash + and current.provider_kind = prior.provider_kind + and current.outcome = 'succeeded' + and current.upstream_transport = 'http_sse' + and current.completed_at is not null + and current.started_at >= prior.completed_at + and current.started_at <= prior.completed_at + interval '30 seconds' + order by current.started_at, current.id + limit 1 + ) recovered + where prior.provider_kind = 'openai' + and prior.outcome = 'failed' + and prior.error_kind = 'upstream_unavailable' + and prior.upstream_transport = 'websocket' + and prior.upstream_send_state = 'ambiguous' + and prior.downstream_committed_at is null + and prior.recovered_at is null + and prior.continuation_affinity_hash is not null +) +update model_requests prior +set recovery_attempt_count = prior.recovery_attempt_count + 1, + recovery_request_id = recovery.recovery_request_id, + recovered_at = recovery.recovered_at, + recovery_retry_delay_ms = greatest( + 0, + floor(extract(epoch from (recovery.recovery_started_at - prior.completed_at)) * 1000) + )::bigint, + recovery_total_latency_ms = greatest( + 0, + floor(extract(epoch from (recovery.recovered_at - prior.completed_at)) * 1000) + )::bigint +from session_transport_recoveries recovery +where prior.id = recovery.failed_request_id + and prior.recovered_at is null; diff --git a/frontend/src/views/usage/components/OpsErrorDetailModal.vue b/frontend/src/views/usage/components/OpsErrorDetailModal.vue index 1fd7688ef..e29f36201 100644 --- a/frontend/src/views/usage/components/OpsErrorDetailModal.vue +++ b/frontend/src/views/usage/components/OpsErrorDetailModal.vue @@ -31,6 +31,10 @@ const accountLabel = computed(() => props.record?.accountEmail || props.record?.accountId) const errorFields = computed(() => visibleFields([ + { + label: '处理状态', + value: props.record?.metadata.recoveredAt ? '已自动恢复' : null, + }, { label: 'Provider 错误码', value: props.record?.providerErrorCode, mono: true }, { label: '失败分类', value: failureClassText(props.record?.failureClass) }, { label: '分类代码', value: props.record?.failureClass, mono: true }, diff --git a/frontend/src/views/usage/components/OpsErrorPanel.vue b/frontend/src/views/usage/components/OpsErrorPanel.vue index 1be6c86c8..aa673a1fe 100644 --- a/frontend/src/views/usage/components/OpsErrorPanel.vue +++ b/frontend/src/views/usage/components/OpsErrorPanel.vue @@ -124,9 +124,17 @@ function upstreamSendStateText(value: string | null | undefined) {