diff --git a/backend/crates/gateway-admin/src/backup/task.rs b/backend/crates/gateway-admin/src/backup/task.rs index 967f9c78a..4cc502642 100644 --- a/backend/crates/gateway-admin/src/backup/task.rs +++ b/backend/crates/gateway-admin/src/backup/task.rs @@ -491,11 +491,13 @@ impl BackupTask { ) .await { - let _ = store.delete_object(&config, &record.object_key).await; - self.fail_task( + self.fail_uploaded_task( record, - error.code(), - format!("上传到对象存储失败:{}", error.message()), + &config, + BackupError::new( + error.code(), + format!("上传到对象存储失败:{}", error.message()), + ), ) .await?; return Ok(()); @@ -533,16 +535,19 @@ impl BackupTask { sha256_matches, "备份远端对象校验不一致" ); - let _ = store.delete_object(&config, &record.object_key).await; - self.fail_task( + self.fail_uploaded_task( record, - code::REMOTE_VERIFICATION_FAILED, - match (size_matches, sha256_matches) { - (false, true) => "远端对象大小不一致", - (true, false) => "远端对象 sha256 校验值不一致", - (false, false) => "远端对象大小和 sha256 校验值均不一致", - (true, true) => "远端对象校验状态异常", - }, + &config, + BackupError::new( + code::REMOTE_VERIFICATION_FAILED, + match (size_matches, sha256_matches) { + (false, true) => "远端对象大小不一致", + (true, false) => "远端对象 sha256 校验值不一致", + (false, false) => "远端对象大小和 sha256 校验值均不一致", + (true, true) => "远端对象校验状态异常", + } + .to_owned(), + ), ) .await?; } @@ -556,11 +561,13 @@ impl BackupTask { .await?; } Err(error) => { - let _ = store.delete_object(&config, &record.object_key).await; - self.fail_task( + self.fail_uploaded_task( record, - error.code(), - format!("远端对象校验失败:{}", error.message()), + &config, + BackupError::new( + error.code(), + format!("远端对象校验失败:{}", error.message()), + ), ) .await?; } @@ -568,6 +575,24 @@ impl BackupTask { Ok(()) } + /// 清理失败并入原终态,后续清理沿用备份记录的到期策略 + async fn fail_uploaded_task( + &self, + record: &BackupRecord, + config: &BackupStorageConfig, + mut error: BackupError, + ) -> Result<(), WorkerTaskError> { + if let Err(cleanup) = self + .object_store + .delete_object(config, &record.object_key) + .await + { + error = + error.with_cleanup_failure("DeleteObject", &cleanup, record.expires_at.is_some()); + } + self.fail_task(record, error.code(), error.message()).await + } + /// 把活跃任务写入失败终态并清理暂存 async fn fail_task( &self, diff --git a/backend/crates/gateway-admin/src/model/backup/mod.rs b/backend/crates/gateway-admin/src/model/backup/mod.rs index 3b3c27dcd..8eafce86f 100644 --- a/backend/crates/gateway-admin/src/model/backup/mod.rs +++ b/backend/crates/gateway-admin/src/model/backup/mod.rs @@ -474,6 +474,26 @@ impl BackupError { pub fn message(&self) -> &str { &self.message } + + /// 清理失败附加到原始终态,不覆盖上传失败或主动取消的分类 + #[must_use] + pub fn with_cleanup_failure( + mut self, + stage: &'static str, + error: &Self, + retry_on_expiration: bool, + ) -> Self { + let recovery = if retry_on_expiration { + "等待备份到期清理" + } else { + "未安排自动重试" + }; + self.message.push_str(&format!( + ";清理阶段 {stage} 失败({}),{recovery}", + error.code() + )); + self + } } /// 构造稳定错误码常量 diff --git a/backend/crates/gateway-admin/tests/backup/support.rs b/backend/crates/gateway-admin/tests/backup/support.rs index 9658b0870..eb4ea23b3 100644 --- a/backend/crates/gateway-admin/tests/backup/support.rs +++ b/backend/crates/gateway-admin/tests/backup/support.rs @@ -536,6 +536,9 @@ impl DatabaseDumpPort for FakeDumpPort { pub(crate) struct FakeObjectStore { pub(crate) objects: Mutex>>, upload_error: Option, + head_error: Option, + delete_error: Option, + connection_error: Option, } impl FakeObjectStore { @@ -543,6 +546,9 @@ impl FakeObjectStore { Self { objects: Mutex::new(std::collections::HashMap::new()), upload_error: None, + head_error: None, + delete_error: None, + connection_error: None, } } @@ -552,6 +558,21 @@ impl FakeObjectStore { self } + pub(crate) fn fail_head(mut self, code: &'static str) -> Self { + self.head_error = Some(BackupError::new(code, "HEAD failed".into())); + self + } + + pub(crate) fn fail_delete(mut self, code: &'static str) -> Self { + self.delete_error = Some(BackupError::new(code, "DELETE failed".into())); + self + } + + pub(crate) fn fail_connection(mut self, code: &'static str) -> Self { + self.connection_error = Some(BackupError::new(code, "DELETE failed".into())); + self + } + pub(crate) fn object(&self, key: &str) -> Option> { self.objects.lock().expect("objects").get(key).cloned() } @@ -563,6 +584,14 @@ impl BackupObjectStorePort for FakeObjectStore { &self, _config: &BackupStorageConfig, ) -> Result { + if let Some(error) = &self.connection_error { + return Ok(ConnectionTestResult { + ok: false, + stage: "deleteObject", + code: Some(error.code()), + message: error.message().into(), + }); + } Ok(ConnectionTestResult { ok: true, stage: "deleteObject", @@ -597,6 +626,9 @@ impl BackupObjectStorePort for FakeObjectStore { _config: &BackupStorageConfig, object_key: &str, ) -> Result, gateway_admin::model::backup::BackupError> { + if let Some(error) = &self.head_error { + return Err(error.clone()); + } let Some(content) = self.object(object_key) else { return Ok(None); }; @@ -613,6 +645,9 @@ impl BackupObjectStorePort for FakeObjectStore { _config: &BackupStorageConfig, object_key: &str, ) -> Result<(), gateway_admin::model::backup::BackupError> { + if let Some(error) = &self.delete_error { + return Err(error.clone()); + } self.objects.lock().expect("objects").remove(object_key); Ok(()) } diff --git a/backend/crates/gateway-admin/tests/backup/task.rs b/backend/crates/gateway-admin/tests/backup/task.rs index 3c46d61ca..39395601f 100644 --- a/backend/crates/gateway-admin/tests/backup/task.rs +++ b/backend/crates/gateway-admin/tests/backup/task.rs @@ -324,3 +324,58 @@ async fn completed_recovery_removes_staging_for_dumping_and_uploading_tasks() { ); } } + +#[tokio::test] +async fn manual_backup_preserves_primary_and_final_delete_failures() { + for (upload_fails, expiring) in [(false, false), (true, false), (false, true)] { + let repository = Arc::new(FakeBackupRepository::new(configured_settings())); + let dump = Arc::new(FakeDumpPort::new()); + let store = FakeObjectStore::new().fail_delete(code::S3_PERMISSION_DENIED); + let store = Arc::new(if upload_fails { + store.fail_upload(code::S3_UPLOAD_FAILED, "上传失败") + } else { + store.fail_head(code::REMOTE_VERIFICATION_FAILED) + }); + let seed = BackupRecordSeed { + id: backup_id("cleanup-failure"), + trigger_kind: BackupTriggerKind::Manual, + scheduled_at: None, + object_key: "codex/cleanup-failure.dump".into(), + expires_at: expiring.then(|| Utc::now() + Duration::hours(1)), + }; + repository.insert_backup_record(seed.clone()).await.unwrap(); + let task = BackupTask::new(repository.clone(), dump.clone(), store.clone()); + task.run_cycle(&CancellationToken::new()).await.unwrap(); + let records = repository.all_records(); + let record = &records[0]; + assert_eq!(record.status, BackupStatus::Failed); + assert_eq!( + record.error_code.as_deref(), + Some(if upload_fails { + code::S3_UPLOAD_FAILED + } else { + code::REMOTE_VERIFICATION_FAILED + }) + ); + let message = record.error_message.as_deref().unwrap(); + assert!(message.contains(if upload_fails { + "上传失败" + } else { + "HEAD failed" + })); + assert!(message.contains("DeleteObject")); + assert!(message.contains(code::S3_PERMISSION_DENIED)); + assert!(message.contains(if expiring { + "等待备份到期清理" + } else { + "未安排自动重试" + })); + assert_eq!(store.object(&seed.object_key).is_some(), !upload_fails); + assert!(dump.inspect_staging(&seed.id).await.unwrap().is_none()); + task.run_cycle(&CancellationToken::new()).await.unwrap(); + assert_eq!( + repository.all_records()[0].error_message, + record.error_message + ); + } +} diff --git a/backend/crates/gateway-admin/tests/use_case/backup.rs b/backend/crates/gateway-admin/tests/use_case/backup.rs index 58739b0e6..787fc6a22 100644 --- a/backend/crates/gateway-admin/tests/use_case/backup.rs +++ b/backend/crates/gateway-admin/tests/use_case/backup.rs @@ -102,3 +102,51 @@ async fn backup_service_rejects_download_of_non_completed_record() { .expect_err("non-completed record must be rejected"); assert_eq!(error.kind(), gateway_admin::model::AdminErrorKind::Conflict); } + +#[tokio::test] +async fn failed_delete_probe_does_not_refresh_verification() { + use gateway_admin::{model::backup::code, ports::backup::BackupRepository as _}; + for succeeds in [false, true] { + let mut settings = configured_settings(); + settings.last_verified_at = None; + let repository = Arc::new(FakeBackupRepository::new(settings)); + let auth = Arc::new(FakeAuthStore::new()); + let store = FakeObjectStore::new(); + let store = if succeeds { + store + } else { + store.fail_connection(code::S3_PERMISSION_DENIED) + }; + let services = AdminHarness::new() + .auth(auth.clone()) + .backup(BackupStorePorts::new( + repository.clone(), + Arc::new(FakeDumpPort::new()), + Arc::new(store), + )) + .build() + .await; + let result = services + .backups() + .test_connection(&system_context()) + .await + .unwrap(); + assert_eq!(result.ok, succeeds); + assert_eq!( + repository + .load_settings() + .await + .unwrap() + .last_verified_at + .is_some(), + succeeds + ); + assert_eq!( + auth.audit_actions() + .iter() + .filter(|action| action.as_str() == "backup.s3_connection_tested") + .count(), + 1 + ); + } +} diff --git a/backend/crates/gateway-api/src/middleware/mod.rs b/backend/crates/gateway-api/src/middleware/mod.rs index 8319a4e2c..aa99343b6 100644 --- a/backend/crates/gateway-api/src/middleware/mod.rs +++ b/backend/crates/gateway-api/src/middleware/mod.rs @@ -228,6 +228,8 @@ async fn handle( Ok(response.map(|body| body.map_err(|error| Box::new(error) as _).boxed_unsync())) }) }); + let request_id = context.request_id.clone(); + let call_id = context.call_id.clone(); let result = match plan { Some(plan) => plan.handle_http(context, request, next).await, None => next.run(request).await, @@ -251,7 +253,18 @@ async fn handle( tokio::select! { biased; () = cancellation.cancelled() => {}, - _ = task => {}, + result = task => { + if let Err(contract::upgrade::SessionError::Failed { kind, plugin_instance_id }) = result { + tracing::warn!( + event = "middleware.session.failed", + request_id, + call_id, + plugin_instance_id, + kind, + "升级后的中间件会话失败" + ); + } + }, } }); } diff --git a/backend/crates/gateway-api/tests/middleware/mod.rs b/backend/crates/gateway-api/tests/middleware/mod.rs index 86817603f..554538926 100644 --- a/backend/crates/gateway-api/tests/middleware/mod.rs +++ b/backend/crates/gateway-api/tests/middleware/mod.rs @@ -1,5 +1,6 @@ //! HTTP 与 WebSocket 中间件测试入口及请求生命周期替身 +mod session; mod websocket; use std::{ @@ -40,6 +41,9 @@ enum Mode { idle: bool, block_messages: bool, }, + SessionEnd { + kind: Option<&'static str>, + }, Service, Echo, Slow { @@ -161,13 +165,43 @@ impl MiddlewarePlan for Plan { session.send(Message { kind: Kind::Text, payload: Bytes::from_static(b"ready") }).await?; let message = receive.await?.unwrap(); session.send(message).await?; - session.send(Message { kind: Kind::Close { code: Some(1000) }, payload: Bytes::from_static(b"done") }).await + session.send(Message { kind: Kind::Close { code: Some(1000) }, payload: Bytes::from_static(b"done") }).await.map_err(Into::into) }))); if reject { *response.status_mut() = StatusCode::FORBIDDEN; } Ok(response) } + Mode::SessionEnd { kind } => { + let upgrade = request + .extensions_mut() + .remove::>() + .unwrap(); + let (parts, _) = request.into_parts(); + let (mut response, session) = upgrade.accept(parts, Vec::new()).await?; + response + .extensions_mut() + .insert(contract::upgrade::Upgraded::new( + StatusCode::SWITCHING_PROTOCOLS, + Box::pin(async move { + let _session = session.await.map_err(|_| MiddlewareError::Fault)?; + match kind { + None => Ok(()), + Some("cancelled") => { + Err(contract::upgrade::SessionError::Cancelled) + } + Some("reported") => { + Err(contract::upgrade::SessionError::Reported) + } + Some(kind) => Err(contract::upgrade::SessionError::Failed { + kind, + plugin_instance_id: Some("plugin-fixture".into()), + }), + } + }), + )); + Ok(response) + } Mode::Rewrite => { assert_eq!(request.uri(), "/unrecognized/client/path?raw=%2f%FF"); assert_eq!(request.headers()["authorization"], "Bearer fixture-secret"); diff --git a/backend/crates/gateway-api/tests/middleware/session.rs b/backend/crates/gateway-api/tests/middleware/session.rs new file mode 100644 index 000000000..5338c6f83 --- /dev/null +++ b/backend/crates/gateway-api/tests/middleware/session.rs @@ -0,0 +1,79 @@ +//! 验证升级后终态只记录一次安全分类,正常结束及已报告失败不重复输出 + +use super::*; +use tracing::{ + Event, Metadata, Subscriber, + span::{Attributes, Id, Record}, +}; + +#[derive(Clone, Default)] +struct Recorder(Arc>>); + +impl Subscriber for Recorder { + fn enabled(&self, _: &Metadata<'_>) -> bool { + true + } + fn new_span(&self, _: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + fn record(&self, _: &Id, _: &Record<'_>) {} + fn record_follows_from(&self, _: &Id, _: &Id) {} + fn enter(&self, _: &Id) {} + fn exit(&self, _: &Id) {} + fn event(&self, event: &Event<'_>) { + struct Fields(String); + impl tracing::field::Visit for Fields { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + use std::fmt::Write as _; + write!(&mut self.0, "{}={value:?} ", field.name()).unwrap(); + } + } + let mut fields = Fields(String::new()); + event.record(&mut fields); + if fields.0.contains("middleware.session.failed") { + self.0.lock().unwrap().push(fields.0); + } + } +} + +#[tokio::test] +async fn upgraded_session_records_one_correlated_failure_and_ignores_expected_endings() { + use tokio_tungstenite::{connect_async, tungstenite::client::IntoClientRequest as _}; + for kind in [ + None, + Some("cancelled"), + Some("reported"), + Some("permission_denied"), + ] { + let recorder = Recorder::default(); + let _subscriber = tracing::subscriber::set_default(recorder.clone()); + let (router, plan) = router(Mode::SessionEnd { kind }, None).await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + }); + let mut request = format!("ws://{address}/plugin-owned/session") + .into_client_request() + .unwrap(); + request + .headers_mut() + .insert("x-request-id", "session-regression".parse().unwrap()); + let (_socket, response) = connect_async(request).await.unwrap(); + assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS); + let cancellation = plan.cancellation.lock().unwrap().clone().unwrap(); + tokio::time::timeout(Duration::from_secs(3), cancellation.cancelled()) + .await + .unwrap(); + let events = recorder.0.lock().unwrap(); + if kind == Some("permission_denied") { + assert_eq!(events.len(), 1, "{events:?}"); + assert!(events[0].contains("session-regression")); + assert!(events[0].contains("plugin-fixture")); + assert!(events[0].contains("permission_denied")); + } else { + assert!(events.is_empty(), "{events:?}"); + } + server.abort(); + } +} diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index a9047faae..4a0e05c72 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -2057,7 +2057,13 @@ where let (outcome, gateway_error) = match error { EngineError::Cancelled => ( ExecutionOutcome::Cancelled, - GatewayError::new(GatewayErrorKind::Cancelled, "request was cancelled"), + GatewayError::new( + GatewayErrorKind::Cancelled, + self.cancellation + .reason() + .unwrap_or(crate::lifecycle::CancellationReason::Requested) + .message(), + ), ), EngineError::Deadline => ( if self.response_commit.committed_at().is_some() { diff --git a/backend/crates/gateway-core/src/engine/middleware/http/upgrade.rs b/backend/crates/gateway-core/src/engine/middleware/http/upgrade.rs index 25b062181..27fdc9784 100644 --- a/backend/crates/gateway-core/src/engine/middleware/http/upgrade.rs +++ b/backend/crates/gateway-core/src/engine/middleware/http/upgrade.rs @@ -8,7 +8,42 @@ use crate::{engine::middleware::MiddlewareError, engine::middleware::websocket:: pub type PendingSession = Shared, Arc>>>; -pub type SessionTask = BoxFuture<'static, Result<(), MiddlewareError>>; +pub type SessionTask = BoxFuture<'static, Result<(), SessionError>>; + +/// 升级后的终态由传输 owner 消费,协议适配器只提供安全分类与诊断归属 +#[derive(Debug)] +pub enum SessionError { + /// 预期取消,不作为异常终态输出 + Cancelled, + /// 下层已记录同一失败,传输 owner 不重复输出 + Reported, + Failed { + kind: &'static str, + plugin_instance_id: Option, + }, +} + +impl From for SessionError { + fn from(error: MiddlewareError) -> Self { + use crate::error::GatewayErrorKind; + if matches!(&error, MiddlewareError::Gateway(error) if error.kind() == GatewayErrorKind::Cancelled) + || matches!( + error, + MiddlewareError::Engine(crate::engine::EngineError::Cancelled) + ) + { + return Self::Cancelled; + } + Self::Failed { + kind: if error.is_rejected() { + "rejected" + } else { + "middleware" + }, + plugin_instance_id: None, + } + } +} /// 具体服务器负责握手与 socket;Core 不依赖 HTTP 服务器实现 pub trait WebSocketUpgrade: Send + Sync { diff --git a/backend/crates/gateway-core/src/lifecycle.rs b/backend/crates/gateway-core/src/lifecycle.rs index c0d2c5e5f..057e8e6b1 100644 --- a/backend/crates/gateway-core/src/lifecycle.rs +++ b/backend/crates/gateway-core/src/lifecycle.rs @@ -1,16 +1,69 @@ //! 请求与连接的截止、租约、取消和 drain 契约 use std::fmt; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, OnceLock}; use std::time::{Duration, SystemTime}; use event_listener::Event; use futures::future::{BoxFuture, pending}; use futures_timer::Delay; +/// 参与业务并发约束的请求租约类别,观测续期不属于此合同 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RequestLeaseKind { + ClientAdmission, + Credential, +} + +/// 续期最终结束的原因,不包含可恢复的中间失败 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LeaseFailure { + Lost, + RenewalTimedOut, + Expired, +} + +/// 只携带稳定原因,不保存凭据、Redis key 或任意基础设施错误文本 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum CancellationReason { + Requested, + Lease { + kind: RequestLeaseKind, + failure: LeaseFailure, + }, +} + +impl CancellationReason { + #[must_use] + pub const fn message(self) -> &'static str { + match self { + Self::Requested => "request was cancelled", + Self::Lease { + kind: RequestLeaseKind::ClientAdmission, + failure, + } => match failure { + LeaseFailure::Lost => "request was cancelled: client admission lease was lost", + LeaseFailure::RenewalTimedOut => { + "request was cancelled: client admission lease renewal timed out" + } + LeaseFailure::Expired => "request was cancelled: client admission lease expired", + }, + Self::Lease { + kind: RequestLeaseKind::Credential, + failure, + } => match failure { + LeaseFailure::Lost => "request was cancelled: credential lease was lost", + LeaseFailure::RenewalTimedOut => { + "request was cancelled: credential lease renewal timed out" + } + LeaseFailure::Expired => "request was cancelled: credential lease expired", + }, + } + } +} + struct CancellationState { - cancelled: AtomicBool, + reason: OnceLock, event: Event, } @@ -41,7 +94,7 @@ impl CancellationToken { pub fn new() -> Self { Self { state: Arc::new(CancellationState { - cancelled: AtomicBool::new(false), + reason: OnceLock::new(), event: Event::new(), }), ancestors: Arc::from([]), @@ -56,7 +109,7 @@ impl CancellationToken { .collect::>(); Self { state: Arc::new(CancellationState { - cancelled: AtomicBool::new(false), + reason: OnceLock::new(), event: Event::new(), }), ancestors: ancestors.into(), @@ -64,19 +117,29 @@ impl CancellationToken { } pub fn cancel(&self) { - if self.state.cancelled.swap(true, Ordering::AcqRel) { + self.cancel_with_reason(CancellationReason::Requested); + } + + /// 首次取消确定原因,后续资源回收不覆盖已观察到的取消事实 + pub fn cancel_with_reason(&self, reason: CancellationReason) { + if self.is_cancelled() || self.state.reason.set(reason).is_err() { return; } self.state.event.notify(usize::MAX); } #[must_use] - pub fn is_cancelled(&self) -> bool { - self.state.cancelled.load(Ordering::Acquire) - || self - .ancestors + pub fn reason(&self) -> Option { + self.state.reason.get().copied().or_else(|| { + self.ancestors .iter() - .any(|state| state.cancelled.load(Ordering::Acquire)) + .find_map(|state| state.reason.get().copied()) + }) + } + + #[must_use] + pub fn is_cancelled(&self) -> bool { + self.reason().is_some() } pub async fn cancelled(&self) { @@ -92,13 +155,13 @@ impl CancellationToken { fn wait_for_state(state: Arc) -> BoxFuture<'static, ()> { Box::pin(async move { - if state.cancelled.load(Ordering::Acquire) { + if state.reason.get().is_some() { return; } // listener 随等待 future 丢弃而注销,避免 select 败选分支在自身及祖先状态中积累 let listener = state.event.listen(); // 注册后复查,覆盖取消发生在首次检查与注册之间的竞态 - if state.cancelled.load(Ordering::Acquire) { + if state.reason.get().is_some() { return; } listener.await; diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index cfad358fd..6f745dcbe 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -53,6 +53,7 @@ use serde_json::{Map, Value, json}; #[derive(Debug, Clone, PartialEq)] struct FinalState { + error_message: Option, error_kind: Option, diagnostic_trace_json: Option, outcome: ExecutionOutcome, @@ -228,6 +229,10 @@ impl ExecutionStore for FakeStore { .expect("store lock") .finalizations .push(FinalState { + error_message: finalization + .error + .as_ref() + .map(|error| error.safe_message().to_owned()), error_kind: finalization.error.as_ref().map(|error| error.kind()), diagnostic_trace_json: finalization.diagnostic_trace_json, outcome: finalization.outcome, @@ -5995,3 +6000,40 @@ fn transport_keepalive_cannot_change_an_already_finalized_failure_to_http_200() assert!(!state.finalizations[0].committed); assert_eq!(state.recorded_statuses, vec![504]); } + +#[test] +fn lease_cancellation_is_preserved_in_the_single_request_terminal() { + use gateway_core::lifecycle::{CancellationReason, LeaseFailure, RequestLeaseKind}; + block_on(async { + let operation = generate_operation(); + let (coordinator, store, _) = coordinator(vec![Script::Pending]); + let cancellation = CancellationToken::new(); + let mut session = coordinator + .start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation.clone(), + plan(&operation), + None, + None, + cancellation.clone(), + ) + .await + .unwrap(); + let reason = CancellationReason::Lease { + kind: RequestLeaseKind::ClientAdmission, + failure: LeaseFailure::RenewalTimedOut, + }; + let mut next = Box::pin(session.next_event()); + assert!(futures::poll!(next.as_mut()).is_pending()); + cancellation.cancel_with_reason(reason); + assert!(matches!(next.await, Err(EngineError::Cancelled))); + session.cancel_and_finalize().await.unwrap(); + let state = store.state.lock().unwrap(); + assert_eq!(state.finalizations.len(), 1); + assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Cancelled); + assert_eq!( + state.finalizations[0].error_message.as_deref(), + Some(reason.message()) + ); + }); +} diff --git a/backend/crates/gateway-core/tests/lifecycle.rs b/backend/crates/gateway-core/tests/lifecycle.rs index 2fbd37f06..cec7191e6 100644 --- a/backend/crates/gateway-core/tests/lifecycle.rs +++ b/backend/crates/gateway-core/tests/lifecycle.rs @@ -322,3 +322,25 @@ fn connection_lifecycle_contract_is_object_safe() { assert!(!object.is_draining()); } + +#[test] +fn cancellation_reason_survives_cleanup_and_propagates_only_to_descendants() { + use gateway_core::lifecycle::{CancellationReason, LeaseFailure, RequestLeaseKind}; + let parent = CancellationToken::new(); + let child = parent.child_token(); + let descendant = child.child_token(); + let reason = CancellationReason::Lease { + kind: RequestLeaseKind::Credential, + failure: LeaseFailure::Lost, + }; + child.cancel_with_reason(reason); + child.cancel(); + descendant.cancel(); + assert_eq!(parent.reason(), None); + assert_eq!(child.reason(), Some(reason)); + assert_eq!(descendant.reason(), Some(reason)); + let requested = CancellationToken::new(); + requested.cancel(); + requested.cancel_with_reason(reason); + assert_eq!(requested.reason(), Some(CancellationReason::Requested)); +} diff --git a/backend/crates/gateway-host/src/system_update/mod.rs b/backend/crates/gateway-host/src/system_update/mod.rs index 38b9e17e4..81f3968c1 100644 --- a/backend/crates/gateway-host/src/system_update/mod.rs +++ b/backend/crates/gateway-host/src/system_update/mod.rs @@ -11,6 +11,7 @@ mod process; mod release; mod state; mod swap; +mod swap_paths; pub use config::SystemUpdateConfig; pub use operations::ProcessSystemOperations; diff --git a/backend/crates/gateway-host/src/system_update/swap.rs b/backend/crates/gateway-host/src/system_update/swap.rs index 4c5893ac3..f1ad344aa 100644 --- a/backend/crates/gateway-host/src/system_update/swap.rs +++ b/backend/crates/gateway-host/src/system_update/swap.rs @@ -245,18 +245,7 @@ fn swap_file(current: &Path, backup: &Path) -> io::Result<()> { if swap.exists() { fs::remove_file(&swap)?; } - move_file(current, &swap)?; - if let Err(error) = move_file(backup, current) { - let _ = move_file(&swap, current); - return Err(error); - } - if let Err(error) = move_file(&swap, backup) { - let _ = move_file(current, &swap); - let _ = move_file(backup, current); - let _ = move_file(&swap, backup); - return Err(error); - } - Ok(()) + super::swap_paths::swap_paths(current, backup, &swap, move_file) } fn swap_dir(current: &Path, backup: &Path) -> io::Result<()> { @@ -270,18 +259,7 @@ fn swap_dir(current: &Path, backup: &Path) -> io::Result<()> { if swap.exists() { remove_dir_all(&swap)?; } - move_dir(current, &swap)?; - if let Err(error) = move_dir(backup, current) { - let _ = move_dir(&swap, current); - return Err(error); - } - if let Err(error) = move_dir(&swap, backup) { - let _ = move_dir(current, &swap); - let _ = move_dir(backup, current); - let _ = move_dir(&swap, backup); - return Err(error); - } - Ok(()) + super::swap_paths::swap_paths(current, backup, &swap, move_dir) } fn restore_dir(current: &Path, backup: &Path) -> io::Result<()> { diff --git a/backend/crates/gateway-host/src/system_update/swap_paths.rs b/backend/crates/gateway-host/src/system_update/swap_paths.rs new file mode 100644 index 000000000..9ff8fb7ae --- /dev/null +++ b/backend/crates/gateway-host/src/system_update/swap_paths.rs @@ -0,0 +1,59 @@ +//! 文件和目录交换共用的恢复顺序,保留主失败及补偿失败 + +use std::{io, path::Path}; + +pub(super) fn swap_paths( + current: &Path, + backup: &Path, + swap: &Path, + mut move_path: impl FnMut(&Path, &Path) -> io::Result<()>, +) -> io::Result<()> { + move_path(current, swap)?; + if let Err(error) = move_path(backup, current) { + return Err(with_recovery( + error, + "restore current from swap", + move_path(swap, current), + [current, backup, swap], + )); + } + if let Err(error) = move_path(swap, backup) { + // 先把备份版本放回 backup,才能把原版本移回 current + // 补偿第一步失败时保留 swap,不能覆盖仍在 current 的备份版本 + if let Err(recovery) = move_path(current, backup) { + return Err(with_recovery( + error, + "restore backup from current", + Err(recovery), + [current, backup, swap], + )); + } + return Err(with_recovery( + error, + "restore current from swap", + move_path(swap, current), + [current, backup, swap], + )); + } + Ok(()) +} + +fn with_recovery( + primary: io::Error, + action: &'static str, + recovery: io::Result<()>, + [current, backup, swap]: [&Path; 3], +) -> io::Error { + match recovery { + Ok(()) => primary, + Err(recovery) => io::Error::new( + primary.kind(), + format!( + "{primary}; rollback failed: {action}: {recovery}; manual recovery required (current={}, backup={}, swap={})", + current.display(), + backup.display(), + swap.display() + ), + ), + } +} diff --git a/backend/crates/gateway-host/tests/system_update/mod.rs b/backend/crates/gateway-host/tests/system_update/mod.rs index 197117bb1..0e32b062f 100644 --- a/backend/crates/gateway-host/tests/system_update/mod.rs +++ b/backend/crates/gateway-host/tests/system_update/mod.rs @@ -7,6 +7,7 @@ mod installation; mod operations; mod release; mod state; +mod swap_paths; use std::fs; use std::path::PathBuf; diff --git a/backend/crates/gateway-host/tests/system_update/swap_paths.rs b/backend/crates/gateway-host/tests/system_update/swap_paths.rs new file mode 100644 index 000000000..af8b81e3c --- /dev/null +++ b/backend/crates/gateway-host/tests/system_update/swap_paths.rs @@ -0,0 +1,76 @@ +//! 对真实暂存文件注入移动双故障,验证恢复次序和人工恢复线索 + +#[path = "../../src/system_update/swap_paths.rs"] +mod implementation; + +use std::{fs, io}; + +#[test] +fn failed_swap_preserves_both_versions_and_reports_compensation_failures() { + for directory in [false, true] { + for primary in [2, 3] { + let mut recoveries = vec![None, Some(primary + 1)]; + if primary == 3 { + recoveries.push(Some(5)); + } + for recovery in recoveries { + let temp = tempfile::tempdir().unwrap(); + let current = temp.path().join("current"); + let backup = temp.path().join("backup"); + let swap = temp.path().join("swap"); + for (path, content) in [(¤t, "current version"), (&backup, "backup version")] + { + if directory { + fs::create_dir(path).unwrap(); + fs::write(path.join("asset"), content).unwrap(); + } else { + fs::write(path, content).unwrap(); + } + } + let mut moves = 0; + let error = implementation::swap_paths(¤t, &backup, &swap, |from, to| { + moves += 1; + if moves == primary { + return Err(io::Error::other("primary injected failure")); + } + if recovery == Some(moves) { + return Err(io::Error::other("recovery injected failure")); + } + fs::rename(from, to) + }) + .unwrap_err(); + let message = error.to_string(); + assert!(message.contains("primary injected failure")); + if recovery.is_some() { + assert!(message.contains("recovery injected failure")); + assert!(message.contains("restore ")); + assert!(message.contains("manual recovery required")); + assert!(message.contains(swap.to_str().unwrap())); + } else { + assert!(!message.contains("rollback failed")); + } + let mut contents = [¤t, &backup, &swap] + .into_iter() + .filter_map(|path| { + fs::read_to_string(if directory { + path.join("asset") + } else { + path.clone() + }) + .ok() + }) + .collect::>(); + contents.sort(); + assert_eq!(contents, ["backup version", "current version"]); + if recovery.is_none() { + let original = if directory { + current.join("asset") + } else { + current.clone() + }; + assert_eq!(fs::read_to_string(original).unwrap(), "current version"); + } + } + } + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/http.rs b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/http.rs index 85ba7426d..951137e43 100644 --- a/backend/crates/gateway-plugin/runtime/src/adapter/middleware/http.rs +++ b/backend/crates/gateway-plugin/runtime/src/adapter/middleware/http.rs @@ -87,16 +87,16 @@ pub(super) async fn invoke( let _binding = _binding; tokio::select! { biased; - () = context.cancellation.cancelled() => Err(MiddlewareError::Fault), + () = context.cancellation.cancelled() => Err(core::upgrade::SessionError::Cancelled), result = async { invocation.session_ready().await?; match stream.next().await { Ok(None) => Ok(()), Ok(Some(_)) => { - session.invalid_response(Stage::Http); - Err(MiddlewareError::InvalidState) + let error = session.invalid_response(Stage::Http); + Err(session_error(error, &entry.instance_id)) } - Err(error) => Err(crate::callback::error::rpc_middleware(error)), + Err(error) => Err(session_error(error, &entry.instance_id)), } } => result, } @@ -143,3 +143,32 @@ pub(super) async fn invoke( .await .map_err(|_| invalid()) } + +fn session_error(error: crate::RpcError, instance_id: &str) -> core::upgrade::SessionError { + use core::upgrade::SessionError; + use gateway_plugin_sdk::ErrorCode; + + let kind = match &error { + crate::RpcError::Cancelled => return SessionError::Cancelled, + // 本地期限已经按调用关联记录,终态消费不能再记同一次超时 + crate::RpcError::Timeout => return SessionError::Reported, + crate::RpcError::Remote(fault) => match fault.code { + ErrorCode::Cancelled => return SessionError::Cancelled, + ErrorCode::Unsupported => "unsupported", + ErrorCode::Rejected => "rejected", + ErrorCode::InvalidInput => "invalid_input", + ErrorCode::PermissionDenied => "permission_denied", + ErrorCode::Upstream => "upstream", + ErrorCode::Fault => "fault", + ErrorCode::Timeout => "timeout", + ErrorCode::Uncertain => "uncertain", + ErrorCode::Capacity => "capacity", + ErrorCode::Conflict => "conflict", + }, + _ => error.diagnostic().0, + }; + SessionError::Failed { + kind, + plugin_instance_id: Some(instance_id.to_owned()), + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/websocket/mod.rs b/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/websocket/mod.rs index b7fa01445..080b084ed 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/websocket/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/policy/http/websocket/mod.rs @@ -347,3 +347,65 @@ async fn dropping_an_uncommitted_upgrade_or_cancelling_a_session_reclaims_its_so runtime.shutdown().await; } } + +#[tokio::test] +async fn upgraded_plugin_business_failure_reaches_terminal_owner_with_safe_classification() { + for mode in ["session-error", "session-cancelled"] { + let (_cache, runtime, generation, plan) = http_plugin(mode).await; + let (_send, incoming) = tokio::sync::mpsc::channel(1); + let (outgoing, _receive) = tokio::sync::mpsc::channel(1); + let transport = Arc::new(Transport { + incoming: tokio::sync::Mutex::new(incoming), + outgoing, + dropped: Arc::default(), + }); + let mut request = ::http::Request::builder() + .uri("/custom/session") + .body(core::empty_body()) + .unwrap(); + request + .extensions_mut() + .insert(Arc::new(Upgrade(transport)) as Arc); + let mut response = plan + .handle_http( + core::Context { + plugin_instance_id: None, + request_id: "session-failure".into(), + call_id: "http-call".into(), + parent_call_id: None, + extensions: Default::default(), + plan: Some(plan.clone()), + cancellation: CancellationToken::new(), + }, + request, + compose(Vec::new(), |_| { + Box::pin(async { panic!("session owns its response") }) + }), + ) + .await + .unwrap(); + assert_eq!(response.status(), 101); + let task = response + .extensions_mut() + .remove::() + .unwrap() + .take_task() + .unwrap(); + let error = tokio::time::timeout(Duration::from_secs(5), task) + .await + .unwrap() + .unwrap_err(); + if mode == "session-error" { + assert!( + matches!(error, upgrade::SessionError::Failed { kind: "permission_denied", plugin_instance_id: Some(ref id) } if id == "http-plugin") + ); + } else { + assert!(matches!(error, upgrade::SessionError::Cancelled)); + } + assert!(!format!("{error:?}").contains("private-plugin-error")); + drop(response); + drop(plan); + drop(generation); + runtime.shutdown().await; + } +} diff --git a/backend/crates/gateway-plugin/runtime/tests/support/middleware/mod.rs b/backend/crates/gateway-plugin/runtime/tests/support/middleware/mod.rs index 4d0f76670..8871d8d29 100644 --- a/backend/crates/gateway-plugin/runtime/tests/support/middleware/mod.rs +++ b/backend/crates/gateway-plugin/runtime/tests/support/middleware/mod.rs @@ -331,6 +331,22 @@ async fn main() -> Result<(), Box> { } return call.host.dispatch_http(call.request).await; } + if mode == "session-error" || mode == "session-cancelled" { + use gateway_plugin_sdk::{ErrorCode, PluginFault}; + let code = if mode == "session-error" { + ErrorCode::PermissionDenied + } else { + ErrorCode::Cancelled + }; + return call + .upgrade(vec!["test-session".into()], move |_session| async move { + Err(PluginFault::new( + code, + "private-plugin-error-must-not-be-logged", + )) + }) + .await; + } if mode == "callbacks" { let host = call.host.clone(); return call.upgrade(vec!["test-session".into()], |mut session| async move { diff --git a/backend/crates/gateway-store/Cargo.toml b/backend/crates/gateway-store/Cargo.toml index 83be6b10b..82405722c 100644 --- a/backend/crates/gateway-store/Cargo.toml +++ b/backend/crates/gateway-store/Cargo.toml @@ -38,7 +38,7 @@ uuid = { version = "1.23.3", features = ["v4", "v7"] } [dev-dependencies] tempfile = "3.23" -tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt-multi-thread", "fs", "io-util", "net", "process", "time"] } +tokio = { version = "1.52.3", default-features = false, features = ["macros", "rt-multi-thread", "fs", "io-util", "net", "process", "time", "test-util"] } [lints] workspace = true diff --git a/backend/crates/gateway-store/src/backup/s3.rs b/backend/crates/gateway-store/src/backup/s3.rs index 2e78fa03a..f15686b65 100644 --- a/backend/crates/gateway-store/src/backup/s3.rs +++ b/backend/crates/gateway-store/src/backup/s3.rs @@ -192,7 +192,15 @@ impl BackupObjectStorePort for S3ObjectStoreAdapter { } } - let _ = best_effort_delete(&client, config, &probe_key).await; + if let Err(error) = client + .delete_object() + .bucket(&config.bucket) + .key(&probe_key) + .send() + .await + { + return Ok(failed(ConnectionTestStage::DeleteObject, &error)); + } Ok(ConnectionTestResult { ok: true, stage: ConnectionTestStage::DeleteObject.as_str(), @@ -246,24 +254,24 @@ impl BackupObjectStorePort for S3ObjectStoreAdapter { Err(BackupError::new(code::CANCELLED, "上传已取消".to_owned())) } }; - if let Err(error) = &outcome { - let _ = client + if let Err(error) = outcome { + let cleanup = client .abort_multipart_upload() .bucket(&config.bucket) .key(&request.object_key) .upload_id(&upload_id) .send() .await; - if error.code() != code::CANCELLED { - let _ = client - .delete_object() - .bucket(&config.bucket) - .key(&request.object_key) - .send() - .await; - } + return Err(match cleanup { + Ok(_) => error, + Err(cleanup) => error.with_cleanup_failure( + "AbortMultipartUpload", + &map_s3_error(cleanup), + false, + ), + }); } - outcome + Ok(()) } async fn head_object( diff --git a/backend/crates/gateway-store/src/lease_renewal.rs b/backend/crates/gateway-store/src/lease_renewal.rs index a184574ce..5289050d2 100644 --- a/backend/crates/gateway-store/src/lease_renewal.rs +++ b/backend/crates/gateway-store/src/lease_renewal.rs @@ -3,7 +3,10 @@ use std::time::Duration; use futures::future::BoxFuture; -use gateway_core::lifecycle::{CancellationToken, Deadline, REQUEST_LEASE_TTL}; +use gateway_core::lifecycle::{ + CancellationReason, CancellationToken, Deadline, LeaseFailure, REQUEST_LEASE_TTL, + RequestLeaseKind, +}; use crate::StoreResult; @@ -12,22 +15,22 @@ pub(crate) struct LeaseRenewal(tokio::task::JoinHandle<()>); impl LeaseRenewal { pub(crate) fn spawn( deadline: Deadline, - cancellation: Option, + cancellation: Option<(CancellationToken, RequestLeaseKind)>, mut renew: impl FnMut(Duration) -> BoxFuture<'static, StoreResult> + Send + 'static, ) -> Self { Self(tokio::spawn(async move { let cancelled = async { match &cancellation { - Some(cancellation) => cancellation.cancelled().await, + Some((cancellation, _)) => cancellation.cancelled().await, None => std::future::pending().await, } }; tokio::pin!(cancelled); let mut expires = tokio::time::Instant::now() + deadline.bounded(REQUEST_LEASE_TTL); - loop { + let failure = loop { let ttl = deadline.bounded(REQUEST_LEASE_TTL); if ttl.is_zero() { - break; + break LeaseFailure::Expired; } let renewed_until = tokio::time::Instant::now() + ttl; let result = tokio::select! { @@ -40,7 +43,8 @@ impl LeaseRenewal { expires = renewed_until; REQUEST_LEASE_TTL / 3 } - Ok(Ok(false)) | Err(_) => break, + Ok(Ok(false)) => break LeaseFailure::Lost, + Err(_) => break LeaseFailure::RenewalTimedOut, Ok(Err(error)) => { tracing::warn!(%error, "请求租约续期失败,将在租约有效期内重试"); Duration::from_secs(1) @@ -49,15 +53,18 @@ impl LeaseRenewal { tokio::select! { biased; () = &mut cancelled => return, - () = tokio::time::sleep_until(expires) => break, + () = tokio::time::sleep_until(expires) => break LeaseFailure::Expired, () = tokio::time::sleep(delay) => {} } - } + }; // 显式执行截止由 Core 归类为 timeout,租约不把它抢先改成 cancelled if !deadline.is_elapsed() - && let Some(cancellation) = &cancellation + && let Some((cancellation, kind)) = &cancellation { - cancellation.cancel(); + cancellation.cancel_with_reason(CancellationReason::Lease { + kind: *kind, + failure, + }); } })) } diff --git a/backend/crates/gateway-store/src/redis/client_admission.rs b/backend/crates/gateway-store/src/redis/client_admission.rs index 889b62bd5..1910d9b36 100644 --- a/backend/crates/gateway-store/src/redis/client_admission.rs +++ b/backend/crates/gateway-store/src/redis/client_admission.rs @@ -399,7 +399,10 @@ impl ClientAdmissionPort for RedisClientAdmissionRepository { let request = request.as_str().to_owned(); Box::new(crate::lease_renewal::LeaseRenewal::spawn( deadline, - Some(cancellation), + Some(( + cancellation, + gateway_core::lifecycle::RequestLeaseKind::ClientAdmission, + )), move |ttl| { let repository = repository.clone(); let key = key.clone(); diff --git a/backend/crates/gateway-store/src/redis/credential_leases.rs b/backend/crates/gateway-store/src/redis/credential_leases.rs index e674fcad4..fc17ef9b0 100644 --- a/backend/crates/gateway-store/src/redis/credential_leases.rs +++ b/backend/crates/gateway-store/src/redis/credential_leases.rs @@ -260,8 +260,13 @@ impl CredentialLeaseGuard { entity: "credential lease", message: "lease has already been released".to_owned(), })?; - let renewal = - crate::lease_renewal::LeaseRenewal::spawn(deadline, Some(cancellation), move |ttl| { + let renewal = crate::lease_renewal::LeaseRenewal::spawn( + deadline, + Some(( + cancellation, + gateway_core::lifecycle::RequestLeaseKind::Credential, + )), + move |ttl| { let repository = repository.clone(); let mut lease_request = lease_request.clone(); lease_request.ttl = ttl; @@ -272,7 +277,8 @@ impl CredentialLeaseGuard { .await .map(|grant| grant.is_some()) }) - }); + }, + ); Ok(RenewingSchedulingLease { _renewal: renewal, _guard: self, diff --git a/backend/crates/gateway-store/tests/backup/s3.rs b/backend/crates/gateway-store/tests/backup/s3.rs index d7a4c2461..e2ee300ae 100644 --- a/backend/crates/gateway-store/tests/backup/s3.rs +++ b/backend/crates/gateway-store/tests/backup/s3.rs @@ -102,9 +102,6 @@ async fn a_failed_part_stops_other_pending_parts_before_upload_returns() { let (mut abort, request, _) = accept_request(&probe.listener).await; assert!(request.starts_with("DELETE ") && request.contains("uploadId=")); respond(&mut abort, &xml_response("")).await; - let (mut delete, request, _) = accept_request(&probe.listener).await; - assert!(request.starts_with("DELETE ") && !request.contains("uploadId=")); - respond(&mut delete, &xml_response("")).await; let error = tokio::time::timeout(Duration::from_secs(3), probe.upload) .await .expect("failed upload returns without waiting for the other part") @@ -359,3 +356,103 @@ fn xml_response(body: &str) -> String { body.len() ) } + +#[tokio::test] +async fn connection_probe_requires_successful_final_delete() { + for delete_allowed in [true, false] { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let storage = BackupStorageConfig { + storage_revision: 1, + endpoint: format!("http://{}", listener.local_addr().unwrap()), + region: "auto".into(), + bucket: "backup-bucket".into(), + access_key_id: "test-access-key".into(), + secret_access_key: SecretString::from("test-secret-key"), + prefix: "backups".into(), + force_path_style: true, + }; + let probe = + tokio::spawn( + async move { S3ObjectStoreAdapter::new().test_connection(&storage).await }, + ); + let (mut put, request, body) = accept_request(&listener).await; + assert!(request.starts_with("PUT ")); + respond(&mut put, &xml_response("")).await; + let (mut head, request, _) = accept_request(&listener).await; + assert!(request.starts_with("HEAD ")); + respond(&mut head, "HTTP/1.1 200 OK\r\ncontent-length: 64\r\nx-amz-meta-probe: backup-connection-test\r\nconnection: close\r\n\r\n").await; + let (mut get, request, _) = accept_request(&listener).await; + assert!(request.starts_with("GET ")); + get.write_all( + format!( + "HTTP/1.1 200 OK\r\ncontent-length: {}\r\nconnection: close\r\n\r\n", + body.len() + ) + .as_bytes(), + ) + .await + .unwrap(); + get.write_all(&body).await.unwrap(); + get.shutdown().await.unwrap(); + let (mut delete, request, _) = accept_request(&listener).await; + assert!(request.starts_with("DELETE ")); + respond( + &mut delete, + &if delete_allowed { + xml_response("") + } else { + error_response("403 Forbidden", "AccessDenied") + }, + ) + .await; + let result = tokio::time::timeout(Duration::from_secs(3), probe) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(result.ok, delete_allowed); + assert_eq!(result.stage, "deleteObject"); + assert_eq!( + result.code, + (!delete_allowed).then_some(code::S3_PERMISSION_DENIED) + ); + } +} + +#[tokio::test] +async fn failed_multipart_retains_abort_failure_without_changing_primary_code() { + for cancelled in [false, true] { + let probe = start_upload(MULTIPART_SIZE).await; + let (mut part, request, _) = accept_request(&probe.listener).await; + assert!(request.starts_with("PUT ")); + if cancelled { + probe.cancellation.cancel(); + } else { + respond(&mut part, &error_response("403 Forbidden", "AccessDenied")).await; + } + let (mut abort, request, _) = accept_request(&probe.listener).await; + assert!(request.starts_with("DELETE ") && request.contains("uploadId=")); + respond(&mut abort, &error_response("403 Forbidden", "AccessDenied")).await; + let error = tokio::time::timeout(Duration::from_secs(3), probe.upload) + .await + .unwrap() + .unwrap() + .unwrap_err(); + assert_eq!( + error.code(), + if cancelled { + code::CANCELLED + } else { + code::S3_PERMISSION_DENIED + } + ); + if cancelled { + assert!(error.message().contains("上传已取消")); + } + assert!(error.message().contains("AbortMultipartUpload")); + assert!(error.message().contains(code::S3_PERMISSION_DENIED)); + assert!(error.message().contains("未安排自动重试")); + assert!(!error.message().contains("probe-upload")); + assert!(!error.message().contains("probe error")); + } +} diff --git a/backend/crates/gateway-store/tests/lease_renewal.rs b/backend/crates/gateway-store/tests/lease_renewal.rs new file mode 100644 index 000000000..08b3d03fc --- /dev/null +++ b/backend/crates/gateway-store/tests/lease_renewal.rs @@ -0,0 +1,125 @@ +//! 注入续期结果验证终态原因、恢复重试及显式截止,不连接外部 Redis + +#[path = "../src/lease_renewal.rs"] +mod implementation; + +use gateway_core::lifecycle::{ + CancellationReason, CancellationToken, Deadline, LeaseFailure, REQUEST_LEASE_TTL, + RequestLeaseKind, +}; +use implementation::LeaseRenewal; +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; + +fn temporary_error() -> gateway_store::StoreError { + gateway_store::StoreError::Unavailable { + backend: gateway_store::StoreBackend::Redis, + message: "injected renewal failure".into(), + source: None, + } +} + +#[tokio::test(start_paused = true)] +async fn missing_lease_and_pending_renewal_preserve_distinct_causes() { + for kind in [ + RequestLeaseKind::ClientAdmission, + RequestLeaseKind::Credential, + ] { + for pending in [false, true] { + let token = CancellationToken::new(); + let calls = Arc::new(AtomicUsize::new(0)); + let observed = calls.clone(); + let _guard = LeaseRenewal::spawn( + Deadline::default(), + Some((token.clone(), kind)), + move |_| { + observed.fetch_add(1, Ordering::SeqCst); + Box::pin(async move { + if pending { + std::future::pending().await + } else { + Ok(false) + } + }) + }, + ); + token.cancelled().await; + token.cancel(); + assert_eq!( + token.reason(), + Some(CancellationReason::Lease { + kind, + failure: if pending { + LeaseFailure::RenewalTimedOut + } else { + LeaseFailure::Lost + }, + }) + ); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + } +} + +#[tokio::test(start_paused = true)] +async fn temporary_failure_recovers_without_terminal_cancellation() { + let token = CancellationToken::new(); + let calls = Arc::new(AtomicUsize::new(0)); + let observed = calls.clone(); + let guard = LeaseRenewal::spawn( + Deadline::default(), + Some((token.clone(), RequestLeaseKind::Credential)), + move |_| { + let first = observed.fetch_add(1, Ordering::SeqCst) == 0; + Box::pin(async move { + if first { + Err(temporary_error()) + } else { + Ok(true) + } + }) + }, + ); + tokio::task::yield_now().await; + for _ in 0..4 { + tokio::time::advance(REQUEST_LEASE_TTL / 3).await; + tokio::task::yield_now().await; + } + assert!(calls.load(Ordering::SeqCst) >= 4); + assert_eq!(token.reason(), None); + drop(guard); +} + +#[tokio::test(start_paused = true)] +async fn exhausted_retry_window_retains_expiration_cause() { + let token = CancellationToken::new(); + let _guard = LeaseRenewal::spawn( + Deadline::default(), + Some((token.clone(), RequestLeaseKind::ClientAdmission)), + |_| Box::pin(async { Err(temporary_error()) }), + ); + token.cancelled().await; + assert_eq!( + token.reason(), + Some(CancellationReason::Lease { + kind: RequestLeaseKind::ClientAdmission, + failure: LeaseFailure::Expired, + }) + ); +} + +#[tokio::test] +async fn explicit_deadline_and_observation_lease_do_not_cancel_business() { + let token = CancellationToken::new(); + let _guard = LeaseRenewal::spawn( + std::time::SystemTime::UNIX_EPOCH.into(), + Some((token.clone(), RequestLeaseKind::Credential)), + |_| panic!("elapsed request does not renew"), + ); + let _observation = + LeaseRenewal::spawn(Deadline::default(), None, |_| Box::pin(async { Ok(false) })); + tokio::task::yield_now().await; + assert_eq!(token.reason(), None); +} diff --git a/backend/crates/gateway-store/tests/main.rs b/backend/crates/gateway-store/tests/main.rs index 965a96667..3b4310231 100644 --- a/backend/crates/gateway-store/tests/main.rs +++ b/backend/crates/gateway-store/tests/main.rs @@ -3,10 +3,13 @@ mod backup; mod bundle; mod config; +mod lease_renewal; mod postgres; mod redis; mod support; +use gateway_store::StoreResult; + #[derive(Default)] struct RecordingDiagnostics(std::sync::Mutex>); #[async_trait::async_trait]