remove finalizer from cache

Author: GreyLilac09

src/sinks/util/buffer/metrics/normalize.rs

@@ -649,7 +649,13 @@ impl MetricSet {
         }
     }
 
-    fn insert(&mut self, metric: Metric, timestamp: Option<Instant>) {
+    fn insert(&mut self, mut metric: Metric, timestamp: Option<Instant>) {
+        // Strip finalizers before caching. The cache is only used for
+        // normalization state (tracking running totals) and must not hold
+        // Arc<EventFinalizer> references, as that prevents the disk buffer
+        // from acknowledging events — leading to a deadlock once the buffer
+        // fills up and no new events can replace cache entries.
+        metric.metadata_mut().take_finalizers();
         let (series, entry) = MetricEntry::from_metric(metric, timestamp);
         self.insert_with_tracking(series, entry);
     }

src/sinks/util/builder.rs

@@ -231,7 +231,7 @@ pub trait SinkBuilderExt: Stream {
         Self: Stream<Item = Metric> + Unpin + Sized,
         N: MetricNormalize + Default,
     {
-        match maybe_ttl_secs {
+        match maybe_ttl_secs.filter(|&ttl| ttl > 0.0) {
             None => Normalizer::new(self, N::default()),
             Some(ttl) => {
                 Normalizer::new_with_ttl(self, N::default(), Duration::from_secs(ttl as u64))