tracing_throttle/application/

emitter.rs

//! Summary emission for suppressed events.
//!
//! Periodically collects and emits summaries of suppressed events to provide
//! visibility into what has been rate limited.

use crate::application::{
    ports::Storage,
    registry::{EventState, SuppressionRegistry},
};
use crate::domain::{
    signature::EventSignature,
    summary::{ClaimedSuppressions, SuppressionSummary},
};
use std::time::Duration;

#[cfg(feature = "async")]
use tokio::{sync::watch, time::interval};

/// Error returned when emitter configuration validation fails.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EmitterConfigError {
    /// Summary interval duration must be greater than zero
    ZeroSummaryInterval,
}

impl std::fmt::Display for EmitterConfigError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            EmitterConfigError::ZeroSummaryInterval => {
                write!(f, "summary interval must be greater than 0")
            }
        }
    }
}

impl std::error::Error for EmitterConfigError {}

/// Error returned when emitter shutdown fails.
#[cfg(feature = "async")]
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ShutdownError {
    /// Task panicked during shutdown
    TaskPanicked,
    /// Task was cancelled before completing
    TaskCancelled,
    /// Shutdown exceeded the specified timeout
    Timeout,
    /// Failed to send shutdown signal (task may have already exited)
    SignalFailed,
}

#[cfg(feature = "async")]
impl std::fmt::Display for ShutdownError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ShutdownError::TaskPanicked => write!(f, "emitter task panicked during shutdown"),
            ShutdownError::TaskCancelled => write!(f, "emitter task was cancelled"),
            ShutdownError::Timeout => write!(f, "shutdown exceeded timeout"),
            ShutdownError::SignalFailed => write!(f, "failed to send shutdown signal"),
        }
    }
}

#[cfg(feature = "async")]
impl std::error::Error for ShutdownError {}

/// Configuration for summary emission.
#[derive(Debug, Clone)]
pub struct EmitterConfig {
    /// How often to emit summaries
    pub interval: Duration,
    /// Minimum suppression count to include in summary
    pub min_count: usize,
}

impl Default for EmitterConfig {
    fn default() -> Self {
        Self {
            interval: Duration::from_secs(30),
            min_count: 1,
        }
    }
}

impl EmitterConfig {
    /// Create a new emitter config with the specified interval.
    ///
    /// # Errors
    /// Returns `EmitterConfigError::ZeroSummaryInterval` if `interval` is zero.
    pub fn new(interval: Duration) -> Result<Self, EmitterConfigError> {
        if interval.is_zero() {
            return Err(EmitterConfigError::ZeroSummaryInterval);
        }
        Ok(Self {
            interval,
            min_count: 1,
        })
    }

    /// Set the minimum suppression count threshold.
    pub fn with_min_count(mut self, min_count: usize) -> Self {
        self.min_count = min_count;
        self
    }
}

/// Handle for controlling a running emitter task.
///
/// # Shutdown Behavior
///
/// You **must** call `shutdown().await` to stop the emitter task. The handle does
/// not implement `Drop` to avoid race conditions and resource leaks.
///
/// If you drop the handle without calling `shutdown()`, the background task will
/// continue running indefinitely, potentially causing:
/// - Resource leaks (the task holds references to the registry)
/// - Unexpected behavior if the task outlives expected lifetime
/// - Inability to observe task failures or panics
///
/// # Examples
///
/// ```rust,no_run
/// # use tracing_throttle::application::emitter::{SummaryEmitter, EmitterConfig};
/// # use tracing_throttle::application::registry::SuppressionRegistry;
/// # use tracing_throttle::domain::policy::Policy;
/// # use tracing_throttle::infrastructure::storage::ShardedStorage;
/// # use tracing_throttle::infrastructure::clock::SystemClock;
/// # use std::sync::Arc;
/// # async fn example() {
/// # let storage = Arc::new(ShardedStorage::new());
/// # let clock = Arc::new(SystemClock::new());
/// # let policy = Policy::count_based(100).unwrap();
/// # let registry = SuppressionRegistry::new(storage, clock, policy);
/// # let config = EmitterConfig::default();
/// # let emitter = SummaryEmitter::new(registry, config);
/// let handle = emitter.start(|_| {}, false);
///
/// // Always call shutdown explicitly
/// handle.shutdown().await.expect("shutdown failed");
/// # }
/// ```
#[cfg(feature = "async")]
pub struct EmitterHandle {
    shutdown_tx: watch::Sender<bool>,
    join_handle: Option<tokio::task::JoinHandle<()>>,
}

