diff --git a/Cargo.lock b/Cargo.lock index 77c8b13..8a7b06c 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index 69b3825..39b29e0 100644 --- a/Cargo.toml +++ b/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" diff --git a/signal-gateway/Cargo.toml b/signal-gateway/Cargo.toml index 5c199d0..31a721b 100644 --- a/signal-gateway/Cargo.toml +++ b/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 } diff --git a/signal-gateway/src/gateway/circular_buffer.rs b/signal-gateway/src/gateway/circular_buffer.rs new file mode 100644 index 0000000..f6620d0 --- /dev/null +++ b/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 { + buf: VecDeque, + capacity: usize, +} + +impl CircularBuffer { + /// 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 { + 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![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![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![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![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![1]); + + buf.push_back(2); + assert_eq!(buf.iter().copied().collect::>(), vec![2]); + } + + #[test] + #[should_panic(expected = "capacity must be > 0")] + fn test_zero_capacity_panics() { + let _buf: CircularBuffer = CircularBuffer::new(0); + } +} diff --git a/signal-gateway/src/gateway/log_handler.rs b/signal-gateway/src/gateway/log_handler.rs index 24916eb..9e2fa2e 100644 --- a/signal-gateway/src/gateway/log_handler.rs +++ b/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, - syslog_buffer: Mutex>, + syslog_buffer: Mutex>, rate_limiters: Vec<(AlertRule, Mutex)>, /// 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, diff --git a/signal-gateway/src/gateway/mod.rs b/signal-gateway/src/gateway/mod.rs index 22d08ec..18565b6 100644 --- a/signal-gateway/src/gateway/mod.rs +++ b/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};