allow runtime configurable log buffer size, drop circular_buffer crate dep

Cargo.lock

@@ -201,12 +201,6 @@ dependencies = [
  "windows-link",
 ]
 
-[[package]]
-name = "circular-buffer"
-version = "1.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "14c638459986b83c2b885179bd4ea6a2cbb05697b001501a56adb3a3d230803b"
-
 [[package]]
 name = "clap"
 version = "4.5.53"
@@ -1682,7 +1676,6 @@ name = "signal-gateway"
 version = "0.1.0"
 dependencies = [
  "chrono",
- "circular-buffer",
  "conf",
  "conf-extra",
  "displaydoc",

Cargo.toml

@@ -25,7 +25,6 @@ prom-client = { path = "prom-client" }
 
 async-trait = "0.1"
 chrono = { version = "0.4", default-features = false, features = ["clock", "serde", "std"] }
-circular-buffer = "1.2"
 conf = "0.4"
 conf-extra = "0.1"
 displaydoc = "0.2"

signal-gateway/Cargo.toml

@@ -10,7 +10,6 @@ workspace = true
 prom-client = { workspace = true }
 
 chrono = { workspace = true }
-circular-buffer = { workspace = true }
 conf = { workspace = true }
 conf-extra = { workspace = true }
 displaydoc = { workspace = true }

signal-gateway/src/gateway/circular_buffer.rs

@@ -0,0 +1,143 @@
+//! A circular buffer with runtime-determined capacity.
+//!
+//! This is a thin wrapper around `VecDeque` that enforces a fixed capacity
+//! set at initialization time. When pushing to a full buffer, the oldest
+//! element is automatically removed.
+
+use std::collections::VecDeque;
+
+/// A circular buffer with a fixed capacity determined at runtime.
+///
+/// Unlike `VecDeque`, this buffer will never grow beyond its initial capacity.
+/// When `push_back` is called on a full buffer, the oldest element is removed first.
+#[derive(Debug)]
+pub struct CircularBuffer<T> {
+    buf: VecDeque<T>,
+    capacity: usize,
+}
+
+impl<T> CircularBuffer<T> {
+    /// Create a new circular buffer with the given capacity.
+    ///
+    /// # Panics
+    /// Panics if capacity is 0.
+    pub fn new(capacity: usize) -> Self {
+        assert!(capacity > 0, "CircularBuffer capacity must be > 0");
+        Self {
+            buf: VecDeque::with_capacity(capacity),
+            capacity,
+        }
+    }
+
+    /// Push an element to the back of the buffer.
+    /// If the buffer is at capacity, the oldest element is removed first.
+    pub fn push_back(&mut self, value: T) {
+        if self.buf.len() == self.capacity {
+            self.buf.pop_front();
+        }
+        self.buf.push_back(value);
+    }
+
+    /// Returns the number of elements in the buffer.
+    pub fn len(&self) -> usize {
+        self.buf.len()
+    }
+
+    /// Returns true if the buffer is empty.
+    pub fn is_empty(&self) -> bool {
+        self.buf.is_empty()
+    }
+
+    /// Returns the capacity of the buffer.
+    pub fn capacity(&self) -> usize {
+        self.capacity
+    }
+
+    /// Returns an iterator over the elements, from oldest to newest.
+    pub fn iter(&self) -> impl DoubleEndedIterator<Item = &T> {
+        self.buf.iter()
+    }
+
+    /// Clears the buffer, removing all elements.
+    pub fn clear(&mut self) {
+        self.buf.clear();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_push_within_capacity() {
+        let mut buf = CircularBuffer::new(3);
+        buf.push_back(1);
+        buf.push_back(2);
+        buf.push_back(3);
+
+        assert_eq!(buf.len(), 3);
+        assert_eq!(buf.iter().copied().collect::<Vec<_>>(), vec![1, 2, 3]);
+    }
+
+    #[test]
+    fn test_push_beyond_capacity() {
+        let mut buf = CircularBuffer::new(3);
+        buf.push_back(1);
+        buf.push_back(2);
+        buf.push_back(3);
+        buf.push_back(4); // Should evict 1
+
+        assert_eq!(buf.len(), 3);
+        assert_eq!(buf.iter().copied().collect::<Vec<_>>(), vec![2, 3, 4]);
+    }
+
+    #[test]
+    fn test_push_many_beyond_capacity() {
+        let mut buf = CircularBuffer::new(3);
+        for i in 1..=10 {
+            buf.push_back(i);
+        }
+
+        assert_eq!(buf.len(), 3);
+        assert_eq!(buf.iter().copied().collect::<Vec<_>>(), vec![8, 9, 10]);
+    }
+
+    #[test]
+    fn test_iter_reverse() {
+        let mut buf = CircularBuffer::new(3);
+        buf.push_back(1);
+        buf.push_back(2);
+        buf.push_back(3);
+
+        assert_eq!(buf.iter().rev().copied().collect::<Vec<_>>(), vec![3, 2, 1]);
+    }
+
+    #[test]
+    fn test_clear() {
+        let mut buf = CircularBuffer::new(3);
+        buf.push_back(1);
+        buf.push_back(2);
+        buf.push_back(3);
+        buf.clear();
+
+        assert!(buf.is_empty());
+        assert_eq!(buf.len(), 0);
+        assert_eq!(buf.capacity(), 3);
+    }
+
+    #[test]
+    fn test_capacity_one() {
+        let mut buf = CircularBuffer::new(1);
+        buf.push_back(1);
+        assert_eq!(buf.iter().copied().collect::<Vec<_>>(), vec![1]);
+
+        buf.push_back(2);
+        assert_eq!(buf.iter().copied().collect::<Vec<_>>(), vec![2]);
+    }
+
+    #[test]
+    #[should_panic(expected = "capacity must be > 0")]
+    fn test_zero_capacity_panics() {
+        let _buf: CircularBuffer<i32> = CircularBuffer::new(0);
+    }
+}

signal-gateway/src/gateway/log_handler.rs

@@ -1,7 +1,7 @@
+use super::circular_buffer::CircularBuffer;
 use super::{AdminMessage, MultiRateLimiter, Origin, RateThreshold, SourceLocationRateLimiter};
 use crate::human_duration::HumanTMinus;
 use chrono::{TimeDelta, Utc};
-use circular_buffer::CircularBuffer;
 use conf::Conf;
 use serde::Deserialize;
 use std::{fmt, time::Duration};
@@ -42,6 +42,9 @@ pub struct LogHandlerConfig {
     /// Structured data ID for tracing metadata (module, file, line) in syslog messages
     #[conf(long, env, default_value = "tracing-meta@64700")]
     pub sd_id: String,
+    /// Number of recent log messages to buffer per origin
+    #[conf(long, env, default_value = "64")]
+    pub log_buffer_size: usize,
 }
 
 /// Specifies both a rate limiting threshold, and criteria for the threshold to apply
@@ -107,7 +110,7 @@ impl AlertRule {
 pub struct LogHandler {
     config: LogHandlerConfig,
     admin_mq_tx: UnboundedSender<AdminMessage>,
-    syslog_buffer: Mutex<CircularBuffer<64, SyslogMessage>>,
+    syslog_buffer: Mutex<CircularBuffer<SyslogMessage>>,
     rate_limiters: Vec<(AlertRule, Mutex<MultiRateLimiter>)>,
     /// Rate limiter keyed by source location (file:line), so different error locations
     /// can alert independently without suppressing each other.
@@ -153,10 +156,12 @@ impl LogHandler {
             OVERALL_LIMITER_MAX_ENTRIES,
         ));
 
+        let syslog_buffer = Mutex::new(CircularBuffer::new(config.log_buffer_size));
+
         Self {
             config,
             admin_mq_tx,
-            syslog_buffer: Default::default(),
+            syslog_buffer,
             rate_limiters,
             overall_limiter,
             any_rule_uses_module,

signal-gateway/src/gateway/mod.rs

@@ -25,6 +25,7 @@ use tokio::{
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};
 
+mod circular_buffer;
 mod log_handler;
 use log_handler::{LogHandler, LogHandlerConfig};