Stop collecting logs once the buffer is full

crates/logs-collector/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "logs-collector"
-version = "2.1.9"
+version = "2.1.10"
 edition = "2021"
 
 [dependencies]

crates/logs-collector/src/collector.rs

@@ -74,6 +74,12 @@ impl<T: Storage + Sync> LogsCollector<T> {
         }
     }
 
+    /// Whether the buffer has reached its memory limit. Callers use this to stop
+    /// collecting more logs that would only be dropped (see `buffer_logs`).
+    pub fn is_full(&self) -> bool {
+        self.buffer.lock().size >= self.max_buffer_size
+    }
+
     pub async fn dump_buffer(&mut self) -> anyhow::Result<()> {
         let logs = {
             let buffer = self.buffer.get_mut();

crates/logs-collector/src/server.rs

@@ -117,6 +117,12 @@ where
     }
 
     async fn collect_logs(&self, worker_id: PeerId, mut from_timestamp_ms: u64) {
+        // Don't even request logs we'd have to drop. The buffer drains every round,
+        // so these workers are picked up again next time.
+        if self.logs_collector.is_full() {
+            log::debug!("Buffer full, skipping log collection from {worker_id}");
+            return;
+        }
         let mut last_query_id = None;
         for page in 0..MAX_PAGES {
             if page == 0 {
@@ -163,6 +169,10 @@ where
             if !logs.has_more {
                 return;
             }
+            if self.logs_collector.is_full() {
+                log::debug!("Buffer full, stopping log collection from {worker_id}");
+                return;
+            }
         }
         log::warn!("Logs from {worker_id} didn't fit in {MAX_PAGES} pages, giving up");
     }