Refactor shutdown coordinator with pluggable exit handler and Kani proof

- Add src/exit_handler.rs for pluggable child exit handling
- Refactor shutdown pipeline stages for deterministic ordering
- Add Kani verification model for shutdown coordinator correctness
- Update examples (job, service, sidecar, supervisor, worker) for new API
- Add check-yield-points.sh for yield point audit
- Fix join timeout and shutdown orphan tests for new coordinator

Author: developerworks

clippy.toml

@@ -0,0 +1,11 @@
+# Disallow blocking I/O and synchronous locks in asynchronous contexts.
+# Subtask code must use tokio::fs, tokio::net, tokio::sync::Mutex,
+# and tokio::time::sleep instead.
+disallowed-methods = [
+    "std::fs::read",
+    "std::fs::write",
+    "std::fs::File",
+    "std::net::TcpStream",
+    "std::thread::sleep",
+    "std::sync::Mutex",
+]

examples/job/main.rs

@@ -33,8 +33,13 @@ async fn main() -> ExampleResult {
     // Keep enough event buffer for the shutdown observation sequence.
     spec.event_channel_capacity = 32;
     // Use short shutdown windows so the cleanup path finishes quickly.
-    let shutdown_policy =
-        ShutdownPolicy::new(Duration::from_millis(250), Duration::from_millis(50), true);
+    let shutdown_policy = ShutdownPolicy::new(
+        Duration::from_millis(250),
+        Duration::from_millis(50),
+        true,
+        Duration::from_millis(250),
+        3,
+    );
     // Start the runtime with the job child.
     let handle = Supervisor::start_with_policy(spec, shutdown_policy).await?;
     // Subscribe to lifecycle event text before commands are sent.

examples/service/main.rs

@@ -36,8 +36,13 @@ async fn main() -> ExampleResult {
     // Keep enough event buffer for the full shutdown observation sequence.
     spec.event_channel_capacity = 32;
     // Use short shutdown windows so the example finishes quickly.
-    let shutdown_policy =
-        ShutdownPolicy::new(Duration::from_millis(250), Duration::from_millis(50), true);
+    let shutdown_policy = ShutdownPolicy::new(
+        Duration::from_millis(250),
+        Duration::from_millis(50),
+        true,
+        Duration::from_millis(250),
+        3,
+    );
     // Start the runtime with the service child.
     let handle = Supervisor::start_with_policy(spec, shutdown_policy).await?;
     // Subscribe to lifecycle event text before commands are sent.

examples/shutdown_pipeline_demo.rs

@@ -32,6 +32,8 @@ fn main() {
         Duration::from_secs(5), // graceful_timeout
         Duration::from_secs(1), // abort_wait
         true,                   // abort_after_timeout
+        Duration::from_secs(5), // force_kill_margin
+        3,                      // max_orphan_threshold
     );
 
     println!("  graceful_timeout = {:?}", policy.graceful_timeout);
@@ -92,7 +94,13 @@ fn main() {
     println!("--- ShutdownCoordinator ---");
     println!();
 
-    let coord_policy = ShutdownPolicy::new(Duration::from_secs(5), Duration::from_secs(1), true);
+    let coord_policy = ShutdownPolicy::new(
+        Duration::from_secs(5),
+        Duration::from_secs(1),
+        true,
+        Duration::from_secs(5),
+        3,
+    );
     let mut coordinator = ShutdownCoordinator::new(coord_policy);
 
     let cause = ShutdownCause::new("operator", "scheduled maintenance");

examples/sidecar/main.rs

@@ -37,8 +37,13 @@ async fn main() -> ExampleResult {
     // Keep enough event buffer for the full shutdown observation sequence.
     spec.event_channel_capacity = 64;
     // Use short shutdown windows so the example finishes quickly.
-    let shutdown_policy =
-        ShutdownPolicy::new(Duration::from_millis(250), Duration::from_millis(50), true);
+    let shutdown_policy = ShutdownPolicy::new(
+        Duration::from_millis(250),
+        Duration::from_millis(50),
+        true,
+        Duration::from_millis(250),
+        3,
+    );
     // Start the runtime with the primary service and sidecar.
     let handle = Supervisor::start_with_policy(spec, shutdown_policy).await?;
     // Subscribe to lifecycle event text before commands are sent.

examples/supervisor/main.rs

@@ -36,8 +36,13 @@ async fn main() -> ExampleResult {
     // Keep enough event buffer for the full shutdown observation sequence.
     spec.event_channel_capacity = 32;
     // Use short shutdown windows so the example finishes quickly.
-    let shutdown_policy =
-        ShutdownPolicy::new(Duration::from_millis(250), Duration::from_millis(50), true);
+    let shutdown_policy = ShutdownPolicy::new(
+        Duration::from_millis(250),
+        Duration::from_millis(50),
+        true,
+        Duration::from_millis(250),
+        3,
+    );
     // Start the runtime with the supervisor role child.
     let handle = Supervisor::start_with_policy(spec, shutdown_policy).await?;
     // Subscribe to lifecycle event text before commands are sent.

examples/worker/main.rs

@@ -33,8 +33,13 @@ async fn main() -> ExampleResult {
     // Keep enough event buffer for the shutdown observation sequence.
     spec.event_channel_capacity = 32;
     // Use short shutdown windows so the cleanup path finishes quickly.
-    let shutdown_policy =
-        ShutdownPolicy::new(Duration::from_millis(250), Duration::from_millis(50), true);
+    let shutdown_policy = ShutdownPolicy::new(
+        Duration::from_millis(250),
+        Duration::from_millis(50),
+        true,
+        Duration::from_millis(250),
+        3,
+    );
     // Start the runtime with the worker child.
     let handle = Supervisor::start_with_policy(spec, shutdown_policy).await?;
     // Subscribe to lifecycle event text before commands are sent.

scripts/check-yield-points.sh

@@ -0,0 +1,22 @@
+#!/usr/bin/env bash
+# 检查 src/ 下 Rust 源码中 loop 块是否在 50 行内含有 yield 点
+# 非阻断告警, CI 中仅输出警告信息
+set -uo pipefail
+
+src_dir="src"
+violations=0
+
+while IFS= read -r file; do
+    # 找到 loop { 的行号
+    while IFS=: read -r line_no content; do
+        # 检查接下来 50 行内是否有 yield_now() 或 .await
+        if ! tail -n +"$line_no" "$file" | head -n 50 | grep -q 'yield_now\|\.await'; then
+            echo "WARNING: $file:$line_no: loop without yield point in next 50 lines"
+            ((violations++))
+        fi
+    done < <(grep -n 'loop\s*{' "$file" || true)
+done < <(find "$src_dir" -name '*.rs' -type f)
+
+if [ "$violations" -gt 0 ]; then
+    echo "⚠️  Found $violations loop(s) without yield points (non-blocking)"
+fi

src/exit_handler.rs

@@ -0,0 +1,74 @@
+//! Exit handler abstraction for testable process termination.
+//!
+//! The default implementation calls `std::process::exit(1)` as a last resort
+//! when orphaned tasks exceed the configured threshold. Tests can swap in a
+//! stub that records the exit request instead of terminating the process.
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+/// Abstraction for process exit, allowing tests to observe exit requests
+/// without actually terminating the process.
+pub trait ExitHandler: Send + Sync + std::fmt::Debug {
+    /// Terminates the process (or records the request in test mode).
+    fn exit(&self, code: i32);
+}
+
+/// Default exit handler that calls `std::process::exit(code)`.
+#[derive(Debug, Clone, Copy)]
+pub struct DefaultExitHandler;
+
+impl ExitHandler for DefaultExitHandler {
+    fn exit(&self, code: i32) {
+        std::process::exit(code);
+    }
+}
+
+/// Test-friendly exit handler that records a flag instead of terminating.
+#[derive(Debug, Clone)]
+pub struct TestExitHandler {
+    /// Set to `true` when `exit()` was called.
+    pub called: Arc<AtomicBool>,
+    /// Captured exit code.
+    pub exit_code: Arc<std::sync::Mutex<Option<i32>>>,
+}
+
+impl TestExitHandler {
+    /// Creates a new test exit handler with `called = false`.
+    pub fn new() -> Self {
+        Self {
+            called: Arc::new(AtomicBool::new(false)),
+            exit_code: Arc::new(std::sync::Mutex::new(None)),
+        }
+    }
+
+    /// Returns `true` when `exit()` was called since the last reset.
+    pub fn was_called(&self) -> bool {
+        self.called.load(Ordering::SeqCst)
+    }
+
+    /// Returns the exit code if `exit()` was called.
+    pub fn last_exit_code(&self) -> Option<i32> {
+        *self.exit_code.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
+    /// Resets the recorded state.
+    pub fn reset(&self) {
+        self.called.store(false, Ordering::SeqCst);
+        *self.exit_code.lock().unwrap_or_else(|e| e.into_inner()) = None;
+    }
+}
+
+impl ExitHandler for TestExitHandler {
+    fn exit(&self, code: i32) {
+        self.called.store(true, Ordering::SeqCst);
+        *self.exit_code.lock().unwrap_or_else(|e| e.into_inner()) = Some(code);
+        // Do NOT call std::process::exit — let the test continue.
+    }
+}
+
+impl Default for TestExitHandler {
+    fn default() -> Self {
+        Self::new()
+    }
+}

src/lib.rs

@@ -11,6 +11,7 @@ pub mod control;
 pub mod dashboard;
 pub mod error;
 pub mod event;
+pub mod exit_handler;
 pub mod health;
 pub mod id;
 #[cfg(unix)]