From 7b1ff0da45dced6166b031f1eda2fa3f073fa68e Mon Sep 17 00:00:00 2001 From: overtrue Date: Mon, 26 Oct 2026 12:15:33 +0800 Subject: [PATCH] fix(scanner): retain the enabled supervisor task handle --- crates/scanner/src/scanner.rs | 20 ++++++---- .../src/scanner/tests/recovery_control.rs | 39 +++++++++++++++++++ 2 files changed, 52 insertions(+), 7 deletions(-) diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 0d455bcf6..95493e620 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -1072,10 +1072,9 @@ async fn run_scanner_usage_recovery_intents_for_startup( Ok(attempted) } -/// Start normal scanning when enabled, or one bounded recovery attempt. -/// The disabled branch returns a finite task for the startup owner to join; -/// it never enables ordinary namespace scanning while it replays durable reset -/// intents and cleanup markers. +/// Return the normal scanner supervisor or one finite disabled recovery task. +/// The disabled branch never enables ordinary namespace scanning while it +/// replays durable reset intents and cleanup markers. pub async fn init_scanner_with_recovery( ctx: CancellationToken, storeapi: Arc, @@ -1096,8 +1095,7 @@ pub async fn init_scanner_with_recovery( "Scanner recovery intent startup discovery failed" ); } - init_data_scanner(ctx, storeapi).await; - return None; + return Some(init_data_scanner_task_with_storage(ctx, storeapi).await); } Some(tokio::spawn(async move { if let Err(error) = run_scanner_usage_recovery_intents_for_startup(ctx.clone(), storeapi.clone()).await { @@ -1126,6 +1124,14 @@ pub async fn init_scanner_with_recovery( } async fn init_data_scanner_with_storage(ctx: CancellationToken, storeapi: Arc) +where + S: ScannerStorage, +{ + // Preserve the existing detached-start contract for legacy callers. + drop(init_data_scanner_task_with_storage(ctx, storeapi).await); +} + +async fn init_data_scanner_task_with_storage(ctx: CancellationToken, storeapi: Arc) -> tokio::task::JoinHandle<()> where S: ScannerStorage, { @@ -1233,7 +1239,7 @@ where break; } } - }); + }) } #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] diff --git a/crates/scanner/src/scanner/tests/recovery_control.rs b/crates/scanner/src/scanner/tests/recovery_control.rs index 5f8f806b5..99619177e 100644 --- a/crates/scanner/src/scanner/tests/recovery_control.rs +++ b/crates/scanner/src/scanner/tests/recovery_control.rs @@ -239,6 +239,45 @@ async fn issue_8297_supervisor_recovers_paused_follower_without_restart() { crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } +#[tokio::test] +#[serial] +async fn enabled_scanner_startup_returns_owned_supervisor_handle() { + temp_env::async_with_vars([(ENV_SCANNER_CYCLE, Some("1")), (ENV_SCANNER_START_DELAY_SECS, Some("0"))], async { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); + crate::scanner_io::clear_dirty_usage_buckets_for_tests(); + global_metrics().set_cycle(None).await; + let (_dir, store) = setup_scanner_cycle_store().await; + assert!(!crate::scanner_runtime_initialized(), "test must own the scanner runtime"); + let ctx = CancellationToken::new(); + let _cancel_on_drop = ctx.clone().drop_guard(); + let startup_probe = ScannerStartupObservedProbe::install(); + let mut supervisor = init_scanner_with_recovery(ctx.clone(), store, true) + .await + .expect("enabled startup must return its actual supervisor handle"); + + startup_probe.wait().await; + assert!(crate::scanner_runtime_initialized(), "the observed supervisor owns its runtime guard"); + assert!(!supervisor.is_finished(), "the observed supervisor remains in flight"); + ctx.cancel(); + assert!( + futures::poll!(&mut supervisor).is_pending(), + "cancellation cannot join the supervisor before its observed work is released" + ); + startup_probe.resume(); + drop(startup_probe); + tokio::time::timeout(Duration::from_secs(15), supervisor) + .await + .expect("cancelled supervisor must join within the existing test bound") + .expect("the exact enabled supervisor must exit successfully"); + assert!( + !crate::scanner_runtime_initialized(), + "joining releases the actual supervisor runtime guard" + ); + }) + .await; + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); +} + async fn run_disabled_startup(ctx: CancellationToken, store: Arc) { let initialized_before = crate::scanner_runtime_initialized(); let cleanup = init_scanner_with_recovery(ctx, store, false).await;