#[cfg(feature = "async")]
impl EmitterHandle {
    /// Trigger graceful shutdown and wait for the task to complete.
    ///
    /// This method uses a default timeout of 10 seconds. For custom timeout durations,
    /// use [`shutdown_with_timeout`](Self::shutdown_with_timeout).
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The task panics during shutdown
    /// - The task is cancelled
    /// - Shutdown exceeds the timeout (10 seconds)
    /// - The shutdown signal fails to send
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use tracing_throttle::application::emitter::{SummaryEmitter, EmitterConfig};
    /// # use tracing_throttle::application::registry::SuppressionRegistry;
    /// # use tracing_throttle::domain::policy::Policy;
    /// # use tracing_throttle::infrastructure::storage::ShardedStorage;
    /// # use tracing_throttle::infrastructure::clock::SystemClock;
    /// # use std::sync::Arc;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let storage = Arc::new(ShardedStorage::new());
    /// # let clock = Arc::new(SystemClock::new());
    /// # let policy = Policy::count_based(100).unwrap();
    /// # let registry = SuppressionRegistry::new(storage, clock, policy);
    /// # let config = EmitterConfig::default();
    /// # let emitter = SummaryEmitter::new(registry, config);
    /// let handle = emitter.start(|_| {}, false);
    ///
    /// // Do some work...
    ///
    /// // Clean shutdown with default 10 second timeout
    /// handle.shutdown().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn shutdown(self) -> Result<(), ShutdownError> {
        self.shutdown_with_timeout(Duration::from_secs(10)).await
    }

    /// Trigger graceful shutdown with a custom timeout.
    ///
    /// This method:
    /// 1. Sends the shutdown signal
    /// 2. Waits for the background task to finish (up to the timeout)
    /// 3. If the timeout is exceeded, the task is aborted
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - The task panics during shutdown
    /// - The task is cancelled
    /// - Shutdown exceeds the specified timeout
    /// - The shutdown signal fails to send
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use tracing_throttle::application::emitter::{SummaryEmitter, EmitterConfig};
    /// # use tracing_throttle::application::registry::SuppressionRegistry;
    /// # use tracing_throttle::domain::policy::Policy;
    /// # use tracing_throttle::infrastructure::storage::ShardedStorage;
    /// # use tracing_throttle::infrastructure::clock::SystemClock;
    /// # use std::sync::Arc;
    /// # use std::time::Duration;
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// # let storage = Arc::new(ShardedStorage::new());
    /// # let clock = Arc::new(SystemClock::new());
    /// # let policy = Policy::count_based(100).unwrap();
    /// # let registry = SuppressionRegistry::new(storage, clock, policy);
    /// # let config = EmitterConfig::default();
    /// # let emitter = SummaryEmitter::new(registry, config);
    /// let handle = emitter.start(|_| {}, false);
    ///
    /// // Do some work...
    ///
    /// // Clean shutdown with 5 second timeout
    /// handle.shutdown_with_timeout(Duration::from_secs(5)).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn shutdown_with_timeout(
        mut self,
        timeout_duration: Duration,
    ) -> Result<(), ShutdownError> {
        use tokio::time::timeout;

        // Send shutdown signal
        if self.shutdown_tx.send(true).is_err() {
            return Err(ShutdownError::SignalFailed);
        }

        // Wait for task to complete with timeout
        if let Some(handle) = self.join_handle.take() {
            match timeout(timeout_duration, handle).await {
                Ok(Ok(())) => Ok(()),
                Ok(Err(e)) if e.is_panic() => Err(ShutdownError::TaskPanicked),
                Ok(Err(e)) if e.is_cancelled() => Err(ShutdownError::TaskCancelled),
                Ok(Err(_)) => Err(ShutdownError::TaskPanicked), // Treat unknown errors as panics
                Err(_) => Err(ShutdownError::Timeout),
            }
        } else {
            Ok(())
        }
    }

    /// Check if the emitter task is still running.
    pub fn is_running(&self) -> bool {
        self.join_handle.as_ref().is_some_and(|h| !h.is_finished())
    }
}

