chore(buffers): Rework counter/gauge increment handling

lib/vector-buffers/src/buffer_usage_data.rs

@@ -15,6 +15,14 @@ use crate::{
     spawn_named,
 };
 
+fn update_counter(counter: &AtomicU64, delta: u64) {
+    counter
+        .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
+            Some(current.saturating_add(delta))
+        })
+        .ok();
+}
+
 /// Snapshot of category metrics.
 struct CategorySnapshot {
     event_count: u64,
@@ -43,9 +51,8 @@ struct CategoryMetrics {
 impl CategoryMetrics {
     /// Increments the event count and byte size by the given amounts.
     fn increment(&self, event_count: u64, event_byte_size: u64) {
-        self.event_count.fetch_add(event_count, Ordering::Relaxed);
-        self.event_byte_size
-            .fetch_add(event_byte_size, Ordering::Relaxed);
+        update_counter(&self.event_count, event_count);
+        update_counter(&self.event_byte_size, event_byte_size);
     }
 
     /// Sets the event count and event byte size to the given amount.
@@ -255,6 +262,7 @@ impl BufferUsage {
                 for stage in &stages {
                     let max_size = stage.max_size.get();
                     emit(BufferCreated {
+                        buffer_id: buffer_id.clone(),
                         idx: stage.idx,
                         max_size_bytes: max_size.event_byte_size,
                         max_size_events: max_size
@@ -313,3 +321,38 @@ impl BufferUsage {
         spawn_named(task.instrument(span.or_current()), task_name.as_str());
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::{sync::Arc, thread};
+
+    use super::*;
+
+    #[test]
+    fn multithreaded_updates_are_correct() {
+        const NUM_THREADS: u64 = 16;
+        const INCREMENTS_PER_THREAD: u64 = 100_000;
+        const EXPECTED_COUNT: u64 = NUM_THREADS * INCREMENTS_PER_THREAD;
+
+        let counter = Arc::new(AtomicU64::new(0));
+
+        let handles: Vec<_> = (0..NUM_THREADS)
+            .map(|_| {
+                let counter = Arc::clone(&counter);
+                thread::spawn(move || {
+                    for _ in 0..INCREMENTS_PER_THREAD {
+                        update_counter(&counter, 1);
+                    }
+                })
+            })
+            .collect();
+
+        for handle in handles {
+            handle.join().unwrap();
+        }
+
+        let final_count = counter.load(Ordering::Relaxed);
+
+        assert_eq!(final_count, EXPECTED_COUNT);
+    }
+}