starred/codex-proxy-rs
1
0
Fork 0
mirror of https://github.com/zyycn/codex-proxy-rs.git synced 2026-11-07 00:05:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve terminal failure and cleanup causes (#417)

Browse Source 
Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-10-13 11:31:10 +08:00
committed by GitHub
co-authored by Zyy Codex
parent 27cef052c9
commit a9a86071bf
29 changed files with 1040 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/crates/gateway-admin/src/backup/task.rs
+42 -17
View File
@@ -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,
backend/crates/gateway-admin/src/model/backup/mod.rs
+20
View File
@@ -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
}
}
/// 构造稳定错误码常量
backend/crates/gateway-admin/tests/backup/support.rs
+35
View File
@@ -536,6 +536,9 @@ impl DatabaseDumpPort for FakeDumpPort {
pub(crate) struct FakeObjectStore {
pub(crate) objects: Mutex<std::collections::HashMap<String, Vec<u8>>>,
upload_error: Option<BackupError>,
head_error: Option<BackupError>,
delete_error: Option<BackupError>,
connection_error: Option<BackupError>,
}
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<Vec<u8>> {
self.objects.lock().expect("objects").get(key).cloned()
}
@@ -563,6 +584,14 @@ impl BackupObjectStorePort for FakeObjectStore {
&self,
_config: &BackupStorageConfig,
) -> Result<ConnectionTestResult, gateway_admin::model::backup::BackupError> {
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<Option<BackupObjectMetadata>, 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(())
}
backend/crates/gateway-admin/tests/backup/task.rs
+55
View File
@@ -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
);
}
}
backend/crates/gateway-admin/tests/use_case/backup.rs
+48
View File
@@ -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
);
}
}
backend/crates/gateway-api/src/middleware/mod.rs
+14 -1
View File
@@ -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,
"升级后的中间件会话失败"
);
}
},
}
});
}
backend/crates/gateway-api/tests/middleware/mod.rs
+35 -1
View File
@@ -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::<Arc<dyn contract::upgrade::WebSocketUpgrade>>()
.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");
backend/crates/gateway-api/tests/middleware/session.rs
+79
View File
@@ -0,0 +1,79 @@
//! 验证升级后终态只记录一次安全分类，正常结束及已报告失败不重复输出
use super::*;
use tracing::{
Event, Metadata, Subscriber,
span::{Attributes, Id, Record},
};
#[derive(Clone, Default)]
struct Recorder(Arc<Mutex<Vec<String>>>);
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();
}
}
backend/crates/gateway-core/src/engine/coordinator.rs
+7 -1
View File
@@ -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() {
backend/crates/gateway-core/src/engine/middleware/http/upgrade.rs
+36 -1
View File
@@ -8,7 +8,42 @@ use crate::{engine::middleware::MiddlewareError, engine::middleware::websocket::
pub type PendingSession =
Shared<BoxFuture<'static, Result<Arc<dyn Session>, Arc<MiddlewareError>>>>;
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<String>,
},
}
impl From<MiddlewareError> 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 {
backend/crates/gateway-core/src/lifecycle.rs
+76 -13
View File
@@ -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<CancellationReason>,
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::<Vec<_>>();
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<CancellationReason> {
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<CancellationState>) -> 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;
backend/crates/gateway-core/tests/engine/coordinator.rs
+42
View File
@@ -53,6 +53,7 @@ use serde_json::{Map, Value, json};
#[derive(Debug, Clone, PartialEq)]
struct FinalState {
error_message: Option<String>,
error_kind: Option<GatewayErrorKind>,
diagnostic_trace_json: Option<String>,
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())
);
});
}
backend/crates/gateway-core/tests/lifecycle.rs
+22
View File
@@ -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));
}
backend/crates/gateway-host/src/system_update/mod.rs
+1
View File
@@ -11,6 +11,7 @@ mod process;
mod release;
mod state;
mod swap;
mod swap_paths;
pub use config::SystemUpdateConfig;
pub use operations::ProcessSystemOperations;
backend/crates/gateway-host/src/system_update/swap.rs
+2 -24
View File
@@ -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<()> {
backend/crates/gateway-host/src/system_update/swap_paths.rs
+59
View File
@@ -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()
),
),
}
}
backend/crates/gateway-host/tests/system_update/mod.rs
+1
View File
@@ -7,6 +7,7 @@ mod installation;
mod operations;
mod release;
mod state;
mod swap_paths;
use std::fs;
use std::path::PathBuf;
backend/crates/gateway-host/tests/system_update/swap_paths.rs
+76
View File
@@ -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 [(&current, "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(&current, &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 = [&current, &backup, &swap]
.into_iter()
.filter_map(|path| {
fs::read_to_string(if directory {
path.join("asset")
} else {
path.clone()
})
.ok()
})
.collect::<Vec<_>>();
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");
}
}
}
}
}
backend/crates/gateway-plugin/runtime/src/adapter/middleware/http.rs
+33 -4
View File
@@ -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()),
}
}
backend/crates/gateway-plugin/runtime/tests/generation/policy/http/websocket/mod.rs
+62
View File
@@ -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<dyn upgrade::WebSocketUpgrade>);
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::<upgrade::Upgraded>()
.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;
}
}
backend/crates/gateway-plugin/runtime/tests/support/middleware/mod.rs
+16
View File
@@ -331,6 +331,22 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}
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 {
backend/crates/gateway-store/Cargo.toml
+1 -1
View File
@@ -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
backend/crates/gateway-store/src/backup/s3.rs
+20 -12
View File
@@ -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(
backend/crates/gateway-store/src/lease_renewal.rs
+17 -10
View File
@@ -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<CancellationToken>,
cancellation: Option<(CancellationToken, RequestLeaseKind)>,
mut renew: impl FnMut(Duration) -> BoxFuture<'static, StoreResult<bool>> + 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,
});
}
}))
}
backend/crates/gateway-store/src/redis/client_admission.rs
+4 -1
View File
@@ -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();
backend/crates/gateway-store/src/redis/credential_leases.rs
+9 -3
View File
@@ -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,
backend/crates/gateway-store/tests/backup/s3.rs
+100 -3
View File
@@ -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"));
}
}
backend/crates/gateway-store/tests/lease_renewal.rs
+125
View File
@@ -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);
}
backend/crates/gateway-store/tests/main.rs
+3
View File
@@ -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<Vec<gateway_core::diagnostics::OperationalFailure>>);
#[async_trait::async_trait]