feat: Add support for flushing the event buffer

devtools/src/aggregator.rs

@@ -80,6 +80,10 @@ impl Aggregator {
         loop {
             let should_publish = tokio::select! {
                 _ = interval.tick() => true,
+                _ = self.shared.flush.notified() => {
+                    tracing::debug!("event buffer approaching capacity, flushing...");
+                    false
+                },
                 cmd = self.cmds.recv() => {
                     if let Some(Command::Instrument(watcher)) = cmd {
                         self.attach_watcher(watcher).await;
@@ -307,7 +311,7 @@ impl<T, const CAP: usize> EventBuf<T, CAP> {
     // TODO does it really make sense to track the dropped events here?
     pub fn push_overwrite(&mut self, item: T) {
         if self.inner.push_overwrite(item).is_some() {
-            self.sent -= 1;
+            self.sent = self.sent.saturating_sub(1);
         }
     }
 

devtools/src/layer.rs

@@ -1,5 +1,5 @@
 use crate::visitors::{EventVisitor, FieldVisitor};
-use crate::{Event, Shared};
+use crate::{Event, Shared, EVENT_BUFFER_CAPACITY};
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
@@ -51,6 +51,11 @@ impl Layer {
                 dropped.fetch_add(1, Ordering::Release);
             }
         }
+
+        let capacity = self.tx.capacity();
+        if capacity <= EVENT_BUFFER_CAPACITY / 2 {
+            self.shared.flush.notify_one();
+        }
     }
 }
 

devtools/src/lib.rs

@@ -40,7 +40,9 @@ use std::sync::atomic::AtomicUsize;
 use std::time::Instant;
 use tauri::Runtime;
 use tauri_devtools_wire_format::{instrument, Field};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, Notify};
+
+const EVENT_BUFFER_CAPACITY: usize = 512;
 
 pub(crate) type Result<T> = std::result::Result<T, Error>;
 
@@ -73,6 +75,7 @@ pub fn try_init<R: Runtime>() -> Result<tauri::plugin::TauriPlugin<R>> {
 pub(crate) struct Shared {
     dropped_log_events: AtomicUsize,
     dropped_span_events: AtomicUsize,
+    flush: Notify,
 }
 
 /// Data sent from the `Layer` to the `Aggregator`

web-client/src/views/dashboard/layout.tsx

@@ -104,6 +104,17 @@ export default function Layout() {
       setMonitorData("metadata", (prev) => updateSpanMetadata(prev, update));
     }
 
+    if (update.logsUpdate && update.spansUpdate) {
+      console.assert(
+        update.logsUpdate.droppedEvents == 0n,
+        "Dropped log events because the internal event buffer was at capacity. This is a bug, please report!"
+      );
+      console.assert(
+        update.spansUpdate.droppedEvents == 0n,
+        "Dropped span events because the internal event buffer was at capacity. This is a bug, please report!"
+      );
+    }
+
     const logsUpdate = update.logsUpdate;
     if (logsUpdate && logsUpdate.logEvents.length > 0) {
       setMonitorData("logs", (prev) => [...prev, ...logsUpdate.logEvents]);