From ef11b72d8a977e0e58613eb2aac2bbc60a66d6ad Mon Sep 17 00:00:00 2001 From: Zyy <0x@ainz.cc> Date: Wed, 21 Oct 2026 11:08:15 +0800 Subject: [PATCH] fix(workers): recover daemon health after successful work (#499) Report per-run daemon progress to reset health failures and restart backoff after successful work, with stale-run isolation and recovery regression tests. Co-authored-by: Zyy Co-authored-by: Codex --- .../crates/gateway-admin/src/backup/task.rs | 12 +- .../src/use_case/import_tasks.rs | 9 +- .../gateway-admin/tests/backup/support.rs | 15 ++ .../crates/gateway-admin/tests/backup/task.rs | 38 ++++ .../tests/use_case/import_tasks.rs | 6 +- .../gateway-core/src/runtime/publisher.rs | 9 +- backend/crates/gateway-core/src/task.rs | 54 ++++- .../crates/gateway-core/tests/runtime/mod.rs | 5 +- backend/crates/gateway-host/src/workers.rs | 70 ++++--- backend/crates/gateway-host/tests/serve.rs | 8 +- backend/crates/gateway-host/tests/workers.rs | 192 +++++++++++++++++- .../src/generation/prepare/maintenance.rs | 11 +- .../runtime/tests/callback/keys.rs | 8 +- .../runtime/tests/generation/maintenance.rs | 5 +- .../gateway-store/src/postgres/client_keys.rs | 7 +- .../src/postgres/execution_buffer.rs | 28 ++- .../src/redis/coordination_buffer.rs | 14 +- backend/crates/gateway-store/src/workers.rs | 3 +- .../tests/postgres/client_keys.rs | 8 +- .../tests/postgres/execution_buffer.rs | 39 ++-- .../tests/redis/coordination_buffer.rs | 3 +- .../providers/openai/src/provider/workers.rs | 9 +- docs/architecture.md | 5 + 23 files changed, 463 insertions(+), 95 deletions(-) diff --git a/backend/crates/gateway-admin/src/backup/task.rs b/backend/crates/gateway-admin/src/backup/task.rs index bd5e1c123..457149746 100644 --- a/backend/crates/gateway-admin/src/backup/task.rs +++ b/backend/crates/gateway-admin/src/backup/task.rs @@ -11,7 +11,7 @@ use chrono::{DateTime, Utc}; use futures::future::BoxFuture; use gateway_core::{ lifecycle::CancellationToken, - task::{DaemonTask, WorkerTaskError}, + task::{DaemonTask, WorkerDaemonContext, WorkerTaskError}, }; use tracing::{info, warn}; @@ -75,18 +75,20 @@ impl BackupTask { } impl DaemonTask for BackupTask { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { - Box::pin(async move { self.run_daemon(cancellation).await }) + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + Box::pin(async move { self.run_daemon(context).await }) } } impl BackupTask { - async fn run_daemon(&self, cancellation: CancellationToken) -> Result<(), WorkerTaskError> { + async fn run_daemon(&self, context: WorkerDaemonContext) -> Result<(), WorkerTaskError> { + let cancellation = context.cancellation(); loop { if cancellation.is_cancelled() { return Ok(()); } - self.run_cycle(&cancellation).await?; + self.run_cycle(cancellation).await?; + context.succeeded(); tokio::select! { _ = cancellation.cancelled() => return Ok(()), diff --git a/backend/crates/gateway-admin/src/use_case/import_tasks.rs b/backend/crates/gateway-admin/src/use_case/import_tasks.rs index d3c3c35a0..a06b1512c 100644 --- a/backend/crates/gateway-admin/src/use_case/import_tasks.rs +++ b/backend/crates/gateway-admin/src/use_case/import_tasks.rs @@ -9,10 +9,7 @@ use std::{ use chrono::Utc; use futures::{FutureExt as _, StreamExt as _, future::BoxFuture, stream::FuturesUnordered}; -use gateway_core::{ - lifecycle::CancellationToken, - task::{DaemonTask, WorkerTaskError}, -}; +use gateway_core::task::{DaemonTask, WorkerDaemonContext, WorkerTaskError}; use tokio::{sync::Notify, time::Instant}; use uuid::Uuid; @@ -332,8 +329,9 @@ impl ImportTasksService for DefaultImportTasksService { pub(crate) struct ImportTaskWorker(pub Arc); impl DaemonTask for ImportTaskWorker { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); let mut active = FuturesUnordered::new(); // 固定一个取消 future，避免循环反复注册中立 CancellationToken 的 waiter let cancelled = cancellation.cancelled(); @@ -354,6 +352,7 @@ impl DaemonTask for ImportTaskWorker { }; active.push(self.0.execute(id, index, input)); } + context.succeeded(); tokio::select! { () = &mut cancelled => {}, _ = active.next(), if !active.is_empty() => {}, diff --git a/backend/crates/gateway-admin/tests/backup/support.rs b/backend/crates/gateway-admin/tests/backup/support.rs index 64103243c..9510999e1 100644 --- a/backend/crates/gateway-admin/tests/backup/support.rs +++ b/backend/crates/gateway-admin/tests/backup/support.rs @@ -5,6 +5,7 @@ //! 供 backup::task 与 use_case::backup 的测试复用 use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, Ordering}; use async_trait::async_trait; use chrono::{DateTime, Utc}; @@ -92,6 +93,7 @@ pub(crate) fn queued_record(seed: &BackupRecordSeed) -> BackupRecord { /// 内存版备份仓储 pub(crate) struct FakeBackupRepository { settings: Mutex, + settings_unavailable: AtomicBool, records: Mutex>, } @@ -99,10 +101,16 @@ impl FakeBackupRepository { pub(crate) fn new(settings: BackupSettings) -> Self { Self { settings: Mutex::new(settings), + settings_unavailable: AtomicBool::new(false), records: Mutex::new(Vec::new()), } } + pub(crate) fn set_settings_unavailable(&self, unavailable: bool) { + self.settings_unavailable + .store(unavailable, Ordering::SeqCst); + } + /// 测试辅助：读取当前全部记录 pub(crate) fn all_records(&self) -> Vec { self.records.lock().expect("records").clone() @@ -134,6 +142,13 @@ impl FakeBackupRepository { #[async_trait] impl BackupRepository for FakeBackupRepository { async fn load_settings(&self) -> AdminStoreResult { + if self.settings_unavailable.load(Ordering::SeqCst) { + return Err(AdminStoreError::new( + AdminStoreErrorKind::Unavailable, + "backup", + "test store unavailable", + )); + } Ok(self.settings.lock().expect("settings").clone()) } diff --git a/backend/crates/gateway-admin/tests/backup/task.rs b/backend/crates/gateway-admin/tests/backup/task.rs index 39395601f..60a1d1182 100644 --- a/backend/crates/gateway-admin/tests/backup/task.rs +++ b/backend/crates/gateway-admin/tests/backup/task.rs @@ -11,6 +11,7 @@ use gateway_admin::{ ports::backup::{BackupRepository, DatabaseDumpPort, DumpRequest}, }; use gateway_core::lifecycle::CancellationToken; +use gateway_core::task::{DaemonTask, WorkerDaemonContext}; use super::support::{ DUMP_CONTENT, FakeBackupRepository, FakeDumpPort, FakeObjectStore, backup_id, @@ -379,3 +380,40 @@ async fn manual_backup_preserves_primary_and_final_delete_failures() { ); } } + +#[tokio::test(start_paused = true)] +async fn daemon_reports_success_only_after_a_complete_cycle_and_recovers_after_store_failure() { + let repository = Arc::new(FakeBackupRepository::new(configured_settings())); + let task = Arc::new(BackupTask::new( + repository.clone(), + Arc::new(FakeDumpPort::new()), + Arc::new(FakeObjectStore::new()), + )); + let context = WorkerDaemonContext::new(CancellationToken::new()); + repository.set_settings_unavailable(true); + let error = task.run(context.clone()).await.unwrap_err(); + assert_eq!(error.as_safe_str(), "backup store failed"); + assert_eq!(context.completed_cycles(), 0); + + repository.set_settings_unavailable(false); + let recovered = WorkerDaemonContext::new(CancellationToken::new()); + let worker = tokio::spawn({ + let task = task.clone(); + let context = recovered.clone(); + async move { task.run(context).await } + }); + recovered.progress_since(0).await; + assert_eq!(recovered.completed_cycles(), 1); + repository.set_settings_unavailable(true); + tokio::time::advance(std::time::Duration::from_secs(30)).await; + assert_eq!( + worker.await.unwrap().unwrap_err().as_safe_str(), + "backup store failed" + ); + assert_eq!(recovered.completed_cycles(), 1); + + let cancelled = WorkerDaemonContext::new(CancellationToken::new()); + cancelled.cancellation().cancel(); + task.run(cancelled.clone()).await.unwrap(); + assert_eq!(cancelled.completed_cycles(), 0); +} diff --git a/backend/crates/gateway-admin/tests/use_case/import_tasks.rs b/backend/crates/gateway-admin/tests/use_case/import_tasks.rs index 2ba8d9d7a..5cf53f9a5 100644 --- a/backend/crates/gateway-admin/tests/use_case/import_tasks.rs +++ b/backend/crates/gateway-admin/tests/use_case/import_tasks.rs @@ -14,7 +14,7 @@ use gateway_admin::{ use gateway_core::{ lifecycle::CancellationToken, routing::ProviderKind, - task::{WorkerContribution, WorkerKind, WorkerRunnable}, + task::{WorkerContribution, WorkerDaemonContext, WorkerKind, WorkerRunnable}, }; use uuid::Uuid; @@ -75,7 +75,9 @@ fn start(bundle: &mut AdminBundle) -> (CancellationToken, tokio::task::JoinHandl let cancellation = CancellationToken::new(); let token = cancellation.clone(); let handle = tokio::spawn(async move { - task.run(token).await.expect("import daemon exits cleanly"); + task.run(WorkerDaemonContext::new(token)) + .await + .expect("import daemon exits cleanly"); }); (cancellation, handle) } diff --git a/backend/crates/gateway-core/src/runtime/publisher.rs b/backend/crates/gateway-core/src/runtime/publisher.rs index 90b586a4f..cd3720c87 100644 --- a/backend/crates/gateway-core/src/runtime/publisher.rs +++ b/backend/crates/gateway-core/src/runtime/publisher.rs @@ -372,8 +372,12 @@ struct RuntimeSnapshotSubscriptionTask { } impl DaemonTask for RuntimeSnapshotSubscriptionTask { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run( + &self, + context: crate::task::WorkerDaemonContext, + ) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); let mut retry_delay = INITIAL_BACKOFF; loop { if cancellation.is_cancelled() { @@ -383,10 +387,11 @@ impl DaemonTask for RuntimeSnapshotSubscriptionTask { let mut subscription = match subscription { Ok(subscription) => { retry_delay = INITIAL_BACKOFF; + context.succeeded(); subscription } Err(_) => { - wait_or_cancel(&cancellation, retry_delay).await; + wait_or_cancel(cancellation, retry_delay).await; retry_delay = (retry_delay * 2).min(MAXIMUM_BACKOFF); continue; } diff --git a/backend/crates/gateway-core/src/task.rs b/backend/crates/gateway-core/src/task.rs index 52d3f4469..2987ab8e8 100644 --- a/backend/crates/gateway-core/src/task.rs +++ b/backend/crates/gateway-core/src/task.rs @@ -6,8 +6,11 @@ use std::fmt; use std::num::NonZeroU64; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; +use event_listener::Event; use futures::future::BoxFuture; use crate::lifecycle::CancellationToken; @@ -341,11 +344,60 @@ pub trait ScheduledTask: Send + Sync { fn run_cycle(&self, context: WorkerCycleContext) -> BoxFuture<'_, Result<(), WorkerTaskError>>; } +/// 一次 daemon 运行的取消与成功进度，重启时必须创建新上下文 +#[derive(Clone)] +pub struct WorkerDaemonContext { + cancellation: CancellationToken, + progress: Arc, +} + +#[derive(Default)] +struct DaemonProgress { + completed: AtomicU64, + changed: Event, +} + +impl WorkerDaemonContext { + #[must_use] + pub fn new(cancellation: CancellationToken) -> Self { + Self { + cancellation, + progress: Arc::default(), + } + } + + #[must_use] + pub const fn cancellation(&self) -> &CancellationToken { + &self.cancellation + } + + /// 仅在实际工作或依赖检查成功后通知 Host；进入 run 不代表恢复 + pub fn succeeded(&self) { + if !self.cancellation.is_cancelled() { + self.progress.completed.fetch_add(1, Ordering::Release); + self.progress.changed.notify(usize::MAX); + } + } + + #[must_use] + pub fn completed_cycles(&self) -> u64 { + self.progress.completed.load(Ordering::Acquire) + } + + /// 先注册再检查计数，避免报告与等待交错时丢失成功通知 + pub async fn progress_since(&self, completed: u64) { + let changed = self.progress.changed.listen(); + if self.completed_cycles() == completed { + changed.await; + } + } +} + /// 长驻任务 /// /// 实现负责连接级重连退避；Host 负责任务退出或 panic 后的重启退避 pub trait DaemonTask: Send + Sync { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>>; + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>>; } /// Host 可执行的两种任务形态，无法组合出“守护任务 + 周期调度”等非法状态 diff --git a/backend/crates/gateway-core/tests/runtime/mod.rs b/backend/crates/gateway-core/tests/runtime/mod.rs index 6189181be..9dc7bf906 100644 --- a/backend/crates/gateway-core/tests/runtime/mod.rs +++ b/backend/crates/gateway-core/tests/runtime/mod.rs @@ -28,7 +28,8 @@ use gateway_core::runtime::{ }; use gateway_core::settings::SettingsValues; use gateway_core::task::{ - ScheduledTask, WorkerContribution, WorkerCycleContext, WorkerKind, WorkerRunnable, + ScheduledTask, WorkerContribution, WorkerCycleContext, WorkerDaemonContext, WorkerKind, + WorkerRunnable, }; type ReadGate = Arc>>>>; @@ -490,7 +491,7 @@ fn subscription_refresh_should_wait_for_inflight_compile_and_reload_authoritativ }) .expect("subscription task"); let cancellation = CancellationToken::new(); - let mut daemon = task.run(cancellation.clone()); + let mut daemon = task.run(WorkerDaemonContext::new(cancellation.clone())); let mut old = Box::pin(publisher.refresh()); assert!(futures::poll!(old.as_mut()).is_pending()); *store.facts.lock().expect("facts lock") = Ok(scoped_facts(11, false)); diff --git a/backend/crates/gateway-host/src/workers.rs b/backend/crates/gateway-host/src/workers.rs index 889bb5045..a5848934f 100644 --- a/backend/crates/gateway-host/src/workers.rs +++ b/backend/crates/gateway-host/src/workers.rs @@ -14,9 +14,9 @@ use gateway_core::health::{ use gateway_core::lifecycle::CancellationToken; use gateway_core::task::{ DaemonRestartPolicy, DaemonTask, ScheduledTask, WorkerContribution, WorkerCycleContext, - WorkerDefinitionError, WorkerDisabledReason, WorkerId, WorkerKind, WorkerLeaderLeaseGuard, - WorkerLeaderLeasePort, WorkerLeaseAcquisition, WorkerLeaseRequest, WorkerRegistration, - WorkerRunnable, WorkerSchedule, WorkerTaskError, + WorkerDaemonContext, WorkerDefinitionError, WorkerDisabledReason, WorkerId, WorkerKind, + WorkerLeaderLeaseGuard, WorkerLeaderLeasePort, WorkerLeaseAcquisition, WorkerLeaseRequest, + WorkerRegistration, WorkerRunnable, WorkerSchedule, WorkerTaskError, }; use tokio::task::JoinHandle; @@ -235,7 +235,7 @@ pub enum WorkerStartError { #[derive(Clone)] struct HealthEntry { snapshot: WorkerHealthSnapshot, - success_freshness: Duration, + success_freshness: Option, } #[derive(Default)] @@ -253,11 +253,14 @@ impl WorkerHealthRegistry { .values() .map(|registration| { let freshness = match ®istration.runnable { - WorkerRunnable::Scheduled { schedule, .. } => (*schedule) - .interval() - .saturating_mul(2) - .saturating_add(schedule.maximum_backoff()), - WorkerRunnable::Daemon { restart, .. } => restart.maximum_backoff(), + WorkerRunnable::Scheduled { schedule, .. } => Some( + (*schedule) + .interval() + .saturating_mul(2) + .saturating_add(schedule.maximum_backoff()), + ), + // daemon 可长期空闲或执行备份，重启退避不约束工作耗时 + WorkerRunnable::Daemon { .. } => None, }; let key = WorkerHealthKey::Task(registration.id.clone()); ( @@ -275,7 +278,7 @@ impl WorkerHealthRegistry { key.clone(), HealthEntry { snapshot: empty_snapshot(key, WorkerRuntimeState::Disabled), - success_freshness: Duration::ZERO, + success_freshness: None, }, ) })); @@ -315,11 +318,11 @@ impl WorkerHealthRegistry { }); } - fn succeeded(&self, id: &WorkerId) { + fn succeeded(&self, id: &WorkerId, runtime_state: WorkerRuntimeState, cycles: u64) { self.update(id, |state| { - state.state = WorkerRuntimeState::Idle; + state.state = runtime_state; state.consecutive_failures = 0; - state.completed_cycles = state.completed_cycles.saturating_add(1); + state.completed_cycles = state.completed_cycles.saturating_add(cycles); state.last_success_at = Some(SystemTime::now()); state.last_error = None; }); @@ -361,8 +364,11 @@ impl WorkerHealthSource for WorkerHealthRegistry { snapshot.state, WorkerRuntimeState::Running | WorkerRuntimeState::Idle ) && snapshot.last_success_at.is_some_and(|last_success| { - now.duration_since(last_success) - .is_ok_and(|age| age > entry.success_freshness) + now.duration_since(last_success).is_ok_and(|age| { + entry + .success_freshness + .is_some_and(|freshness| age > freshness) + }) }); if stale { snapshot.state = WorkerRuntimeState::BackingOff; @@ -407,7 +413,7 @@ async fn supervise_scheduled( let delay = match outcome { ScheduledOutcome::Succeeded => { - health.succeeded(&id); + health.succeeded(&id, WorkerRuntimeState::Idle, 1); backoff = schedule.initial_backoff(); schedule.interval() } @@ -595,17 +601,33 @@ async fn supervise_daemon( break; } health.running(&id, None); - let daemon_cancel = CancellationToken::new(); - let run = AssertUnwindSafe(task.run(daemon_cancel.clone())).catch_unwind(); + let context = WorkerDaemonContext::new(CancellationToken::new()); + let run = AssertUnwindSafe(task.run(context.clone())).catch_unwind(); tokio::pin!(run); - let result = tokio::select! { - result = &mut run => Some(result), - () = cancellation.cancelled() => { - daemon_cancel.cancel(); - let _ = run.await; - None + let mut completed = 0; + let result = loop { + let outcome = tokio::select! { + biased; + () = cancellation.cancelled() => { + context.cancellation().cancel(); + let _ = run.await; + break None; + } + result = &mut run => Some(result), + () = context.progress_since(completed) => None, + }; + // 成功与退出可在同一 poll 内发生，先消费进度再记录本次退出故障 + let cycles = context.completed_cycles(); + if cycles > completed { + health.succeeded(&id, WorkerRuntimeState::Running, cycles - completed); + completed = cycles; + backoff = restart.initial_backoff(); + } + if let Some(result) = outcome { + break Some(result); } }; + context.cancellation().cancel(); let Some(result) = result else { break }; let error = match result { Ok(Ok(())) => WorkerTaskError::safe("daemon exited unexpectedly"), diff --git a/backend/crates/gateway-host/tests/serve.rs b/backend/crates/gateway-host/tests/serve.rs index 6acaccf5b..51bb3067a 100644 --- a/backend/crates/gateway-host/tests/serve.rs +++ b/backend/crates/gateway-host/tests/serve.rs @@ -12,7 +12,7 @@ use gateway_admin::ports::system::{ use gateway_core::diagnostics::{OperationalDiagnostics, OperationalFailure}; use gateway_core::lifecycle::CancellationToken; use gateway_core::task::{ - DaemonRestartPolicy, DaemonTask, WorkerContribution, WorkerId, WorkerKind, + DaemonRestartPolicy, DaemonTask, WorkerContribution, WorkerDaemonContext, WorkerId, WorkerKind, WorkerLeaderLeasePort, WorkerLeaseAcquisition, WorkerLeaseError, WorkerLeaseRequest, WorkerRegistration, WorkerRunnable, WorkerTaskError, }; @@ -365,8 +365,9 @@ impl Drop for TaskLifetime { } impl DaemonTask for WriterTask { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); let _lifetime = TaskLifetime(self.probe.dropped.clone()); let mut received = self.received.lock().await; self.probe.started.cancel(); @@ -394,8 +395,9 @@ impl DaemonTask for WriterTask { struct IdleTask; impl DaemonTask for IdleTask { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); cancellation.cancelled().await; Ok(()) }) diff --git a/backend/crates/gateway-host/tests/workers.rs b/backend/crates/gateway-host/tests/workers.rs index 767503d7e..f518f9514 100644 --- a/backend/crates/gateway-host/tests/workers.rs +++ b/backend/crates/gateway-host/tests/workers.rs @@ -11,10 +11,10 @@ use futures::future::BoxFuture; use gateway_core::health::{WorkerHealthKey, WorkerHealthSnapshot, WorkerRuntimeState}; use gateway_core::lifecycle::CancellationToken; use gateway_core::task::{ - ScheduledTask, WorkerContribution, WorkerCycleContext, WorkerDisabledReason, - WorkerFencingToken, WorkerId, WorkerKind, WorkerLeaderLeaseGuard, WorkerLeaderLeasePort, - WorkerLeaseAcquisition, WorkerLeaseError, WorkerLeaseRequest, WorkerRegistration, - WorkerRunnable, WorkerSchedule, WorkerTaskError, + DaemonRestartPolicy, DaemonTask, ScheduledTask, WorkerContribution, WorkerCycleContext, + WorkerDaemonContext, WorkerDisabledReason, WorkerFencingToken, WorkerId, WorkerKind, + WorkerLeaderLeaseGuard, WorkerLeaderLeasePort, WorkerLeaseAcquisition, WorkerLeaseError, + WorkerLeaseRequest, WorkerRegistration, WorkerRunnable, WorkerSchedule, WorkerTaskError, }; use gateway_host::workers::{WorkerStartError, WorkerSupervisor}; use tokio::sync::Notify; @@ -1023,3 +1023,187 @@ async fn worker_boundary_records_one_native_failure_and_recovers_health() { .contains("PRIVATE_WORKER_NATIVE_CAUSE") ); } + +// 每次运行保存自己的上下文，测试可模拟已退出任务的迟到通知 +struct ControlledDaemon { + contexts: Arc>>, + commands: tokio::sync::Mutex>, +} + +enum DaemonCommand { + Succeed, + Fail, + SucceedThenFail, + Panic, + Exit, +} + +impl DaemonTask for ControlledDaemon { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + Box::pin(async move { + lock_unpoisoned(&self.contexts).push(context.clone()); + let mut commands = self.commands.lock().await; + loop { + let command = tokio::select! { + () = context.cancellation().cancelled() => return Ok(()), + command = commands.recv() => command.expect("daemon command"), + }; + match command { + DaemonCommand::Succeed => context.succeeded(), + DaemonCommand::SucceedThenFail => { + context.succeeded(); + return Err(WorkerTaskError::safe("test daemon failure")); + } + DaemonCommand::Fail => { + return Err(WorkerTaskError::safe("test daemon failure")); + } + DaemonCommand::Panic => panic!("test daemon panic"), + DaemonCommand::Exit => return Ok(()), + } + } + }) + } +} + +fn daemon_plan(task: impl DaemonTask + 'static) -> Vec { + complete_plan(vec![WorkerContribution::Registration( + WorkerRegistration::try_new( + WorkerId::try_new(WorkerKind::Backup, "backup").unwrap(), + WorkerRunnable::Daemon { + restart: DaemonRestartPolicy::try_new( + Duration::from_millis(10), + Duration::from_millis(20), + ) + .unwrap(), + task: Box::new(task), + }, + ) + .unwrap(), + )]) +} + +#[tokio::test(start_paused = true)] +async fn daemon_recovers_only_after_current_run_succeeds_and_resets_backoff() { + let (commands, receiver) = tokio::sync::mpsc::unbounded_channel(); + let contexts = Arc::new(Mutex::new(Vec::new())); + let supervisor = supervisor(); + let id = WorkerId::try_new(WorkerKind::Backup, "backup").unwrap(); + supervisor + .start( + daemon_plan(ControlledDaemon { + contexts: contexts.clone(), + commands: tokio::sync::Mutex::new(receiver), + }), + Arc::new(FakeLeasePort::default()), + Arc::new(RecordingDiagnostics::default()), + ) + .unwrap(); + yield_until(|| lock_unpoisoned(&contexts).len() == 1).await; + + // 连续失败逐步退避；仅重新进入 run 不能恢复健康 + for (attempt, delay) in [(1, 10), (2, 20)] { + commands.send(DaemonCommand::Fail).unwrap(); + yield_until(|| task_health(&supervisor, &id).state == WorkerRuntimeState::BackingOff).await; + assert_eq!(task_health(&supervisor, &id).consecutive_failures, attempt); + tokio::time::advance(Duration::from_millis(delay - 1)).await; + assert_eq!(lock_unpoisoned(&contexts).len(), attempt as usize); + tokio::time::advance(Duration::from_millis(1)).await; + yield_until(|| lock_unpoisoned(&contexts).len() == attempt as usize + 1).await; + assert_eq!(task_health(&supervisor, &id).consecutive_failures, attempt); + } + lock_unpoisoned(&contexts)[0].succeeded(); + tokio::task::yield_now().await; + assert_eq!(task_health(&supervisor, &id).consecutive_failures, 2); + + commands.send(DaemonCommand::Succeed).unwrap(); + yield_until(|| task_health(&supervisor, &id).consecutive_failures == 0).await; + let recovered = task_health(&supervisor, &id); + assert_eq!(recovered.state, WorkerRuntimeState::Running); + assert_eq!(recovered.completed_cycles, 1); + assert!(recovered.last_error.is_none()); + assert!(recovered.last_success_at.is_some()); + assert!(recovered.last_failure_at.is_some()); + + commands.send(DaemonCommand::Fail).unwrap(); + yield_until(|| task_health(&supervisor, &id).state == WorkerRuntimeState::BackingOff).await; + assert_eq!(task_health(&supervisor, &id).consecutive_failures, 1); + tokio::time::advance(Duration::from_millis(10)).await; + yield_until(|| lock_unpoisoned(&contexts).len() == 4).await; + commands.send(DaemonCommand::Fail).unwrap(); + yield_until(|| task_health(&supervisor, &id).state == WorkerRuntimeState::BackingOff).await; + tokio::time::advance(Duration::from_millis(20)).await; + yield_until(|| lock_unpoisoned(&contexts).len() == 5).await; + + // 同一 poll 中先成功再退出，必须保留最新故障并重新从初始退避开始 + commands.send(DaemonCommand::SucceedThenFail).unwrap(); + yield_until(|| task_health(&supervisor, &id).state == WorkerRuntimeState::BackingOff).await; + assert_eq!(task_health(&supervisor, &id).consecutive_failures, 1); + assert_eq!(task_health(&supervisor, &id).completed_cycles, 2); + tokio::time::advance(Duration::from_millis(10)).await; + yield_until(|| lock_unpoisoned(&contexts).len() == 6).await; + supervisor.shutdown(Duration::from_secs(1)).await; + lock_unpoisoned(&contexts)[5].succeeded(); + assert_eq!( + task_health(&supervisor, &id).state, + WorkerRuntimeState::Stopped + ); +} + +#[tokio::test] +async fn daemon_success_remains_healthy_while_idle_beyond_restart_backoff() { + let (commands, receiver) = tokio::sync::mpsc::unbounded_channel(); + let supervisor = supervisor(); + let id = WorkerId::try_new(WorkerKind::Backup, "backup").unwrap(); + supervisor + .start( + daemon_plan(ControlledDaemon { + contexts: Arc::default(), + commands: tokio::sync::Mutex::new(receiver), + }), + Arc::new(FakeLeasePort::default()), + Arc::new(RecordingDiagnostics::default()), + ) + .unwrap(); + commands.send(DaemonCommand::Succeed).unwrap(); + wait_until(|| task_health(&supervisor, &id).completed_cycles == 1).await; + tokio::time::sleep(Duration::from_millis(50)).await; + let health = task_health(&supervisor, &id); + assert_eq!(health.state, WorkerRuntimeState::Running); + assert_eq!(health.consecutive_failures, 0); + supervisor.shutdown(Duration::from_secs(1)).await; +} + +#[tokio::test(start_paused = true)] +async fn daemon_panic_and_unexpected_exit_remain_unhealthy_until_success() { + for (command, error) in [ + (DaemonCommand::Panic, "daemon panicked"), + (DaemonCommand::Exit, "daemon exited unexpectedly"), + ] { + let (commands, receiver) = tokio::sync::mpsc::unbounded_channel(); + let contexts = Arc::new(Mutex::new(Vec::new())); + let supervisor = supervisor(); + let id = WorkerId::try_new(WorkerKind::Backup, "backup").unwrap(); + supervisor + .start( + daemon_plan(ControlledDaemon { + contexts: contexts.clone(), + commands: tokio::sync::Mutex::new(receiver), + }), + Arc::new(FakeLeasePort::default()), + Arc::new(RecordingDiagnostics::default()), + ) + .unwrap(); + commands.send(command).unwrap(); + yield_until(|| task_health(&supervisor, &id).state == WorkerRuntimeState::BackingOff).await; + assert_eq!( + task_health(&supervisor, &id).last_error.as_deref(), + Some(error) + ); + tokio::time::advance(Duration::from_millis(10)).await; + yield_until(|| lock_unpoisoned(&contexts).len() == 2).await; + assert_eq!(task_health(&supervisor, &id).consecutive_failures, 1); + commands.send(DaemonCommand::Succeed).unwrap(); + yield_until(|| task_health(&supervisor, &id).consecutive_failures == 0).await; + supervisor.shutdown(Duration::from_secs(1)).await; + } +} diff --git a/backend/crates/gateway-plugin/runtime/src/generation/prepare/maintenance.rs b/backend/crates/gateway-plugin/runtime/src/generation/prepare/maintenance.rs index 637befefb..514c9a12e 100644 --- a/backend/crates/gateway-plugin/runtime/src/generation/prepare/maintenance.rs +++ b/backend/crates/gateway-plugin/runtime/src/generation/prepare/maintenance.rs @@ -7,8 +7,9 @@ use gateway_core::{ lifecycle::CancellationToken, runtime::RuntimeSnapshotHandle, task::{ - DaemonRestartPolicy, DaemonTask, WorkerContribution, WorkerDefinitionError, WorkerId, - WorkerKind, WorkerRegistration, WorkerRunnable, WorkerTaskError, + DaemonRestartPolicy, DaemonTask, WorkerContribution, WorkerDaemonContext, + WorkerDefinitionError, WorkerId, WorkerKind, WorkerRegistration, WorkerRunnable, + WorkerTaskError, }, }; use gateway_plugin_sdk::Stage; @@ -52,8 +53,9 @@ impl PluginRuntime { } impl DaemonTask for MaintenanceTask { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); let mut publications = self.snapshots.subscribe_publications(); let mut active: Option> = None; let mut revision = None; @@ -91,6 +93,9 @@ impl DaemonTask for MaintenanceTask { } } revision = snapshot.as_ref().map(|s| s.revision()); + if snapshot.is_some() { + context.succeeded(); + } // 发布变更优先于下一轮维护；不在快照发布锁中等待插件或数据库 tokio::select! { biased; diff --git a/backend/crates/gateway-plugin/runtime/tests/callback/keys.rs b/backend/crates/gateway-plugin/runtime/tests/callback/keys.rs index 95f4bf7bb..a16bc8a05 100644 --- a/backend/crates/gateway-plugin/runtime/tests/callback/keys.rs +++ b/backend/crates/gateway-plugin/runtime/tests/callback/keys.rs @@ -16,7 +16,7 @@ use gateway_admin::{ use gateway_core::{ lifecycle::CancellationToken, policy::ClientApiKeyId, - task::{WorkerContribution, WorkerRunnable}, + task::{WorkerContribution, WorkerDaemonContext, WorkerRunnable}, }; use serde_json::{Value, json}; use tokio::{sync::Notify, time::timeout}; @@ -277,7 +277,11 @@ async fn published_maintenance_resets_budget_with_only_budget_permission() { }; let stop = CancellationToken::new(); let cancellation = stop.clone(); - let worker = tokio::spawn(async move { task.run(cancellation).await.unwrap() }); + let worker = tokio::spawn(async move { + task.run(WorkerDaemonContext::new(cancellation)) + .await + .unwrap() + }); let result: Value = timeout(Duration::from_secs(15), async { loop { if let Ok(content) = std::fs::read_to_string(&marker) diff --git a/backend/crates/gateway-plugin/runtime/tests/generation/maintenance.rs b/backend/crates/gateway-plugin/runtime/tests/generation/maintenance.rs index 163140ed9..56c1604a9 100644 --- a/backend/crates/gateway-plugin/runtime/tests/generation/maintenance.rs +++ b/backend/crates/gateway-plugin/runtime/tests/generation/maintenance.rs @@ -9,6 +9,7 @@ use gateway_admin::{ }, ports::plugin_resources::PluginResourceAccess, }; +use gateway_core::task::WorkerDaemonContext; use gateway_core::{ account::FastMode, lifecycle::CancellationToken, @@ -165,7 +166,9 @@ async fn start( let stop = CancellationToken::new(); let cancellation = stop.clone(); let task = tokio::spawn(async move { - task.run(cancellation).await.unwrap(); + task.run(WorkerDaemonContext::new(cancellation)) + .await + .unwrap(); }); (runtime, core, resources, stop, task) } diff --git a/backend/crates/gateway-store/src/postgres/client_keys.rs b/backend/crates/gateway-store/src/postgres/client_keys.rs index 4fd6be6fd..252c254f9 100644 --- a/backend/crates/gateway-store/src/postgres/client_keys.rs +++ b/backend/crates/gateway-store/src/postgres/client_keys.rs @@ -31,11 +31,10 @@ use gateway_core::{ account::OpaqueProviderData, engine::budget::{ClientBudgetLimits, ClientBudgetStatus}, engine::execution::ClientApiKeyUsageSink, - lifecycle::CancellationToken, metering::Decimal, policy::{ClientApiKeyId, PlaintextClientApiKey, RateLimits}, routing::{AccountGroupId, ProviderKind}, - task::{DaemonTask, WorkerTaskError}, + task::{DaemonTask, WorkerDaemonContext, WorkerTaskError}, }; use serde::Deserialize; use sqlx::{PgPool, Postgres, QueryBuilder, Transaction}; @@ -626,8 +625,9 @@ impl ClientApiKeyUsageSink for PgClientApiKeyUsageSink { } impl DaemonTask for PgClientApiKeyUsageWriter { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); loop { tokio::select! { () = cancellation.cancelled() => { @@ -649,6 +649,7 @@ impl DaemonTask for PgClientApiKeyUsageWriter { "client API key last-used flush failed", )); } + context.succeeded(); } }) } diff --git a/backend/crates/gateway-store/src/postgres/execution_buffer.rs b/backend/crates/gateway-store/src/postgres/execution_buffer.rs index ec816e4c8..3d2ff596e 100644 --- a/backend/crates/gateway-store/src/postgres/execution_buffer.rs +++ b/backend/crates/gateway-store/src/postgres/execution_buffer.rs @@ -14,8 +14,7 @@ use gateway_core::engine::{ ModelRequestId, NewModelRequest, ProbeFailure, RecoveryReport, }; use gateway_core::error::{ProviderError, StoreError}; -use gateway_core::lifecycle::CancellationToken; -use gateway_core::task::{DaemonTask, WorkerTaskError}; +use gateway_core::task::{DaemonTask, WorkerDaemonContext, WorkerTaskError}; use gateway_core::upstream::UpstreamSendState; use tokio::sync::{Mutex, Notify, mpsc}; @@ -486,7 +485,7 @@ where { fn run( &self, - cancellation: CancellationToken, + context: WorkerDaemonContext, ) -> futures::future::BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { let shutdown_deadline = OnceLock::new(); @@ -496,7 +495,7 @@ where receiver, self.inner.as_ref(), self.state.as_ref(), - cancellation.clone(), + context.clone(), &shutdown_deadline, ) })) @@ -511,12 +510,13 @@ async fn run_lane( receiver: &Mutex>, store: &S, state: &ExecutionBufferState, - cancellation: CancellationToken, + context: WorkerDaemonContext, shutdown_deadline: &OnceLock, ) -> Result<(), WorkerTaskError> where S: ExecutionStore + OperationalDiagnostics + ?Sized, { + let cancellation = context.cancellation(); let mut receiver = receiver.lock().await; loop { let queued = tokio::select! { @@ -564,7 +564,11 @@ where drain_on_shutdown(lane, &mut receiver, store, state, deadline).await; return Ok(()); }, - () = &mut persistence => {}, + succeeded = &mut persistence => { + if succeeded { + context.succeeded(); + } + }, } } } @@ -573,7 +577,7 @@ fn shared_shutdown_deadline(deadline: &OnceLock) -> Instant { *deadline.get_or_init(|| Instant::now() + SHUTDOWN_DRAIN_TIMEOUT) } -async fn persist_queued(mut queued: QueuedExecutionObservation, store: &S) +async fn persist_queued(mut queued: QueuedExecutionObservation, store: &S) -> bool where S: ExecutionStore + OperationalDiagnostics + ?Sized, { @@ -587,18 +591,21 @@ where request_id = ?request_id, "执行观测队列内部状态无效，已丢弃本次写入" ); - return; + return false; }; // Store 写入没有统一幂等键；超时可能表示已提交，不能在这里盲目重试并 // 制造重复 ops_events // 失败会被计数并丢弃，数据面始终不等待补偿 match write.persist(store).await { - Ok(()) => state.record_persisted(), + Ok(()) => { + state.record_persisted(); + true + } Err(error) => { state.record_write_failure(); let stats = state.snapshot(); if !stats.write_failure_total.is_power_of_two() { - return; + return false; } tracing::warn!( operation, @@ -607,6 +614,7 @@ where write_failure_total = stats.write_failure_total, "执行观测写入失败，数据面不受影响" ); + false } } } diff --git a/backend/crates/gateway-store/src/redis/coordination_buffer.rs b/backend/crates/gateway-store/src/redis/coordination_buffer.rs index 18f818342..0eacaceaf 100644 --- a/backend/crates/gateway-store/src/redis/coordination_buffer.rs +++ b/backend/crates/gateway-store/src/redis/coordination_buffer.rs @@ -17,7 +17,7 @@ use gateway_core::engine::admission::{ }; use gateway_core::lifecycle::CancellationToken; use gateway_core::policy::ClientApiKeyId; -use gateway_core::task::{DaemonTask, WorkerTaskError}; +use gateway_core::task::{DaemonTask, WorkerDaemonContext, WorkerTaskError}; use tokio::sync::{Mutex, mpsc}; const DEFAULT_QUEUE_CAPACITY: usize = 4_096; @@ -139,8 +139,9 @@ pub struct ClientAdmissionReleaseWriter { } impl DaemonTask for ClientAdmissionReleaseWriter { - fn run(&self, cancellation: CancellationToken) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); let mut receiver = self.receiver.lock().await; loop { let release = tokio::select! { @@ -155,7 +156,10 @@ impl DaemonTask for ClientAdmissionReleaseWriter { "client admission release queue closed", )); }; - release_admission(self.inner.as_ref(), self.diagnostics.as_ref(), release).await; + if release_admission(self.inner.as_ref(), self.diagnostics.as_ref(), release).await + { + context.succeeded(); + } } }) } @@ -185,7 +189,7 @@ async fn release_admission( port: &dyn ClientAdmissionPort, diagnostics: &dyn gateway_core::diagnostics::OperationalDiagnostics, release: AdmissionRelease, -) { +) -> bool { if let Err(source) = port .release(&release.client_api_key_id, &release.model_request_id) .await @@ -205,5 +209,7 @@ async fn release_admission( "admission diagnostic could not be recorded" ); } + return false; } + true } diff --git a/backend/crates/gateway-store/src/workers.rs b/backend/crates/gateway-store/src/workers.rs index 489735712..b33ccba58 100644 --- a/backend/crates/gateway-store/src/workers.rs +++ b/backend/crates/gateway-store/src/workers.rs @@ -2,6 +2,7 @@ use super::*; use gateway_core::task::DaemonTask; +use gateway_core::task::WorkerDaemonContext; use std::sync::atomic::{AtomicU32, Ordering}; const COMMAND_DRAIN_TIMEOUT: Duration = Duration::from_secs(4); @@ -52,7 +53,7 @@ fn spawn_command_writer( where T: DaemonTask + Send + 'static, { - tokio::spawn(async move { writer.run(cancellation).await }) + tokio::spawn(async move { writer.run(WorkerDaemonContext::new(cancellation)).await }) } impl CommandStoreDrain { diff --git a/backend/crates/gateway-store/tests/postgres/client_keys.rs b/backend/crates/gateway-store/tests/postgres/client_keys.rs index 687278983..3d44c77e0 100644 --- a/backend/crates/gateway-store/tests/postgres/client_keys.rs +++ b/backend/crates/gateway-store/tests/postgres/client_keys.rs @@ -12,8 +12,10 @@ use gateway_admin::{ ports::store::ClientKeyStore as _, }; use gateway_core::{ - engine::execution::ClientApiKeyUsageSink as _, lifecycle::CancellationToken, - policy::ClientApiKeyId, task::DaemonTask as _, + engine::execution::ClientApiKeyUsageSink as _, + lifecycle::CancellationToken, + policy::ClientApiKeyId, + task::{DaemonTask as _, WorkerDaemonContext}, }; use gateway_store::postgres::{ ClientApiKeyCursor, ClientApiKeyCursorValue, ClientApiKeyListQuery, ClientApiKeyRepository, @@ -592,7 +594,7 @@ async fn client_key_usage_daemon_should_flush_pending_touch_on_shutdown() { let task = { let writer = Arc::clone(&writer); let cancellation = cancellation.clone(); - tokio::spawn(async move { writer.run(cancellation).await }) + tokio::spawn(async move { writer.run(WorkerDaemonContext::new(cancellation)).await }) }; sink.record_used(&ClientApiKeyId::new("key_usage_daemon").expect("valid client API Key ID")); cancellation.cancel(); diff --git a/backend/crates/gateway-store/tests/postgres/execution_buffer.rs b/backend/crates/gateway-store/tests/postgres/execution_buffer.rs index c697a2496..2dec2f372 100644 --- a/backend/crates/gateway-store/tests/postgres/execution_buffer.rs +++ b/backend/crates/gateway-store/tests/postgres/execution_buffer.rs @@ -16,7 +16,7 @@ use gateway_core::error::{ }; use gateway_core::lifecycle::CancellationToken; use gateway_core::routing::{ProviderKind, UpstreamModelId}; -use gateway_core::task::DaemonTask as _; +use gateway_core::task::{DaemonTask as _, WorkerDaemonContext}; use gateway_core::upstream::UpstreamSendState; use gateway_store::postgres::{BufferedExecutionStore, PgExecutionStore}; use tokio::sync::{Notify, Semaphore}; @@ -288,7 +288,7 @@ async fn observation_writer_persists_commands_in_enqueue_order() { let task = tokio::spawn({ let writer = Arc::clone(&writer); let cancellation = cancellation.clone(); - async move { writer.run(cancellation).await } + async move { writer.run(WorkerDaemonContext::new(cancellation)).await } }); tokio::time::timeout(Duration::from_secs(1), async { loop { @@ -338,7 +338,7 @@ async fn request_affine_lanes_run_independently_and_preserve_each_request_order( let task = tokio::spawn({ let writer = Arc::clone(&writer); let cancellation = cancellation.clone(); - async move { writer.run(cancellation).await } + async move { writer.run(WorkerDaemonContext::new(cancellation)).await } }); inner.wait_for_started(2).await; assert_eq!(inner.maximum_active.load(Ordering::Acquire), 2); @@ -383,7 +383,7 @@ async fn idle_wait_includes_a_dequeued_write_until_persistence_finishes() { let task = tokio::spawn({ let writer = Arc::clone(&writer); let cancellation = cancellation.clone(); - async move { writer.run(cancellation).await } + async move { writer.run(WorkerDaemonContext::new(cancellation)).await } }); inner.wait_for_started(1).await; @@ -430,7 +430,7 @@ async fn global_item_budget_includes_writes_running_in_every_lane() { let task = tokio::spawn({ let writer = Arc::clone(&writer); let cancellation = cancellation.clone(); - async move { writer.run(cancellation).await } + async move { writer.run(WorkerDaemonContext::new(cancellation)).await } }); inner.wait_for_started(2).await; assert_eq!(store.stats().queued_items, 2); @@ -477,7 +477,7 @@ async fn all_lanes_share_one_shutdown_deadline_for_inflight_writes() { let task = tokio::spawn({ let writer = Arc::clone(&writer); let cancellation = cancellation.clone(); - async move { writer.run(cancellation).await } + async move { writer.run(WorkerDaemonContext::new(cancellation)).await } }); inner.wait_for_started(4).await; cancellation.cancel(); @@ -512,7 +512,7 @@ async fn observation_writer_continues_after_a_database_write_failure() { let cancellation = CancellationToken::new(); cancellation.cancel(); writer - .run(cancellation) + .run(WorkerDaemonContext::new(cancellation)) .await .expect("shutdown drains observations"); @@ -545,7 +545,7 @@ async fn shutdown_drains_observations_already_accepted_by_the_queue() { let cancellation = CancellationToken::new(); cancellation.cancel(); writer - .run(cancellation) + .run(WorkerDaemonContext::new(cancellation)) .await .expect("shutdown drains observations"); @@ -653,7 +653,7 @@ async fn zero_attempt_create_and_finalize_drain_in_order_with_trace() { let cancellation = CancellationToken::new(); cancellation.cancel(); writer - .run(cancellation) + .run(WorkerDaemonContext::new(cancellation)) .await .expect("shutdown drains accepted commands"); let stats = store.stats(); @@ -703,7 +703,7 @@ async fn request_affine_lanes_persist_a_postgres_lifecycle_burst_without_orphans let task = tokio::spawn({ let writer = Arc::clone(&writer); let cancellation = cancellation.clone(); - async move { writer.run(cancellation).await } + async move { writer.run(WorkerDaemonContext::new(cancellation)).await } }); tokio::time::timeout(Duration::from_secs(15), async { while store.stats().persisted_total != WRITES { @@ -760,7 +760,10 @@ async fn zero_attempt_dropped_finalize_is_fail_open_and_recoverable_at_deadline( assert_eq!(store.stats().dropped_total, 1); let cancellation = CancellationToken::new(); cancellation.cancel(); - writer.run(cancellation).await.expect("drain create only"); + writer + .run(WorkerDaemonContext::new(cancellation)) + .await + .expect("drain create only"); let repository = observability_repository(&database.pool); let pending = repository .usage_record_detail(request.id.as_str()) @@ -842,7 +845,7 @@ async fn zero_attempt_dropped_create_does_not_make_finalize_an_upsert() { let cancellation = CancellationToken::new(); cancellation.cancel(); writer - .run(cancellation) + .run(WorkerDaemonContext::new(cancellation)) .await .expect("drain orphan finalization"); assert_eq!(store.stats().persisted_total, 0); @@ -913,7 +916,7 @@ async fn zero_attempt_postgres_write_failures_are_not_retried_and_do_not_stop_th let cancellation = CancellationToken::new(); cancellation.cancel(); writer - .run(cancellation) + .run(WorkerDaemonContext::new(cancellation)) .await .expect("continue draining after PG errors"); let stats = store.stats(); @@ -1023,7 +1026,10 @@ async fn operational_failure_uses_the_bounded_queue_without_creating_a_model_req assert_eq!(store.stats().dropped_total, 1); let cancellation = CancellationToken::new(); cancellation.cancel(); - writer.run(cancellation).await.unwrap(); + writer + .run(WorkerDaemonContext::new(cancellation)) + .await + .unwrap(); let row: ( Option, String, @@ -1105,7 +1111,10 @@ async fn progress_congestion_preserves_failure_reserve_without_waiting() { assert_eq!(store.stats().dropped_total, 1); let cancelled = CancellationToken::new(); cancelled.cancel(); - writer.run(cancelled).await.unwrap(); + writer + .run(WorkerDaemonContext::new(cancelled)) + .await + .unwrap(); assert_eq!(store.stats().persisted_total, 5); let detail: String = sqlx::query_scalar("select error_details from ops_events where component = 'admin'") diff --git a/backend/crates/gateway-store/tests/redis/coordination_buffer.rs b/backend/crates/gateway-store/tests/redis/coordination_buffer.rs index 121a3b351..9778b93ca 100644 --- a/backend/crates/gateway-store/tests/redis/coordination_buffer.rs +++ b/backend/crates/gateway-store/tests/redis/coordination_buffer.rs @@ -13,6 +13,7 @@ use gateway_core::engine::admission::{ }; use gateway_core::lifecycle::CancellationToken; use gateway_core::policy::{ClientApiKeyId, RateLimits}; +use gateway_core::task::WorkerDaemonContext; use gateway_store::redis::BufferedClientAdmissionPort; #[derive(Default)] @@ -280,5 +281,5 @@ fn spawn_writer( where T: gateway_core::task::DaemonTask + 'static, { - tokio::spawn(async move { writer.run(cancellation).await }) + tokio::spawn(async move { writer.run(WorkerDaemonContext::new(cancellation)).await }) } diff --git a/backend/crates/providers/openai/src/provider/workers.rs b/backend/crates/providers/openai/src/provider/workers.rs index c48342aec..99cc94850 100644 --- a/backend/crates/providers/openai/src/provider/workers.rs +++ b/backend/crates/providers/openai/src/provider/workers.rs @@ -3,6 +3,7 @@ use super::*; use crate::transport::profile::cli_release::CliReleaseService; use crate::transport::profile::platform_release::PlatformDesktopReleaseService; +use gateway_core::task::WorkerDaemonContext; pub(crate) struct ClientReleaseServices { pub desktop: Arc, @@ -299,11 +300,9 @@ impl ScheduledTask for OpenAiCatalogTask { } impl DaemonTask for OpenAiCatalogEtagTask { - fn run( - &self, - cancellation: gateway_core::lifecycle::CancellationToken, - ) -> BoxFuture<'_, Result<(), WorkerTaskError>> { + fn run(&self, context: WorkerDaemonContext) -> BoxFuture<'_, Result<(), WorkerTaskError>> { Box::pin(async move { + let cancellation = context.cancellation(); loop { tokio::select! { () = cancellation.cancelled() => return Ok(()), @@ -314,6 +313,8 @@ impl DaemonTask for OpenAiCatalogEtagTask { error = %error, "OpenAI model catalog ETag refresh failed" ); + } else { + context.succeeded(); } } }) diff --git a/docs/architecture.md b/docs/architecture.md index e73769202..fa774e525 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -1022,6 +1022,11 @@ Worker 由各 Bundle 贡献、由 Host 统一监督： 以及账号冻结恢复 worker（容量熔断的自适应并发下调与到期探测解冻） - Provider：credential refresh、quota/catalog 健康、账号预热和官方版本/etag 检查 +Daemon 每次运行持有独立的取消与成功进度上下文，实际工作或依赖检查成功后由 owner 上报。 +Host 据此清除连续失败与当前错误、重置重启退避，并保持 `Running`；仅重启不会恢复健康， +已退出运行的通知不影响新运行。周期任务按调度间隔检查成功新鲜度，daemon 的空闲或长任务 +不受重启退避时长限制，运行中的依赖健康由相应探针和业务 owner 负责 + 预热与备份按部署时区解释本地时刻，缺失时刻跳过、重复时刻取较早一次。 预热由 Store 在运行设置记录中持久化 UTC 执行游标，只领取晚于游标的时刻，重启或回拨不重复执行历史时段。 游标独立于管理配置的替换与版本发布；取消仍沿既有 worker lease 生命周期释放。