Improve cancellation safety and cooperative shutdown for Tokio async code

[henry0715-dev]

Background

crusher still has Tokio async paths where shutdown/reload/reconnect can leave work partially abandoned or state inconsistent.

Key candidates:

• top-level spawned tasks are not always joined to completion
• some paths hold Mutex/RwLock guards across .await
• important shared state (active_policy_list, delete_policy_ids, INGEST_CHANNEL, LAST_TRANSFER_TIME) is directly shared instead of actor-owned
• timestamp persistence is not guaranteed to be safely flushed during shutdown

Goal

Apply cancellation safety to crusher's Tokio async code so that the system has cooperative cancellation, safe drain behavior, and restart-time state consistency.

Scope

1. Identify vulnerable paths

• audit lock-across-await code
• list long-running and detached spawned tasks
• identify work that can be abandoned during shutdown

2. Define cancellation policy

• evaluate CancellationToken / TaskTracker
• define shutdown requested / draining / completed phases
• add a common mechanism to verify all child tasks have stopped

3. Improve shared-state ownership

• ban Mutex/RwLock guards across .await
• move important shared state to actors or single-owner tasks
• redesign the lock-based SendStream serialization path

4. Harden drain and restart consistency

• guarantee pending-work drain and timestamp flush on shutdown
• make timestamp-file persistence safer
• add restart consistency tests

Acceptance Criteria

• no detached tasks remain after shutdown/reload/reconnect
• core lock-across-await paths are removed or structurally redesigned
• top-level shutdown completion means real child-task drain completion
• policy/stream/timestamp state is managed consistently
• tests cover cancellation, shutdown, and restart scenarios

[octoaide]

I propose breaking this work into a set of self-contained sub-issues so we can implement, review, and land changes incrementally. Each sub-issue requires implementation, tests that demonstrate the fix/behavior, and documentation/notes explaining the design. Please react with 👍 to approve these sub-issues or reply with requested changes. Once approved I can create the issues (or start work on them).

Proposed sub-issues (summary):

1. Audit and baseline tests for cancellation and lock-across-await

• Add an automated audit run (grep/Clippy rules or a dedicated script) and a set of unit/integration tests that expose current problematic paths (locks held across .await, detached tasks not joined, timestamp flush races). The goal is to produce failing tests that describe the problem surface before we change behavior.

2. Add a repository-wide cooperative cancellation framework (CancellationToken + TaskTracker)

• Implement a small common CancellationToken/TaskTracker abstraction (or adopt a vetted crate) and define shutdown phases (shutdown-requested, draining, complete). Provide helpers for top-level tasks to register and await child shutdown, and a single place to request graceful shutdown/reload.

3. Move critical shared state to single-owner actors

• Rework active_policy_list, delete_policy_ids, INGEST_CHANNEL, LAST_TRANSFER_TIME (and similar) to be owned by single tasks/actors and accessed via mpsc requests or channels. Remove direct sharing of these globals across threads/tasks. Include migration changes and tests.

4. Remove/replace lock-across-await patterns and redesign SendStream serialization

• Find and remove Mutex/RwLock guard usage across .await points. Redesign the SendStream serialization path so per-connection serialization is single-owner or uses explicit async-safe message passing rather than held locks across awaits.

5. Make timestamp persistence and shutdown flush durable and safe

• Ensure LAST_TRANSFER_TIME and any timestamp files are atomically persisted (write-to-temp + rename) and flushed to disk before shutdown completes. Guarantee persistence in the shutdown/drain phase defined by the cancellation framework, with tests that simulate abrupt shutdown/restart.

6. Integration tests for shutdown/drain/restart and CI gating

• Add integration tests that verify: no detached tasks remain after shutdown/reload/reconnect, core lock-across-await paths are removed, shutdown completion implies real child-task drain, and restart-time state (policies/timestamps) is consistent. Wire these tests into CI or provide a CI job that gates the changes.

If this plan looks good, please react with 👍; I will then create the sub-issues (each marked as a sub-issue of #302) or begin implementing them (starting with the audit and the cancellation framework). If you prefer different scope or order, please suggest changes.

[henry0715-dev]

@octoaide
Please proceed with implementation as one issue without separating it as a sub issue.
Instead, please proceed by committing step by step.