test(ci): guard watchdog alert wiring

.github/actions/schedule-failure-issue/action.yml

@@ -14,9 +14,10 @@
 
 name: "Schedule Failure Issue"
 description: >-
-  Open (or update) a tracking issue when a scheduled workflow run fails.
+  Open (or update) a tracking issue when a scheduled workflow run fails or
+  does not complete normally.
   Dedupes by workflow name: if an open issue titled
-  "[scheduled-failure] <workflow name>" already exists, the failure is
+  "[scheduled-failure] <workflow name>" already exists, the result is
   appended as a comment; otherwise a new issue is created. This is the
   single alerting mechanism for all scheduled pipelines (backlog#1149 ci-8).
 
@@ -38,6 +39,26 @@ inputs:
       Set to an empty string to skip labeling.
     required: false
     default: "infrastructure"
+  source-run-id:
+    description: "Run ID to report. Defaults to the current workflow run."
+    required: false
+    default: ${{ github.run_id }}
+  source-run-attempt:
+    description: "Run attempt to report. Defaults to the current attempt."
+    required: false
+    default: ${{ github.run_attempt }}
+  source-event:
+    description: "Trigger event of the run being reported."
+    required: false
+    default: ${{ github.event_name }}
+  source-ref-name:
+    description: "Ref name of the run being reported."
+    required: false
+    default: ${{ github.ref_name }}
+  source-sha:
+    description: "Commit SHA of the run being reported."
+    required: false
+    default: ${{ github.sha }}
 
 runs:
   using: "composite"
@@ -48,17 +69,21 @@ runs:
         GH_TOKEN: ${{ inputs.github-token }}
         WORKFLOW_NAME: ${{ inputs.workflow-name }}
         ISSUE_LABEL: ${{ inputs.label }}
+        SOURCE_RUN_ID: ${{ inputs.source-run-id }}
+        SOURCE_RUN_ATTEMPT: ${{ inputs.source-run-attempt }}
+        SOURCE_EVENT: ${{ inputs.source-event }}
+        SOURCE_REF_NAME: ${{ inputs.source-ref-name }}
+        SOURCE_SHA: ${{ inputs.source-sha }}
       run: |
         set -euo pipefail
 
         title="[scheduled-failure] ${WORKFLOW_NAME}"
-        run_url="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
+        run_url="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${SOURCE_RUN_ID}"
 
-        # Failed job names for this run attempt. The alert job runs while the
-        # run as a whole is still in progress, so inspect the jobs that have
-        # already completed with a non-success conclusion.
+        # Inspect the reported run attempt. It can be the current in-workflow
+        # failure or a completed run observed by the external watchdog.
         failed_jobs="$(gh api \
-          "repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/attempts/${GITHUB_RUN_ATTEMPT}/jobs" \
+          "repos/${GITHUB_REPOSITORY}/actions/runs/${SOURCE_RUN_ID}/attempts/${SOURCE_RUN_ATTEMPT}/jobs" \
           --paginate \
           --jq '.jobs[]
                 | select(.conclusion == "failure" or .conclusion == "timed_out" or .conclusion == "cancelled")
@@ -68,13 +93,13 @@ runs:
         fi
 
         body="$(cat <<EOF
-        Scheduled run of **${WORKFLOW_NAME}** failed.
+        Scheduled run of **${WORKFLOW_NAME}** did not complete successfully.
 
-        - Run: ${run_url} (attempt ${GITHUB_RUN_ATTEMPT})
-        - Event: \`${GITHUB_EVENT_NAME}\`
-        - Ref: \`${GITHUB_REF_NAME}\` @ \`${GITHUB_SHA}\`
+        - Run: ${run_url} (attempt ${SOURCE_RUN_ATTEMPT})
+        - Event: \`${SOURCE_EVENT}\`
+        - Ref: \`${SOURCE_REF_NAME}\` @ \`${SOURCE_SHA}\`
 
-        Failed jobs:
+        Non-success jobs:
         ${failed_jobs}
         EOF
         )"

.github/workflows/build.yml

@@ -1032,3 +1032,23 @@ jobs:
 
           echo "🎉 Released $TAG successfully!"
           echo "📄 Release URL: ${{ needs.create-release.outputs.release_url }}"
+
+  alert-on-failure:
+    name: Alert on scheduled failure
+    needs: [build-check, prepare-platform-matrix, build-rustfs, build-summary]
+    if: >-
+      always() && github.event_name == 'schedule' &&
+      (contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Open or update failure-tracking issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/ci.yml

@@ -1032,3 +1032,37 @@ jobs:
           path: artifacts/s3tests-single/**
           if-no-files-found: ignore
           retention-days: 3
+
+  alert-on-failure:
+    name: Alert on scheduled failure
+    needs:
+      - typos
+      - quick-checks
+      - test-and-lint
+      - test-ilm-integration-serial
+      - test-and-lint-rio-v2
+      - test-and-lint-protocols
+      - build-rustfs-debug-binary
+      - build-rustfs-debug-binary-rio-v2
+      - uring-integration
+      - e2e-tests
+      - e2e-full
+      - e2e-tests-rio-v2
+      - s3-implemented-tests
+      - s3-lifecycle-behavior-tests
+    if: >-
+      always() && github.event_name == 'schedule' &&
+      (contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Open or update failure-tracking issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/minio-interop.yml

@@ -121,3 +121,21 @@ jobs:
           cargo nextest run --run-ignored ignored-only --no-tests=fail \
             -p "$INTEROP_PACKAGE" --features "$INTEROP_FEATURES" \
             -E "$INTEROP_FILTER"
+
+  alert-on-failure:
+    name: Alert on scheduled failure
+    needs: [minio-interop]
+    if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Open or update failure-tracking issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/nightly-gnu.yml

@@ -194,3 +194,23 @@ jobs:
 
       - name: Run HA leader failover live checks (three-node Raft cluster in Docker)
         run: bash scripts/test/vault_ha_kms_live.sh
+
+  alert-on-failure:
+    name: Alert on scheduled failure
+    needs: [build, kms-vault-lane, kms-vault-ha-failover]
+    if: >-
+      always() && github.event_name == 'schedule' &&
+      (contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Open or update failure-tracking issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/scheduled-validation-watchdog.yml

@@ -0,0 +1,63 @@
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+name: Scheduled Validation Watchdog
+
+on:
+  workflow_run:
+    workflows:
+      - "Security Audit"
+      - "Build and Release"
+      - "Continuous Integration"
+      - "coverage"
+      - "e2e-nightly"
+      - "e2e-s3tests"
+      - "Fuzz"
+      - "mint"
+      - "minio-interop"
+      - "Nightly GNU Build"
+      - "Performance A/B"
+      - "Runner Hygiene"
+    types: [completed]
+
+permissions:
+  contents: read
+
+jobs:
+  alert-on-incomplete-run:
+    name: Alert on incomplete scheduled run
+    if: >-
+      github.event.workflow_run.event == 'schedule' &&
+      github.event.workflow_run.conclusion != 'success' &&
+      github.event.workflow_run.conclusion != 'failure'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      actions: read
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+      - name: Open or update incomplete-run issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.GITHUB_TOKEN }}
+          workflow-name: ${{ github.event.workflow_run.name }}
+          source-run-id: ${{ github.event.workflow_run.id }}
+          source-run-attempt: ${{ github.event.workflow_run.run_attempt }}
+          source-event: ${{ github.event.workflow_run.event }}
+          source-ref-name: ${{ github.event.workflow_run.head_branch }}
+          source-sha: ${{ github.event.workflow_run.head_sha }}

crates/config/src/constants/runtime.rs

@@ -57,13 +57,6 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
 pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
 pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
 
-/// Dedicated blocking thread pool for fsync/fdatasync operations.
-/// When > 1, fsync operations are isolated from the main blocking pool to
-/// prevent device-bound fsync from starving read operations (pread/stat/open).
-/// Default 0 means auto (no isolation, use main runtime).
-pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
-pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
-
 // Dial9 Tokio Telemetry Default values
 pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
 pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";

crates/ecstore/src/disk/os.rs

@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
     #[cfg(unix)]
     {
         let dir = dir.as_ref().to_path_buf();
-        fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
+        tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
     }
 
     #[cfg(not(unix))]
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
     #[cfg(test)]
     let dir = group.dir.clone();
     let dir_file = group.dir_file.clone();
-    fsync_spawn_blocking(move || {
+    tokio::task::spawn_blocking(move || {
         #[cfg(test)]
         {
             if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -1080,44 +1080,6 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
 
 static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
 static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
-
-/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
-/// configured with >1 threads, isolates device-bound fsync from the main
-/// blocking pool so reads (pread/stat/open) are not starved. `None` means
-/// fall back to the main runtime (zero behavior change).
-static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
-    let threads =
-        rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
-    if threads <= 1 {
-        return None;
-    }
-    let mut builder = tokio::runtime::Builder::new_multi_thread();
-    builder
-        .worker_threads(num_cpus::get().min(8))
-        .max_blocking_threads(threads)
-        .thread_name("rustfs-fsync")
-        .thread_stack_size(512 * 1024)
-        .enable_all();
-    match builder.build() {
-        Ok(rt) => {
-            tracing::info!(threads, "fsync dedicated blocking pool enabled");
-            Some(rt)
-        }
-        Err(err) => {
-            tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
-            None
-        }
-    }
-});
-
-/// Spawn a blocking task on the fsync-dedicated runtime if configured,
-/// otherwise fall back to the main tokio blocking pool.
-fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
-    match FSYNC_RUNTIME.as_ref() {
-        Some(rt) => rt.spawn_blocking(f),
-        None => tokio::task::spawn_blocking(f),
-    }
-}
 static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
 type NamespaceMutationLock = AsyncMutex<()>;
@@ -1255,7 +1217,7 @@ where
     F: FnOnce() -> io::Result<T> + Send + 'static,
 {
     let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
-    let result = fsync_spawn_blocking(move || {
+    let result = tokio::task::spawn_blocking(move || {
         let _disk_permit = disk_permit;
         work()
     })
@@ -2184,7 +2146,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
         wait_started,
     );
     let disk_permit = admission.disk_permit.clone();
-    let result = fsync_spawn_blocking(move || {
+    let result = tokio::task::spawn_blocking(move || {
         let _lease = lease;
         let _disk_permit = disk_permit;
         operation()

crates/heal/src/heal/manager/tests.rs

@@ -110,10 +110,7 @@ impl HealStorageAPI for MockStorage {
         Ok(Vec::new())
     }
 
-    async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
-        if bucket == "panic" {
-            panic!("test-only panic payload must not escape the scheduler");
-        }
+    async fn get_bucket_info(&self, _bucket: &str) -> Result<Option<BucketInfo>> {
         Ok(None)
     }
 
@@ -1024,231 +1021,6 @@ async fn test_task_alias_is_removed_after_terminal_completion() {
     assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
 }
 
-#[tokio::test]
-#[serial_test::serial]
-async fn scheduler_panic_releases_active_slot_and_allows_same_target_readmission() {
-    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
-    let manager = HealManager::new(storage, None);
-    let request = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
-    let task_id = request.id.clone();
-
-    assert_eq!(
-        manager
-            .submit_heal_request(request)
-            .await
-            .expect("panic request should be admitted"),
-        HealAdmissionResult::Accepted
-    );
-    let duplicate = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
-    let duplicate_id = duplicate.id.clone();
-    assert_eq!(
-        manager
-            .submit_heal_request(duplicate)
-            .await
-            .expect("same target should merge while active is queued"),
-        HealAdmissionResult::Merged
-    );
-    assert_eq!(manager.canonical_task_id(&duplicate_id).await, task_id);
-    process_manager_queue_once(&manager).await;
-
-    let status = tokio::time::timeout(Duration::from_secs(1), async {
-        loop {
-            if let Ok(status) = manager.get_task_status(&task_id).await
-                && matches!(status, HealTaskStatus::Failed { .. })
-            {
-                break status;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("panic task should reach a terminal status");
-    assert_eq!(
-        status,
-        HealTaskStatus::Failed {
-            error: PANICKED_HEAL_TASK_ERROR.to_string()
-        }
-    );
-    assert_eq!(manager.get_active_task_count().await, 0);
-    assert_eq!(manager.get_queue_length().await, 0);
-    assert!(manager.retrying_heals.lock().await.is_empty());
-    assert!(manager.task_aliases.lock().await.is_empty());
-    assert!(manager.completed_heals.lock().await.contains_key(&task_id));
-    assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
-
-    let readmitted = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
-    assert_eq!(
-        manager
-            .submit_heal_request(readmitted)
-            .await
-            .expect("same target should be re-admitted after a panic"),
-        HealAdmissionResult::Accepted
-    );
-}
-
-#[tokio::test]
-#[serial_test::serial]
-async fn retry_child_panic_finishes_parent_once() {
-    clear_scheduler_panic();
-    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
-    let manager = HealManager::new(storage, None);
-    let request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
-    let task_id = request.id.clone();
-    assert_eq!(
-        manager
-            .submit_heal_request(request)
-            .await
-            .expect("retry request should be admitted"),
-        HealAdmissionResult::Accepted
-    );
-    arm_scheduler_panic(SchedulerPanicPoint::RetryChild, &task_id);
-    process_manager_queue_once(&manager).await;
-
-    let status = tokio::time::timeout(Duration::from_secs(1), async {
-        loop {
-            if let Ok(status) = manager.get_task_status(&task_id).await
-                && matches!(status, HealTaskStatus::Failed { .. })
-            {
-                break status;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("retry child panic should finish the parent");
-    clear_scheduler_panic();
-    assert_eq!(
-        status,
-        HealTaskStatus::Failed {
-            error: PANICKED_HEAL_TASK_ERROR.to_string()
-        }
-    );
-    assert_eq!(manager.get_active_task_count().await, 0);
-    assert_eq!(manager.get_queue_length().await, 0);
-    assert!(manager.retrying_heals.lock().await.is_empty());
-    assert!(manager.task_aliases.lock().await.is_empty());
-    assert_eq!(manager.completed_heals.lock().await.len(), 1);
-    assert_eq!(manager.get_statistics().await.failed_tasks, 1);
-}
-
-#[tokio::test]
-#[serial_test::serial]
-async fn cleanup_panic_is_supervised() {
-    clear_scheduler_panic();
-    let notice_bucket = "cleanup-panic-mrf";
-    let notice_object = "object";
-    let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(notice_bucket);
-    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
-    let manager = HealManager::new(storage, None);
-    let mut request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::Normal);
-    request.source = HealRequestSource::Admin;
-    let task_id = request.id.clone();
-    assert_eq!(
-        manager
-            .submit_heal_request(request)
-            .await
-            .expect("cleanup request should be admitted"),
-        HealAdmissionResult::Accepted
-    );
-    manager
-        .mrf_repair_notice_targets
-        .lock()
-        .expect("mrf repair notice registry poisoned")
-        .insert(
-            task_id.clone(),
-            vec![MrfRepairNoticeTarget {
-                bucket: Arc::from(notice_bucket),
-                object: Arc::from(notice_object),
-                version_id: None,
-            }],
-        );
-    arm_scheduler_panic(SchedulerPanicPoint::Cleanup, &task_id);
-    process_manager_queue_once(&manager).await;
-
-    let status = tokio::time::timeout(Duration::from_secs(1), async {
-        loop {
-            if let Ok(status) = manager.get_task_status(&task_id).await
-                && matches!(status, HealTaskStatus::Completed)
-            {
-                break status;
-            }
-            tokio::time::sleep(Duration::from_millis(10)).await;
-        }
-    })
-    .await
-    .expect("cleanup panic should leave a terminal status");
-    clear_scheduler_panic();
-    assert_eq!(status, HealTaskStatus::Completed);
-    assert_eq!(manager.get_active_task_count().await, 0);
-    assert!(manager.task_aliases.lock().await.is_empty());
-    assert_eq!(manager.completed_heals.lock().await.len(), 1);
-    assert_eq!(manager.get_statistics().await.successful_tasks, 1);
-    let events = rustfs_common::mrf_channel::take_mrf_repaired_events_for(notice_bucket);
-    assert_eq!(events.len(), 1, "cleanup panic must preserve successful MRF notice delivery");
-    assert_eq!(events[0].object.as_ref(), notice_object);
-}
-
-#[tokio::test]
-#[serial_test::serial]
-async fn cancelled_retry_child_panic_does_not_rearchive_failed_status() {
-    let manager = HealManager::new(Arc::new(MockStorage), None);
-    let request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
-    let task_id = request.id.clone();
-    let retry_cancel_token = insert_retrying_request(&manager, request.clone()).await;
-
-    manager
-        .cancel_task(&task_id)
-        .await
-        .expect("retry cancellation should succeed");
-    assert!(retry_cancel_token.is_cancelled());
-
-    let state = PanicCleanupState {
-        active_heals: manager.active_heals.clone(),
-        heal_queue: manager.heal_queue.clone(),
-        completed_heals: manager.completed_heals.clone(),
-        task_aliases: manager.task_aliases.clone(),
-        retrying_heals: manager.retrying_heals.clone(),
-        mrf_repair_notice_targets: manager.mrf_repair_notice_targets.clone(),
-        replacement_recovery_anchors: manager.replacement_recovery_anchors.clone(),
-        statistics: manager.statistics.clone(),
-    };
-    finish_panicked_retry_child(task_id.clone(), request.heal_type, retry_cancel_token, state).await;
-
-    assert!(manager.retrying_heals.lock().await.is_empty());
-    assert!(manager.completed_heals.lock().await.is_empty());
-    assert_eq!(manager.get_statistics().await.failed_tasks, 0);
-}
-
-#[tokio::test]
-async fn active_cancel_wins_parent_panic_cleanup_without_completed_status() {
-    let manager = HealManager::new(Arc::new(MockStorage), None);
-    let request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::Normal);
-    let task_id = request.id.clone();
-    let task = Arc::new(HealTask::from_request(request, Arc::new(MockStorage)));
-    manager.active_heals.lock().await.insert(task_id.clone(), task.clone());
-
-    manager
-        .cancel_task(&task_id)
-        .await
-        .expect("active task cancellation should win");
-    assert_eq!(task.get_status().await, HealTaskStatus::Cancelled);
-
-    let state = PanicCleanupState {
-        active_heals: manager.active_heals.clone(),
-        heal_queue: manager.heal_queue.clone(),
-        completed_heals: manager.completed_heals.clone(),
-        task_aliases: manager.task_aliases.clone(),
-        retrying_heals: manager.retrying_heals.clone(),
-        mrf_repair_notice_targets: manager.mrf_repair_notice_targets.clone(),
-        replacement_recovery_anchors: manager.replacement_recovery_anchors.clone(),
-        statistics: manager.statistics.clone(),
-    };
-    finish_panicked_heal_task(task, task_id, state).await;
-
-    assert!(manager.completed_heals.lock().await.is_empty());
-    assert_eq!(manager.get_statistics().await.failed_tasks, 0);
-}
-
 #[tokio::test]
 async fn test_duplicate_admission_is_atomic_with_queue_to_active_transition() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

scripts/check_test_wiring.py

@@ -15,6 +15,20 @@ from pathlib import Path
 
 
 ROOT = Path(__file__).resolve().parents[1]
+SCHEDULED_ALERT_WORKFLOWS = (
+    ".github/workflows/audit.yml",
+    ".github/workflows/build.yml",
+    ".github/workflows/ci.yml",
+    ".github/workflows/coverage.yml",
+    ".github/workflows/e2e-replication-nightly.yml",
+    ".github/workflows/e2e-s3tests.yml",
+    ".github/workflows/fuzz.yml",
+    ".github/workflows/mint.yml",
+    ".github/workflows/minio-interop.yml",
+    ".github/workflows/nightly-gnu.yml",
+    ".github/workflows/performance-ab.yml",
+    ".github/workflows/runner-hygiene.yml",
+)
 
 
 def words(value: str) -> set[str]:
@@ -252,6 +266,79 @@ def check_profile_definitions(root: Path) -> list[str]:
     return errors
 
 
+def check_scheduled_alerts(root: Path) -> list[str]:
+    errors: list[str] = []
+    for relative in SCHEDULED_ALERT_WORKFLOWS:
+        path = root / relative
+        try:
+            lines = path.read_text().splitlines()
+        except FileNotFoundError:
+            errors.append(f"{relative}: missing scheduled validation workflow")
+            continue
+
+        try:
+            start = lines.index("  alert-on-failure:") + 1
+        except ValueError:
+            errors.append(f"{relative}: missing alert-on-failure job")
+            continue
+        end = next(
+            (index for index in range(start, len(lines)) if re.fullmatch(r"  [A-Za-z0-9_-]+:", lines[index])),
+            len(lines),
+        )
+        job = "\n".join(line.split("#", 1)[0] for line in lines[start:end])
+        required = (
+            "always()",
+            "github.event_name == 'schedule'",
+            "contains(needs.*.result, 'failure')",
+            "issues: write",
+            "uses: ./.github/actions/schedule-failure-issue",
+            "github-token: ${{ secrets.GITHUB_TOKEN }}",
+        )
+        missing = [token for token in required if token not in job]
+        if missing:
+            errors.append(f"{relative}: alert-on-failure missing {', '.join(missing)}")
+
+    watchdog_path = root / ".github/workflows/scheduled-validation-watchdog.yml"
+    try:
+        watchdog = "\n".join(
+            line.split("#", 1)[0] for line in watchdog_path.read_text().splitlines()
+        )
+    except FileNotFoundError:
+        errors.append(".github/workflows/scheduled-validation-watchdog.yml: missing completion watchdog")
+        return errors
+    for relative in SCHEDULED_ALERT_WORKFLOWS:
+        path = root / relative
+        if not path.is_file():
+            continue
+        source = path.read_text()
+        match = re.search(r"^name:\s*[\"']?([^\"'\n]+)", source, re.MULTILINE)
+        if not match:
+            errors.append(f"{relative}: missing workflow name")
+        elif f'- "{match.group(1).strip()}"' not in watchdog:
+            errors.append(f"{relative}: missing from scheduled completion watchdog")
+    required = (
+        "github.event.workflow_run.event == 'schedule'",
+        "github.event.workflow_run.conclusion != 'success'",
+        "github.event.workflow_run.conclusion != 'failure'",
+        "actions: read",
+        "issues: write",
+        "uses: ./.github/actions/schedule-failure-issue",
+        "github-token: ${{ secrets.GITHUB_TOKEN }}",
+        "workflow-name: ${{ github.event.workflow_run.name }}",
+        "source-event: ${{ github.event.workflow_run.event }}",
+        "source-ref-name: ${{ github.event.workflow_run.head_branch }}",
+        "source-sha: ${{ github.event.workflow_run.head_sha }}",
+        "source-run-id: ${{ github.event.workflow_run.id }}",
+        "source-run-attempt: ${{ github.event.workflow_run.run_attempt }}",
+    )
+    missing = [token for token in required if token not in watchdog]
+    if missing:
+        errors.append(
+            ".github/workflows/scheduled-validation-watchdog.yml: missing " + ", ".join(missing)
+        )
+    return errors
+
+
 def check_profile_listing(root: Path, profile: str, listing: Path) -> list[str]:
     try:
         expected_digest = profile_selection(root, profile)
@@ -281,6 +368,7 @@ def validate(root: Path) -> list[str]:
     errors.extend(check_runner_selection(root))
     errors.extend(check_s3_tests_runner(root))
     errors.extend(check_profile_definitions(root))
+    errors.extend(check_scheduled_alerts(root))
     return errors
 
 
@@ -363,6 +451,7 @@ class SelfTests(unittest.TestCase):
                 mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
                 mock.patch(__name__ + ".check_runner_selection", return_value=[]),
                 mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
+                mock.patch(__name__ + ".check_scheduled_alerts", return_value=[]),
             ):
                 self.assertEqual(len(validate(root)), 1)
 
@@ -413,6 +502,75 @@ class SelfTests(unittest.TestCase):
             with mock.patch.object(sys, "platform", "linux"):
                 self.assertEqual(len(check_profile_listing(root, "e2e-full", listing)), 1)
 
+    def test_scheduled_alerts_require_completion_watchdog(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            alert = (
+                "  alert-on-failure:\n"
+                "    if: always() && github.event_name == 'schedule' && "
+                "contains(needs.*.result, 'failure')\n"
+                "    permissions:\n"
+                "      issues: write\n"
+                "    steps:\n"
+                "      - uses: ./.github/actions/schedule-failure-issue\n"
+                "        with:\n"
+                "          github-token: ${{ secrets.GITHUB_TOKEN }}\n"
+            )
+            names: list[str] = []
+            for relative in SCHEDULED_ALERT_WORKFLOWS:
+                path = root / relative
+                path.parent.mkdir(parents=True, exist_ok=True)
+                names.append(path.stem)
+                path.write_text(f'name: "{path.stem}"\n{alert}')
+            watchdog = root / ".github/workflows/scheduled-validation-watchdog.yml"
+            watchdog_source = (
+                "\n".join(f'- "{name}"' for name in names)
+                + "\ngithub.event.workflow_run.event == 'schedule'\n"
+                + "github.event.workflow_run.conclusion != 'success'\n"
+                + "github.event.workflow_run.conclusion != 'failure'\n"
+                + "actions: read\n"
+                + "issues: write\n"
+                + "uses: ./.github/actions/schedule-failure-issue\n"
+                + "github-token: ${{ secrets.GITHUB_TOKEN }}\n"
+                + "workflow-name: ${{ github.event.workflow_run.name }}\n"
+                + "source-event: ${{ github.event.workflow_run.event }}\n"
+                + "source-ref-name: ${{ github.event.workflow_run.head_branch }}\n"
+                + "source-sha: ${{ github.event.workflow_run.head_sha }}\n"
+                + "source-run-id: ${{ github.event.workflow_run.id }}\n"
+                + "source-run-attempt: ${{ github.event.workflow_run.run_attempt }}\n"
+            )
+            watchdog.write_text(watchdog_source)
+            self.assertEqual(check_scheduled_alerts(root), [])
+
+            first = root / SCHEDULED_ALERT_WORKFLOWS[0]
+            mutations = (
+                ("contains(needs.*.result, 'failure')", "false"),
+                ("issues: write", "issues: read"),
+                ("uses: ./.github/actions/schedule-failure-issue", "uses: actions/checkout@v7"),
+                ("github-token: ${{ secrets.GITHUB_TOKEN }}", "github-token: missing"),
+            )
+            for required, replacement in mutations:
+                original = first.read_text()
+                first.write_text(original.replace(required, replacement))
+                self.assertEqual(len(check_scheduled_alerts(root)), 1)
+                first.write_text(original)
+
+            watchdog.write_text(watchdog_source.replace(f'- "{names[0]}"\n', ""))
+            self.assertEqual(len(check_scheduled_alerts(root)), 1)
+
+            watchdog_mutations = (
+                ("actions: read", "actions: none"),
+                ("issues: write", "issues: read"),
+                ("uses: ./.github/actions/schedule-failure-issue", "uses: actions/checkout@v7"),
+                ("github-token: ${{ secrets.GITHUB_TOKEN }}", "github-token: missing"),
+                ("source-event: ${{ github.event.workflow_run.event }}", "source-event: wrong"),
+                ("source-ref-name: ${{ github.event.workflow_run.head_branch }}", "source-ref-name: wrong"),
+                ("source-sha: ${{ github.event.workflow_run.head_sha }}", "source-sha: wrong"),
+            )
+            for required, replacement in watchdog_mutations:
+                watchdog.write_text(watchdog_source.replace(required, replacement))
+                self.assertEqual(len(check_scheduled_alerts(root)), 1)
+
 def main() -> int:
     if sys.argv[1:] == ["--self-test"]:
         suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
@@ -436,7 +594,7 @@ def main() -> int:
         for error in errors:
             print(f"ERROR: {error}", file=sys.stderr)
         return 1
-    print("OK: e2e modules, runner selection, fuzz matrices, profiles, and bounded diagnostics are wired")
+    print("OK: e2e modules, runner selection, fuzz matrices, profiles, scheduled alerts, and bounded diagnostics are wired")
     return 0