/// Emits periodic summaries of suppressed events.
pub struct SummaryEmitter<S>
where
    S: Storage<EventSignature, EventState> + Clone,
{
    registry: SuppressionRegistry<S>,
    config: EmitterConfig,
}

struct ClaimedSummary {
    signature: EventSignature,
    claim: ClaimedSuppressions,
}

struct ClaimedSummaryBatch {
    summaries: Vec<SuppressionSummary>,
    claims: Vec<ClaimedSummary>,
}

impl<S> SummaryEmitter<S>
where
    S: Storage<EventSignature, EventState> + Clone,
{
    /// Create a new summary emitter.
    pub fn new(registry: SuppressionRegistry<S>, config: EmitterConfig) -> Self {
        Self { registry, config }
    }

    /// Collect suppression summaries for newly reported suppressions.
    ///
    /// Returns summaries for events that have accumulated at least `min_count`
    /// suppressions since their previous emission.
    pub fn collect_summaries(&self) -> Vec<SuppressionSummary> {
        self.collect_claimed_summaries().summaries
    }

    fn collect_claimed_summaries(&self) -> ClaimedSummaryBatch {
        let mut summaries = Vec::new();
        let mut claims = Vec::new();
        let min_count = self.config.min_count;

        self.registry.cleanup(|signature, state| {
            if let Some(claim) = state.counter.claim_unreported(min_count) {
                #[cfg(feature = "human-readable")]
                let summary = SuppressionSummary::from_claim_with_metadata(
                    *signature,
                    claim.clone(),
                    state.metadata.clone(),
                );

                #[cfg(not(feature = "human-readable"))]
                let summary = SuppressionSummary::from_claim(*signature, claim.clone());

                claims.push(ClaimedSummary {
                    signature: *signature,
                    claim,
                });
                summaries.push(summary);
            }

            true
        });

        ClaimedSummaryBatch { summaries, claims }
    }

    fn rollback_claimed_summaries(&self, claims: &[ClaimedSummary]) {
        if claims.is_empty() {
            return;
        }

        self.registry.cleanup(|signature, state| {
            for claim in claims {
                if claim.signature == *signature {
                    state.counter.rollback_claim(&claim.claim);
                }
            }

            true
        });
    }

    fn emit_claimed_summaries<F>(&self, batch: ClaimedSummaryBatch, emit_fn: &mut F) -> bool
    where
        F: FnMut(Vec<SuppressionSummary>),
    {
        if batch.summaries.is_empty() {
            return true;
        }

        let claims = batch.claims;
        let summaries = batch.summaries;

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            emit_fn(summaries);
        }));

        if result.is_err() {
            self.rollback_claimed_summaries(&claims);
            false
        } else {
            true
        }
    }

    /// Start emitting summaries periodically (async version).
    ///
    /// This spawns a background task that emits summaries at the configured interval.
    /// The task will run until `shutdown()` is called on the returned `EmitterHandle`.
    ///
    /// # Graceful Shutdown
    ///
    /// When `shutdown()` is called on the `EmitterHandle`:
    /// 1. The shutdown signal is prioritized over tick events
    /// 2. The current emission completes if in progress
    /// 3. If `emit_final` is true, one final emission occurs with current summaries
    /// 4. The background task completes gracefully
    ///
    /// # Cancellation Safety
    ///
    /// The spawned task is designed to shut down cleanly:
    /// - `collect_summaries()` atomically claims unreported suppressions
    /// - Claimed suppressions are not emitted again on later ticks
    /// - Panics in `emit_fn` are caught and don't abort the task
    /// - The `emit_fn` closure should be cancellation-safe (avoid holding locks across `.await`)
    ///
    /// # Type Parameters
    ///
    /// * `F` - The emission function. Must be `Send + 'static` because it runs
    ///   in a spawned task that may execute on any thread in the tokio runtime.
    ///   The function receives ownership of the summaries vector.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # use tracing_throttle::application::emitter::{SummaryEmitter, EmitterConfig};
    /// # use tracing_throttle::application::registry::SuppressionRegistry;
    /// # use tracing_throttle::domain::policy::Policy;
    /// # use tracing_throttle::infrastructure::storage::ShardedStorage;
    /// # use tracing_throttle::infrastructure::clock::SystemClock;
    /// # use std::sync::Arc;
    /// # use std::time::Duration;
    /// # async fn example() {
    /// # let storage = Arc::new(ShardedStorage::new());
    /// # let clock = Arc::new(SystemClock::new());
    /// # let policy = Policy::count_based(100).unwrap();
    /// # let registry = SuppressionRegistry::new(storage, clock, policy);
    /// # let config = EmitterConfig::default();
    /// let emitter = SummaryEmitter::new(registry, config);
    /// let handle = emitter.start(|summaries| {
    ///     for summary in summaries {
    ///         tracing::warn!("{}", summary.format_message());
    ///     }
    /// }, true);
    ///
    /// // Later, trigger graceful shutdown
    /// handle.shutdown().await.expect("shutdown failed");
    /// # }
    /// ```
    #[cfg(feature = "async")]
    pub fn start<F>(self, mut emit_fn: F, emit_final: bool) -> EmitterHandle
    where
        F: FnMut(Vec<SuppressionSummary>) + Send + 'static,
        S: Send + 'static,
    {
        let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);

        let handle = tokio::spawn(async move {
            let mut ticker = interval(self.config.interval);
            // Skip missed ticks to prevent backpressure if emissions are slow
            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);

            loop {
                tokio::select! {
                    // Prioritize shutdown signal to ensure fast shutdown
                    biased;

                    _ = shutdown_rx.changed() => {
                        if *shutdown_rx.borrow_and_update() {
                            // Emit final summaries if requested
                            if emit_final {
                                let batch = self.collect_claimed_summaries();
                                if !self.emit_claimed_summaries(batch, &mut emit_fn) {
                                    #[cfg(debug_assertions)]
                                    eprintln!("Warning: emit_fn panicked during final emission");
                                }
                            }
                            break;
                        }
                    }
                    _ = ticker.tick() => {
                        let batch = self.collect_claimed_summaries();
                        if !self.emit_claimed_summaries(batch, &mut emit_fn) {
                            // emit_fn panicked - claims were rolled back, continue running
                            #[cfg(debug_assertions)]
                            eprintln!("Warning: emit_fn panicked during emission");
                        }
                    }
                }
            }
        });

        EmitterHandle {
            shutdown_tx,
            join_handle: Some(handle),
        }
    }

    /// Get the emitter configuration.
    pub fn config(&self) -> &EmitterConfig {
        &self.config
    }

    /// Get a reference to the registry.
    pub fn registry(&self) -> &SuppressionRegistry<S> {
        &self.registry
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::{policy::Policy, signature::EventSignature};
    use crate::infrastructure::clock::SystemClock;
    use crate::infrastructure::storage::ShardedStorage;
    use std::sync::Arc;

    #[test]
    fn test_collect_summaries_empty() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::default();
        let emitter = SummaryEmitter::new(registry.clone(), config);

        let summaries = emitter.collect_summaries();
        assert!(summaries.is_empty());
    }

    #[test]
    fn test_collect_summaries_with_suppressions() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::default();

        // Add some suppressed events
        for i in 0..3 {
            let sig = EventSignature::simple("INFO", &format!("Message {}", i));
            registry.with_event_state(sig, |state, now| {
                // Simulate some suppressions
                for _ in 0..(i + 1) * 5 {
                    state.counter.record_suppression(now);
                }
            });
        }

        let emitter = SummaryEmitter::new(registry.clone(), config);
        let summaries = emitter.collect_summaries();

        assert_eq!(summaries.len(), 3);

        // Verify counts
        let counts: Vec<usize> = summaries.iter().map(|s| s.count).collect();
        assert!(counts.contains(&5));
        assert!(counts.contains(&10));
        assert!(counts.contains(&15));

        let summaries = emitter.collect_summaries();
        assert!(
            summaries.is_empty(),
            "already emitted summaries should not be emitted again"
        );
    }

    #[test]
    fn test_min_count_filtering() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::default().with_min_count(10);

        // Add event with low count (below threshold)
        let sig1 = EventSignature::simple("INFO", "Low count");
        registry.with_event_state(sig1, |state, now| {
            for _ in 0..4 {
                state.counter.record_suppression(now);
            }
        });

        // Add event with high count (above threshold)
        let sig2 = EventSignature::simple("INFO", "High count");
        registry.with_event_state(sig2, |state, now| {
            for _ in 0..14 {
                state.counter.record_suppression(now);
            }
        });

        let emitter = SummaryEmitter::new(registry.clone(), config);
        let summaries = emitter.collect_summaries();

        // Only the high-count event should be included
        assert_eq!(summaries.len(), 1);
        assert_eq!(summaries[0].count, 14);

        let summaries = emitter.collect_summaries();
        assert!(
            summaries.is_empty(),
            "claimed summaries should not be emitted again"
        );
    }

    #[test]
    fn test_min_count_accumulates_unreported_suppressions() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::default().with_min_count(10);

        let sig = EventSignature::simple("INFO", "Accumulated");
        registry.with_event_state(sig, |state, now| {
            for _ in 0..4 {
                state.counter.record_suppression(now);
            }
        });

        let emitter = SummaryEmitter::new(registry.clone(), config);
        assert!(emitter.collect_summaries().is_empty());

        registry.with_event_state(sig, |state, now| {
            for _ in 0..6 {
                state.counter.record_suppression(now);
            }
        });

        let summaries = emitter.collect_summaries();
        assert_eq!(summaries.len(), 1);
        assert_eq!(summaries[0].count, 10);
    }

    #[test]
    fn test_collect_summaries_reports_only_new_suppressions() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::default();

        let sig = EventSignature::simple("INFO", "Delta");
        registry.with_event_state(sig, |state, now| {
            for _ in 0..5 {
                state.counter.record_suppression(now);
            }
        });

        let emitter = SummaryEmitter::new(registry.clone(), config);
        let summaries = emitter.collect_summaries();
        assert_eq!(summaries.len(), 1);
        assert_eq!(summaries[0].count, 5);

        assert!(emitter.collect_summaries().is_empty());

        registry.with_event_state(sig, |state, now| {
            for _ in 0..3 {
                state.counter.record_suppression(now);
            }
        });

        let summaries = emitter.collect_summaries();
        assert_eq!(summaries.len(), 1);
        assert_eq!(summaries[0].count, 3);
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_async_emission() {
        use std::sync::Mutex;

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(100)).unwrap();

        // Add a suppressed event
        let sig = EventSignature::simple("INFO", "Test");
        registry.with_event_state(sig, |state, now| {
            state.counter.record_suppression(now);
        });

        let emitter = SummaryEmitter::new(registry.clone(), config);

        // Track emissions
        let emissions = Arc::new(Mutex::new(Vec::new()));
        let emissions_clone = Arc::clone(&emissions);

        let handle = emitter.start(
            move |summaries| {
                emissions_clone.lock().unwrap().push(summaries.len());
            },
            false,
        );

        // Wait for a couple of intervals
        tokio::time::sleep(Duration::from_millis(250)).await;

        handle.shutdown().await.expect("shutdown failed");

        // Should emit the suppression once, then stay quiet until new suppressions arrive.
        let emission_count = emissions.lock().unwrap().len();
        assert_eq!(emission_count, 1);
    }

    #[test]
    fn test_emitter_config_zero_interval() {
        let result = EmitterConfig::new(Duration::from_secs(0));
        assert!(matches!(
            result,
            Err(EmitterConfigError::ZeroSummaryInterval)
        ));
    }

    #[test]
    fn test_emitter_config_valid_interval() {
        let config = EmitterConfig::new(Duration::from_secs(30)).unwrap();
        assert_eq!(config.interval, Duration::from_secs(30));
        assert_eq!(config.min_count, 1);
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_graceful_shutdown() {
        use std::sync::Mutex;

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(100)).unwrap();

        // Add a suppressed event so there's something to emit
        let sig = EventSignature::simple("INFO", "Test");
        registry.with_event_state(sig, |state, now| {
            state.counter.record_suppression(now);
        });

        let emitter = SummaryEmitter::new(registry.clone(), config);

        let emissions = Arc::new(Mutex::new(0));
        let emissions_clone = Arc::clone(&emissions);

        let handle = emitter.start(
            move |_| {
                *emissions_clone.lock().unwrap() += 1;
            },
            false,
        );

        // Let it run for a bit
        tokio::time::sleep(Duration::from_millis(250)).await;

        // Trigger graceful shutdown
        handle.shutdown().await.expect("shutdown failed");

        // Verify task is no longer running
        let final_count = *emissions.lock().unwrap();
        assert!(final_count >= 1);

        // Wait a bit more to ensure task really stopped
        tokio::time::sleep(Duration::from_millis(150)).await;
        let count_after_shutdown = *emissions.lock().unwrap();
        assert_eq!(count_after_shutdown, final_count);
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_shutdown_with_final_emission() {
        use std::sync::Mutex;

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_secs(60)).unwrap(); // Long interval

        let emitter = SummaryEmitter::new(registry.clone(), config);

        let emissions = Arc::new(Mutex::new(Vec::new()));
        let emissions_clone = Arc::clone(&emissions);

        let handle = emitter.start(
            move |summaries| {
                emissions_clone.lock().unwrap().push(summaries.len());
            },
            true, // Emit final summaries
        );

        // Wait for first tick to complete (interval's first tick is immediate)
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Now add some suppressions after the first tick
        let sig = EventSignature::simple("INFO", "Test event");
        registry.with_event_state(sig, |state, now| {
            for _ in 0..10 {
                state.counter.record_suppression(now);
            }
        });

        // Shutdown before next interval (which is 60 seconds away)
        tokio::time::sleep(Duration::from_millis(50)).await;
        handle.shutdown().await.expect("shutdown failed");

        // Should have emitted final summaries
        let emission_list = emissions.lock().unwrap();
        assert_eq!(emission_list.len(), 1);
        assert_eq!(emission_list[0], 1); // 1 summary
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_shutdown_without_final_emission() {
        use std::sync::Mutex;

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_secs(60)).unwrap();

        let emitter = SummaryEmitter::new(registry.clone(), config);

        let emissions = Arc::new(Mutex::new(0));
        let emissions_clone = Arc::clone(&emissions);

        let handle = emitter.start(
            move |_| {
                *emissions_clone.lock().unwrap() += 1;
            },
            false, // No final emission
        );

        // Wait for first tick (immediate, but no emissions since no suppressions yet)
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Add some suppressions after first tick
        let sig = EventSignature::simple("INFO", "Test event");
        registry.with_event_state(sig, |state, now| {
            state.counter.record_suppression(now);
        });

        // Shutdown immediately (before next 60-second interval)
        tokio::time::sleep(Duration::from_millis(50)).await;
        handle.shutdown().await.expect("shutdown failed");

        // Should not have emitted anything (no final emission)
        assert_eq!(*emissions.lock().unwrap(), 0);
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_is_running() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(100)).unwrap();

        let emitter = SummaryEmitter::new(registry.clone(), config);
        let handle = emitter.start(|_| {}, false);

        // Should be running
        assert!(handle.is_running());

        // Shutdown
        handle.shutdown().await.expect("shutdown failed");

        // Should no longer be running
        tokio::time::sleep(Duration::from_millis(50)).await;
        // Note: is_running() consumes self, so we can't check after shutdown
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_shutdown_during_emission() {
        use std::sync::{Arc, Mutex};

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(50)).unwrap();

        // Add suppressions
        let sig = EventSignature::simple("INFO", "Test");
        registry.with_event_state(sig, |state, now| {
            state.counter.record_suppression(now);
        });

        let emitter = SummaryEmitter::new(registry, config);

        let emissions = Arc::new(Mutex::new(0));
        let emissions_clone = Arc::clone(&emissions);

        let handle = emitter.start(
            move |_| {
                // Simulate slow emission
                std::thread::sleep(Duration::from_millis(30));
                *emissions_clone.lock().unwrap() += 1;
            },
            false,
        );

        // Let first emission start
        tokio::time::sleep(Duration::from_millis(60)).await;

        // Shutdown should wait for current emission to complete
        handle.shutdown().await.expect("shutdown failed");

        // Current emission should have completed
        assert!(*emissions.lock().unwrap() >= 1);
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_shutdown_with_custom_timeout() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(100)).unwrap();

        let emitter = SummaryEmitter::new(registry, config);
        let handle = emitter.start(|_| {}, false);

        // Let it run briefly
        tokio::time::sleep(Duration::from_millis(150)).await;

        // Shutdown with custom timeout should succeed quickly
        let result = handle.shutdown_with_timeout(Duration::from_secs(5)).await;
        assert!(result.is_ok());
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_panic_in_emit_fn() {
        use std::sync::atomic::{AtomicUsize, Ordering};
        use std::sync::Mutex;

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(50)).unwrap();

        // Add suppressions
        let sig = EventSignature::simple("INFO", "Test");
        registry.with_event_state(sig, |state, now| {
            for _ in 0..5 {
                state.counter.record_suppression(now);
            }
        });

        let emitter = SummaryEmitter::new(registry.clone(), config);

        let call_count = Arc::new(AtomicUsize::new(0));
        let call_count_clone = Arc::clone(&call_count);
        let emitted_counts = Arc::new(Mutex::new(Vec::new()));
        let emitted_counts_clone = Arc::clone(&emitted_counts);

        let handle = emitter.start(
            move |summaries| {
                let total: usize = summaries.iter().map(|summary| summary.count).sum();
                emitted_counts_clone.lock().unwrap().push(total);
                let count = call_count_clone.fetch_add(1, Ordering::SeqCst);

                // Panic on first call, succeed on subsequent calls
                if count == 0 {
                    panic!("intentional panic for testing");
                }
                // If we get here, panic was handled and task continued
            },
            false,
        );

        // The first emission panics and rolls back; the next tick retries the same batch.
        tokio::time::sleep(Duration::from_millis(150)).await;

        handle.shutdown().await.expect("shutdown failed");

        // Should have attempted multiple emissions (first panicked, others succeeded)
        let final_count = call_count.load(Ordering::SeqCst);
        assert!(
            final_count > 1,
            "Task should continue after panic in emit_fn"
        );

        let emitted_counts = emitted_counts.lock().unwrap();
        assert!(
            emitted_counts.len() > 1,
            "Panicked emission should be retried"
        );
        assert_eq!(emitted_counts[0], 5);
        assert_eq!(emitted_counts[1], 5);
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_repeated_panic_in_emit_fn() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(30)).unwrap();

        let sig = EventSignature::simple("INFO", "Test");
        registry.with_event_state(sig, |state, now| {
            state.counter.record_suppression(now);
        });

        let emitter = SummaryEmitter::new(registry.clone(), config);

        let call_count = Arc::new(AtomicUsize::new(0));
        let call_count_clone = Arc::clone(&call_count);

        let handle = emitter.start(
            move |_summaries| {
                call_count_clone.fetch_add(1, Ordering::SeqCst);
                panic!("always panic");
            },
            false,
        );

        tokio::time::sleep(Duration::from_millis(70)).await;

        for _ in 0..3 {
            let sig = EventSignature::simple("INFO", "Test");
            registry.with_event_state(sig, |state, now| {
                state.counter.record_suppression(now);
            });
            tokio::time::sleep(Duration::from_millis(40)).await;
        }

        handle.shutdown().await.expect("shutdown failed");

        // Should have attempted multiple times despite continuous panics
        let final_count = call_count.load(Ordering::SeqCst);
        assert!(
            final_count >= 3,
            "Task should continue despite repeated panics, got {} calls",
            final_count
        );
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_panic_during_final_emission() {
        use std::sync::atomic::{AtomicBool, Ordering};

        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_secs(3600)).unwrap(); // Long interval

        let sig = EventSignature::simple("INFO", "Test");
        registry.with_event_state(sig, |state, now| {
            state.counter.record_suppression(now);
        });

        let emitter = SummaryEmitter::new(registry, config);

        let panicked = Arc::new(AtomicBool::new(false));
        let panicked_clone = Arc::clone(&panicked);

        let handle = emitter.start(
            move |_summaries| {
                panicked_clone.store(true, Ordering::SeqCst);
                panic!("panic during final emission");
            },
            true, // emit_on_shutdown
        );

        // Shutdown immediately - will trigger final emission which panics
        handle
            .shutdown()
            .await
            .expect("shutdown should succeed even if final emission panics");

        // Verify final emission was attempted
        assert!(
            panicked.load(Ordering::SeqCst),
            "Final emission should have been attempted"
        );
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_shutdown_timeout_with_slow_emit_fn() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_secs(3600)).unwrap();

        let emitter = SummaryEmitter::new(registry, config);

        let handle = emitter.start(
            move |_summaries| {
                // Simulate slow emission - use a shorter duration for testing
                std::thread::sleep(Duration::from_millis(500));
            },
            true,
        );

        // Shutdown with very short timeout
        let result = handle
            .shutdown_with_timeout(Duration::from_millis(10))
            .await;

        // Should timeout (or at minimum not panic)
        // Note: timing-sensitive test, may occasionally pass if emit completes quickly
        let _ = result; // Don't assert - just verify no panic
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_handle_dropped_without_shutdown() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(50)).unwrap();

        let emitter = SummaryEmitter::new(registry, config);

        let handle = emitter.start(|_summaries| {}, false);

        // Drop handle without calling shutdown
        drop(handle);

        // Task should continue running
        // This is documented behavior - not a bug
        tokio::time::sleep(Duration::from_millis(100)).await;

        // No assertions - just verify no panic
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_concurrent_shutdown_calls() {
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_secs(3600)).unwrap();

        let emitter = SummaryEmitter::new(registry, config);
        let handle = emitter.start(|_summaries| {}, false);

        // Create multiple shutdown senders
        let sender = handle.shutdown_tx.clone();
        let mut handles_vec = vec![];

        // Multiple tasks try to send shutdown signal concurrently
        for _ in 0..5 {
            let sender_clone = sender.clone();
            handles_vec.push(tokio::spawn(async move {
                let _ = sender_clone.send(true);
            }));
        }

        // All send operations should complete without panic
        for h in handles_vec {
            let _ = h.await;
        }

        // Clean shutdown
        let _ = handle.shutdown().await;
    }

    #[cfg(feature = "async")]
    #[tokio::test]
    async fn test_shutdown_signal_failure() {
        // Test that shutdown handles channel errors gracefully
        let storage = Arc::new(ShardedStorage::new());
        let clock = Arc::new(SystemClock::new());
        let policy = Policy::count_based(100).unwrap();
        let registry = SuppressionRegistry::new(storage, clock, policy);
        let config = EmitterConfig::new(Duration::from_millis(30)).unwrap();

        let emitter = SummaryEmitter::new(registry, config);
        let handle = emitter.start(|_summaries| {}, false);

        // Normal shutdown
        handle.shutdown().await.expect("shutdown should succeed");

        // Verify task stopped
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}