fix: RAII cleanup for registries and addon lifecycle [latest]

- Add Drop impl for TrunkRegistrar (cancel parent_cancel)
- Add Drop impl for ClusterEventHub (abort background task)
- Add shutdown() to Addon trait + shutdown_all() to AddonRegistry
- Call shutdown_all() in app::run() during shutdown
- Add Drop impl for TelemetryAddon (flush OTel spans)
- Add CancellationToken to ACME and Archive addon background tasks

Author: shenjinti

src/addons/acme/handlers.rs

@@ -725,11 +725,21 @@ fn get_cert_expiry(cert_path: &StdPath) -> anyhow::Result<std::time::SystemTime>
 // ============================================================================
 
 /// Spawn the background task that checks certificate expiry and triggers auto-renewal
-pub async fn spawn_auto_renew_checker(acme_state: super::AcmeState, app_state: AppState) {
+pub async fn spawn_auto_renew_checker(
+    acme_state: super::AcmeState,
+    app_state: AppState,
+    cancel: tokio_util::sync::CancellationToken,
+) {
     let check_interval = tokio::time::Duration::from_secs(3600); // Check every hour
 
     loop {
-        tokio::time::sleep(check_interval).await;
+        tokio::select! {
+            _ = cancel.cancelled() => {
+                info!("ACME auto-renew checker stopped");
+                return;
+            }
+            _ = tokio::time::sleep(check_interval) => {}
+        }
 
         let config = acme_state.auto_renew_config.read().await.clone();
 

src/addons/acme/mod.rs

@@ -114,8 +114,9 @@ impl Addon for AcmeAddon {
         // Spawn background task for certificate expiry checking
         let state_clone = self.state.clone();
         let app_state = state.clone();
+        let cancel_token = state.token().child_token();
         crate::utils::spawn(async move {
-            handlers::spawn_auto_renew_checker(state_clone, app_state).await;
+            handlers::spawn_auto_renew_checker(state_clone, app_state, cancel_token).await;
         });
 
         tracing::info!("ACME Addon initialized");

src/addons/archive/mod.rs

@@ -334,10 +334,20 @@ impl ArchiveAddon {
         Ok(())
     }
 
-    async fn run_scheduler(state: AppState, archive_state: ArchiveState) {
+    async fn run_scheduler(
+        state: AppState,
+        archive_state: ArchiveState,
+        cancel: tokio_util::sync::CancellationToken,
+    ) {
         let mut interval = time::interval(time::Duration::from_secs(60));
         loop {
-            interval.tick().await;
+            tokio::select! {
+                _ = cancel.cancelled() => {
+                    info!("Archive scheduler stopped");
+                    return;
+                }
+                _ = interval.tick() => {}
+            }
 
             let archive_config = {
                 let guard = archive_state.config.read().unwrap();
@@ -560,8 +570,9 @@ impl Addon for ArchiveAddon {
         }
 
         let archive_state = self.state.clone();
+        let cancel = state.token().child_token();
         crate::utils::spawn(async move {
-            Self::run_scheduler(state, archive_state).await;
+            Self::run_scheduler(state, archive_state, cancel).await;
         });
         Ok(())
     }

src/addons/mod.rs

@@ -160,6 +160,10 @@ pub trait Addon: Send + Sync {
         None
     }
 
+    /// Shutdown the addon, releasing any resources (background tasks, connections, etc.).
+    /// Called during application shutdown after all servers have stopped.
+    async fn shutdown(&self) {}
+
     /// Return database migrations for this addon.
     fn migrations(&self) -> Vec<Box<dyn sea_orm_migration::MigrationTrait>> {
         vec![]

src/addons/registry.rs

@@ -419,6 +419,13 @@ impl AddonRegistry {
         }
     }
 
+    /// Shutdown all addons, releasing resources.
+    pub async fn shutdown_all(&self) {
+        for addon in &self.addons {
+            addon.shutdown().await;
+        }
+    }
+
     /// Run database migrations for all enabled addons.
     pub async fn run_migrations(&self, db: &sea_orm::DatabaseConnection) -> anyhow::Result<()> {
         let manager = sea_orm_migration::SchemaManager::new(db);

src/app.rs

@@ -745,6 +745,10 @@ pub async fn run(state: AppState, mut router: Router) -> Result<()> {
             info!("Application shutting down due to cancellation");
         }
     }
+
+    // Shutdown addons (release background tasks, connections, etc.)
+    state.addon_registry.shutdown_all().await;
+
     Ok(())
 }
 

src/proxy/cluster_event.rs

@@ -12,7 +12,7 @@ use crate::proxy::locator::LocatorEvent;
 use crate::proxy::presence::{PresenceManager, PresenceState, PresenceStatus};
 use anyhow::{Result, anyhow};
 use async_trait::async_trait;
-use parking_lot::RwLock;
+use parking_lot::{Mutex, RwLock};
 use rsipstack::sip::Param;
 use rsipstack::sip::headers::typed::CSeq;
 use rsipstack::sip::headers::{CallId, ContentType};
@@ -251,6 +251,8 @@ pub struct ClusterEventHub {
     handlers: RwLock<Vec<Arc<dyn ClusterEventHandler>>>,
     /// Child of the SIP server's cancel token; used to stop the dispatcher.
     cancel: tokio_util::sync::CancellationToken,
+    /// Handle to the background dispatcher task (aborted on drop).
+    task_handle: Mutex<Option<tokio::task::JoinHandle<()>>>,
 }
 
 impl ClusterEventHub {
@@ -270,6 +272,7 @@ impl ClusterEventHub {
             peers,
             handlers: RwLock::new(Vec::new()),
             cancel,
+            task_handle: Mutex::new(None),
         }
     }
 
@@ -286,7 +289,7 @@ impl ClusterEventHub {
         let mut rx = self.locator_events.subscribe();
         let cancel = self.cancel.clone();
         let this = self.clone();
-        crate::utils::spawn(async move {
+        let handle = crate::utils::spawn(async move {
             loop {
                 tokio::select! {
                     _ = cancel.cancelled() => break,
@@ -306,6 +309,7 @@ impl ClusterEventHub {
                 }
             }
         });
+        *self.task_handle.lock() = Some(handle);
     }
 
     async fn dispatch_local_locator_event(&self, event: LocatorEvent) {
@@ -542,6 +546,15 @@ impl ClusterEventHub {
     }
 }
 
+impl Drop for ClusterEventHub {
+    fn drop(&mut self) {
+        self.cancel.cancel();
+        if let Some(handle) = self.task_handle.lock().take() {
+            handle.abort();
+        }
+    }
+}
+
 // ── ClusterEventModule (ProxyModule) ────────────────────────────────────────
 
 pub struct ClusterEventModule {

src/proxy/trunk_registrar.rs

@@ -51,6 +51,12 @@ pub struct TrunkRegistrar {
     endpoint: RwLock<Option<EndpointInnerRef>>,
 }
 
+impl Drop for TrunkRegistrar {
+    fn drop(&mut self) {
+        self.parent_cancel.cancel();
+    }
+}
+
 impl Default for TrunkRegistrar {
     fn default() -> Self {
         Self::new()