introduce and use new route framework

in this change:

* new "concurrent map" API using RwLock<HashMap> optimized for
  frequent reads.
* we no longer have a concurrent map of log handlers, we instead
  have a log handler containing a concurrent map from origins to
  circular buffers
* we have a general concept of a "Limit" now which replaces AlertRule
* A route consists of a filter, plus any number of limits, which
  can be per-origin limits or global limits
* The set of routes is disjunctive, while the filters and limits are conjunctive
* Limits can also apply per source location
* Different types of rate limiters are unified under the "Limiter"
  abstraction
* The log handler also has a set of overall limiters that can be applied
  at the end
* This means we can relatively easily express disjunctions of conjunctions

I think the route structure will also be relatively easy to express in toml

signal-gateway-bin/src/json/mod.rs

@@ -156,6 +156,7 @@ async fn handle_tcp_connection(stream: TcpStream, gateway: &Gateway) -> std::io:
 /// A flexible JSON log message format compatible with logstash and similar systems.
 ///
 /// Supports various field names and formats commonly used in logging systems.
+#[non_exhaustive]
 #[derive(Debug, Deserialize)]
 pub struct JsonLogMessage {
     /// The log message text - accepts "message" or "msg"
@@ -163,7 +164,7 @@ pub struct JsonLogMessage {
     pub message: String,
 
     /// Log level - accepts various formats (error, ERROR, err, etc.)
-    #[serde(default, alias = "severity", deserialize_with = "deserialize_level")]
+    #[serde(default, alias = "severity", deserialize_with = "deserialize_opt_level")]
     pub level: Option<Level>,
 
     /// Timestamp - accepts Unix epoch seconds (int or string) or RFC3339 string
@@ -228,29 +229,13 @@ impl JsonLogMessage {
     }
 }
 
-/// Deserialize a log level from various string formats
+/// Deserialize an optional log level, returning None for unknown values.
-fn deserialize_level<'de, D>(deserializer: D) -> Result<Option<Level>, D::Error>
+fn deserialize_opt_level<'de, D>(deserializer: D) -> Result<Option<Level>, D::Error>
 where
     D: serde::Deserializer<'de>,
 {
     let opt: Option<String> = Option::deserialize(deserializer)?;
-    Ok(opt.and_then(|s| parse_level(&s)))
+    Ok(opt.and_then(|s| Level::from_str(&s)))
-}
-
-/// Parse a level string into a Level enum
-fn parse_level(s: &str) -> Option<Level> {
-    match s.to_lowercase().as_str() {
-        "emergency" | "emerg" => Some(Level::EMERGENCY),
-        "alert" => Some(Level::ALERT),
-        "critical" | "crit" | "fatal" => Some(Level::CRITICAL),
-        "error" | "err" => Some(Level::ERROR),
-        "warning" | "warn" => Some(Level::WARNING),
-        "notice" => Some(Level::NOTICE),
-        "info" | "information" => Some(Level::INFO),
-        "debug" => Some(Level::DEBUG),
-        "trace" => Some(Level::TRACE),
-        _ => None,
-    }
 }
 
 /// Deserialize a timestamp from Unix epoch (int or string) or RFC3339 string

signal-gateway-bin/src/main.rs

@@ -1,3 +1,7 @@
+//! Signal Gateway binary - receives alerts and logs, forwards to Signal messenger.
+
+#![deny(missing_docs)]
+
 use conf::{Conf, Subcommands};
 use hyper::service::service_fn;
 use hyper_util::rt::TokioIo;
@@ -18,7 +22,7 @@ use admin_netcat::AdminNetcatConfig;
 mod syslog;
 use syslog::SyslogConfig;
 
-mod json;
+pub mod json;
 use json::JsonConfig;
 
 /// Admin message handler configuration - select how non-command messages are handled

signal-gateway/src/concurrent_map.rs

@@ -0,0 +1,146 @@
+//! A concurrent map with read-preferring access pattern.
+//!
+//! This module provides a concurrent hash map that optimizes for the common case
+//! where values already exist, using a read lock first before falling back to a
+//! write lock for insertions.
+
+use std::collections::HashMap;
+use std::hash::Hash;
+use tokio::sync::RwLock;
+
+/// A concurrent hash map that uses read-preferring locking.
+///
+/// When accessing a value, it first tries to acquire a read lock. If the key
+/// exists, it uses the value immediately. If the key doesn't exist, it upgrades
+/// to a write lock and inserts a new value.
+#[derive(Debug)]
+pub struct ConcurrentMap<K, V> {
+    inner: RwLock<HashMap<K, V>>,
+}
+
+impl<K, V> ConcurrentMap<K, V>
+where
+    K: Eq + Hash + Clone,
+{
+    /// Create a new empty concurrent map.
+    pub fn new() -> Self {
+        Self {
+            inner: RwLock::new(HashMap::new()),
+        }
+    }
+
+    /// Get or insert a value, then access it.
+    ///
+    /// This method uses a read-preferring pattern:
+    /// 1. First acquires a read lock and looks for the key
+    /// 2. If found, calls `access` with a reference to the value
+    /// 3. If not found, acquires a write lock, inserts using `create`, then calls `access`
+    ///
+    /// The lock is held while `access` runs, so `access` can safely use the reference.
+    /// For async operations on the value, consider having `access` return a future
+    /// that owns any data it needs.
+    pub async fn get_or_insert_with<R, F, A>(&self, key: K, create: F, access: A) -> R
+    where
+        F: FnOnce() -> V,
+        A: FnOnce(&V) -> R,
+    {
+        // Try to get existing value with read lock first
+        {
+            let guard = self.inner.read().await;
+            if let Some(value) = guard.get(&key) {
+                return access(value);
+            }
+        }
+
+        // Value doesn't exist, need to create with write lock
+        let mut guard = self.inner.write().await;
+        // Use entry API - handles the race where another task inserted while we waited
+        let value = guard.entry(key).or_insert_with(create);
+        access(value)
+    }
+
+    /// Access all entries in the map with a read lock.
+    ///
+    /// Acquires a read lock and calls `access` with a reference to the underlying HashMap.
+    /// The lock is held while `access` runs.
+    pub async fn read_all<R, A>(&self, access: A) -> R
+    where
+        A: FnOnce(&HashMap<K, V>) -> R,
+    {
+        let guard = self.inner.read().await;
+        access(&guard)
+    }
+}
+
+impl<K, V> Default for ConcurrentMap<K, V>
+where
+    K: Eq + Hash + Clone,
+{
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_get_or_insert_new_key() {
+        let map: ConcurrentMap<String, i32> = ConcurrentMap::new();
+
+        let result = map
+            .get_or_insert_with("key1".to_string(), || 42, |v| *v)
+            .await;
+
+        assert_eq!(result, 42);
+    }
+
+    #[tokio::test]
+    async fn test_get_or_insert_existing_key() {
+        let map: ConcurrentMap<String, i32> = ConcurrentMap::new();
+
+        // Insert first time
+        map.get_or_insert_with("key1".to_string(), || 42, |_| ())
+            .await;
+
+        // Access again - should get existing value, not call create
+        let mut create_called = false;
+        let result = map
+            .get_or_insert_with(
+                "key1".to_string(),
+                || {
+                    create_called = true;
+                    100
+                },
+                |v| *v,
+            )
+            .await;
+
+        assert_eq!(result, 42);
+        assert!(!create_called);
+    }
+
+    #[tokio::test]
+    async fn test_get_or_insert_multiple_keys() {
+        let map: ConcurrentMap<String, i32> = ConcurrentMap::new();
+
+        map.get_or_insert_with("a".to_string(), || 1, |_| ()).await;
+        map.get_or_insert_with("b".to_string(), || 2, |_| ()).await;
+        map.get_or_insert_with("c".to_string(), || 3, |_| ()).await;
+
+        let a = map
+            .get_or_insert_with("a".to_string(), || 0, |v| *v)
+            .await;
+        let b = map
+            .get_or_insert_with("b".to_string(), || 0, |v| *v)
+            .await;
+        let c = map
+            .get_or_insert_with("c".to_string(), || 0, |v| *v)
+            .await;
+
+        assert_eq!(a, 1);
+        assert_eq!(b, 2);
+        assert_eq!(c, 3);
+    }
+}

signal-gateway/src/gateway/log_handler.rs

@@ -1,30 +1,45 @@
 use super::circular_buffer::CircularBuffer;
-use super::{AdminMessage, MultiRateLimiter, RateThreshold, SourceLocationRateLimiter};
+use super::route::{Destination, Limit, Route};
+use super::{AdminMessage, LimitResult, Limiter, LimiterSet};
 use crate::{
+    concurrent_map::ConcurrentMap,
     human_duration::HumanTMinus,
-    log_message::{Level, LogFilter, LogMessage, Origin},
+    log_message::{LogMessage, Origin},
 };
 use chrono::{TimeDelta, Utc};
 use conf::Conf;
-use serde::Deserialize;
+use std::fmt;
-use std::{fmt, time::Duration};
 use tokio::sync::{Mutex, mpsc::UnboundedSender};
-use tracing::{error, info, warn};
+use tracing::{error, info};
 
-/// Reason why an alert was suppressed
+/// Reason why an alert was suppressed by rate limiting.
-enum SuppressionReason {
+#[derive(Debug)]
-    /// Suppressed by a configured alert rule (with 0-based rule index)
+pub enum SuppressionReason {
-    Rule(usize),
+    /// No route's filter matched the message.
-    /// Suppressed by the source-location rate limiter
+    NoRoutes,
-    SourceLocation { file: Box<str>, line: Box<str> },
+    /// Suppressed by route limiters. Contains the index and result for each
+    /// route whose filter matched but whose limiter blocked the message.
+    Routes(Vec<(usize, LimitResult)>),
+    /// Suppressed by an overall limiter. Contains the limiter index and result.
+    Overall(usize, LimitResult),
 }
 
 impl fmt::Display for SuppressionReason {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
-            SuppressionReason::Rule(idx) => write!(f, "rule[{idx}]"),
+            SuppressionReason::NoRoutes => write!(f, "no-routes"),
-            SuppressionReason::SourceLocation { file, line } => {
+            SuppressionReason::Routes(failures) => {
-                write!(f, "source-location({file}:{line})")
+                write!(f, "routes[")?;
+                for (i, (idx, result)) in failures.iter().enumerate() {
+                    if i > 0 {
+                        write!(f, ", ")?;
+                    }
+                    write!(f, "{idx}:{result:?}")?;
+                }
+                write!(f, "]")
+            }
+            SuppressionReason::Overall(idx, result) => {
+                write!(f, "overall[{idx}]:{result:?}")
             }
         }
     }
@@ -33,10 +48,12 @@ impl fmt::Display for SuppressionReason {
 /// Config options related to the log handler, and what log messages it chooses to alert on.
 #[derive(Clone, Conf, Debug)]
 pub struct LogHandlerConfig {
-    #[conf(long, env, value_parser = serde_json::from_str)]
+    /// Routes for matching and rate-limiting log messages.
-    pub alert_rate_limits: Vec<AlertRule>,
+    #[conf(long, env, value_parser = serde_json::from_str, default_value = "[]")]
-    #[conf(long, env, default_value = "10m", value_parser = conf_extra::parse_duration)]
+    pub routes: Vec<Route>,
-    pub overall_alert_limit: Duration,
+    /// Overall rate limits applied after route checks pass.
+    #[conf(long, env, value_parser = serde_json::from_str, default_value = "[]")]
+    pub overall_limits: Vec<Limit>,
     #[conf(long, env)]
     pub format_module: bool,
     #[conf(long, env)]
@@ -46,149 +63,154 @@ pub struct LogHandlerConfig {
     pub log_buffer_size: usize,
 }
 
-/// Specifies both a rate limiting threshold, and criteria for the threshold to apply
+/// The log handler takes log messages and decides what to do with them.
-#[derive(Clone, Debug, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct AlertRule {
-    #[serde(flatten)]
-    pub filter: LogFilter,
-    pub threshold: RateThreshold,
-}
-
-/// The log handler takes log messages from a single origin and decides what
-/// to do with them.
 ///
-/// 1. Store them in a small circular buffer
+/// 1. Store them in a small circular buffer (per origin)
 /// 2. If it is an error, and meets other criteria, trigger an alert,
 ///    i.e. send a message to admins containing this log and other recent logs.
 /// 3. The maximum rate of alerts can also be configured.
 ///
 /// Additionally, the log handler can format the buffer of recent logs into a string,
 /// if requested.
-///
-/// Each origin (app + host pair) gets its own LogHandler instance, managed by the Gateway.
 #[derive(Debug)]
 pub struct LogHandler {
     config: LogHandlerConfig,
     admin_mq_tx: UnboundedSender<AdminMessage>,
-    log_buffer: Mutex<CircularBuffer<LogMessage>>,
+    /// Log buffers keyed by origin (app + host). Lazily created.
-    rate_limiters: Vec<(AlertRule, Mutex<MultiRateLimiter>)>,
+    log_buffers: ConcurrentMap<Origin, Mutex<CircularBuffer<LogMessage>>>,
-    /// Rate limiter keyed by source location (file:line), so different error locations
+    /// Routes with their associated limiter sets.
-    /// can alert independently without suppressing each other.
+    routes: Vec<(Route, Mutex<LimiterSet>)>,
-    overall_limiter: Mutex<SourceLocationRateLimiter>,
+    /// Overall rate limiters applied after route checks pass.
-    /// True if any configured rule uses the module structured data field.
+    overall_limits: Vec<Mutex<Limiter>>,
-    any_rule_uses_module: bool,
-    /// True if any configured rule uses the file structured data field.
-    any_rule_uses_file: bool,
-    /// True if any configured rule uses the line structured data field.
-    any_rule_uses_line: bool,
 }
 
-/// Maximum entries in the source-location rate limiter before triggering cleanup
-const OVERALL_LIMITER_MAX_ENTRIES: usize = 2000;
-
 impl LogHandler {
     /// Initialize a new log handler
     pub fn new(config: LogHandlerConfig, admin_mq_tx: UnboundedSender<AdminMessage>) -> Self {
-        let any_rule_uses_module = config
+        let routes = config
-            .alert_rate_limits
+            .routes
             .iter()
-            .any(|r| r.filter.uses_module());
+            .map(|route| (route.clone(), Mutex::new(route.make_limiter_set())))
-        let any_rule_uses_file = config
+            .collect();
-            .alert_rate_limits
-            .iter()
-            .any(|r| r.filter.uses_file());
-        let any_rule_uses_line = config
-            .alert_rate_limits
-            .iter()
-            .any(|r| r.filter.uses_line());
-        let rate_limiters = config
-            .alert_rate_limits
-            .iter()
-            .map(|rule| {
-                (
-                    rule.clone(),
-                    Mutex::new(MultiRateLimiter::from(rule.threshold)),
-                )
-            })
-            .collect::<Vec<_>>();
-        let overall_limiter = Mutex::new(SourceLocationRateLimiter::new(
-            config.overall_alert_limit,
-            OVERALL_LIMITER_MAX_ENTRIES,
-        ));
 
-        let log_buffer = Mutex::new(CircularBuffer::new(config.log_buffer_size));
+        let overall_limits = config
+            .overall_limits
+            .iter()
+            .map(|limit| Mutex::new(limit.make_limiter()))
+            .collect();
 
         Self {
             config,
             admin_mq_tx,
-            log_buffer,
+            log_buffers: ConcurrentMap::new(),
-            rate_limiters,
+            routes,
-            overall_limiter,
+            overall_limits,
-            any_rule_uses_module,
-            any_rule_uses_file,
-            any_rule_uses_line,
         }
     }
 
-    /// Format recent logs into a string
+    /// Format recent logs into a string for all origins, optionally filtered.
-    pub async fn format_logs(&self) -> String {
+    ///
-        let lk = self.log_buffer.lock().await;
+    /// If `filter` is provided, only origins matching the filter are included.
+    pub async fn format_logs(&self, filter: Option<&str>) -> String {
+        self.log_buffers
+            .read_all(|buffers| {
+                if buffers.is_empty() {
+                    return "No log sources registered yet".to_string();
+                }
 
-        let mut text = format!("{} log messages (newest first):\n", lk.len());
+                let mut text = String::new();
+                let now = Utc::now().timestamp();
 
-        // Calculate now once for consistent relative timestamps
+                for (origin, buffer_mutex) in buffers.iter() {
-        let now = Utc::now().timestamp();
+                    // Apply filter if present
+                    if let Some(f) = filter {
+                        if !origin.matches_filter(f) {
+                            continue;
+                        }
+                    }
 
-        // Collect and reverse to show newest first
+                    // We can't await inside read_all, so use try_lock
-        let messages: Vec<_> = lk.iter().collect();
+                    // If the buffer is locked, skip it (rare case)
-        for log_msg in messages.into_iter().rev() {
+                    let Some(buffer) = buffer_mutex.try_lock().ok() else {
-            self.write_log_msg(&mut text, log_msg, now);
+                        continue;
-        }
+                    };
-        text
+
+                    use std::fmt::Write;
+                    writeln!(&mut text, "=== [{origin}] ===").unwrap();
+                    writeln!(&mut text, "{} log messages (newest first):", buffer.len()).unwrap();
+
+                    // Collect and reverse to show newest first
+                    let messages: Vec<_> = buffer.iter().collect();
+                    for log_msg in messages.into_iter().rev() {
+                        self.write_log_msg(&mut text, log_msg, now);
+                    }
+                    text.push('\n');
+                }
+
+                if text.is_empty() {
+                    "No matching log sources".to_string()
+                } else {
+                    text
+                }
+            })
+            .await
     }
 
     /// Consume a new log message from the given origin
     pub async fn handle_log_message(&self, mut log_msg: LogMessage, origin: Origin) {
-        let suppression_reason = self.check_suppression(&mut log_msg).await;
+        let ts_sec = *log_msg
-        if let Some(reason) = &suppression_reason
+            .timestamp
-            && log_msg.level <= Level::ERROR
+            .get_or_insert_with(|| Utc::now().timestamp());
-        {
+
+        let rate_limit_result = self.check_rate_limiters(&log_msg, &origin, ts_sec).await;
+
+        if let Err(reason) = &rate_limit_result {
             let sev = log_msg.level.to_str();
             info!("Suppressed {sev} ({reason}):\n{}", log_msg.msg);
         }
 
-        // Record this new message.
+        let buffer_size = self.config.log_buffer_size;
-        // Then, if we should alert now, also format the whole buffer to a string,
+
-        // and then release the lock.
+        // Get or create the buffer for this origin, then record the message
-        let formatted_text = {
+        let formatted_text = self
-            let mut lk = self.log_buffer.lock().await;
+            .log_buffers
-            lk.push_back(log_msg);
+            .get_or_insert_with(
-            if suppression_reason.is_some() {
+                origin.clone(),
-                return;
+                || Mutex::new(CircularBuffer::new(buffer_size)),
+                |buffer_mutex| {
+                    // Lock the buffer and process the message
+                    let mut buffer = buffer_mutex.blocking_lock();
+                    buffer.push_back(log_msg);
+
+                    if rate_limit_result.is_err() {
+                        return None;
+                    }
+
+                    let mut text = String::default();
+                    let now = Utc::now().timestamp();
+
+                    // Iterate in reverse (newest first) without copying
+                    for log_msg in buffer.iter().rev() {
+                        self.write_log_msg(&mut text, log_msg, now);
+                    }
+                    buffer.clear();
+
+                    Some(text)
+                },
+            )
+            .await;
+
+        // Send alert if we have formatted text
+        if let Some(text) = formatted_text {
+            // TODO: Use destination override from rate_limit_result.ok() if present
+            if let Err(_err) = self.admin_mq_tx.send(AdminMessage {
+                origin: Some(origin),
+                text,
+                attachment_paths: Default::default(),
+                summary: None,
+            }) {
+                error!("Could not send alert message, queue is closed");
             }
-
-            let mut text = String::default();
-
-            // Calculate now once for consistent relative timestamps
-            let now = Utc::now().timestamp();
-
-            // Iterate in reverse (newest first) without copying
-            for log_msg in lk.iter().rev() {
-                self.write_log_msg(&mut text, log_msg, now);
-            }
-            lk.clear();
-
-            text
-        };
-
-        if let Err(_err) = self.admin_mq_tx.send(AdminMessage {
-            origin: Some(origin),
-            text: formatted_text,
-            attachment_paths: Default::default(),
-            summary: None,
-        }) {
-            error!("Could not send alert message, queue is closed");
         }
     }
 
@@ -251,63 +273,75 @@ impl LogHandler {
         }
     }
 
-    /// Check if an alert should be suppressed for a given error message.
+    /// Check if a log message passes all rate limiters.
     ///
-    /// Returns `None` if the alert should fire, or `Some(reason)` if suppressed.
+    /// Tests the message against each route's filter in succession (no early return).
-    async fn check_suppression(&self, log_msg: &mut LogMessage) -> Option<SuppressionReason> {
+    /// For routes where the filter matches, evaluates the limiter set.
-        let ts_sec = *log_msg
+    ///
-            .timestamp
+    /// Returns:
-            .get_or_insert_with(|| Utc::now().timestamp());
+    /// - `Ok(Some(destination))` if passed and a route specified a destination override
+    /// - `Ok(None)` if passed with no destination override
+    /// - `Err(SuppressionReason::Routes(...))` if no route's limiter passed
+    /// - `Err(SuppressionReason::Overall(...))` if routes passed but overall limiter failed
+    async fn check_rate_limiters(
+        &self,
+        log_msg: &LogMessage,
+        origin: &Origin,
+        ts_sec: i64,
+    ) -> Result<Option<Destination>, SuppressionReason> {
+        let mut route_failures: Vec<(usize, LimitResult)> = Vec::new();
+        let mut first_passed_destination: Option<Option<Destination>> = None;
 
-        let high_severity = log_msg.level <= Level::ERROR;
+        // Test against each route's filter and limiter
-        if !high_severity {
+        for (idx, (route, limiter_set)) in self.routes.iter().enumerate() {
-            // Low severity messages are always "suppressed" (not alerted on)
+            // Check if message level meets route's alert threshold
-            // but we don't need to log a reason for this
+            if log_msg.level > route.alert_level {
-            return Some(SuppressionReason::Rule(usize::MAX));
+                continue;
-        }
+            }
 
-        // Warn if rules expect structured data but the message doesn't have it
+            // Check if message matches route's filter (if any)
-        let has_module = log_msg.module_path.is_some();
+            let filter_matches = route
-        let has_file = log_msg.file.is_some();
+                .filter
-        let has_line = log_msg.line.is_some();
+                .as_ref()
-        if (self.any_rule_uses_module && !has_module)
+                .map_or(true, |f| f.matches(log_msg));
-            || (self.any_rule_uses_file && !has_file)
-            || (self.any_rule_uses_line && !has_line)
-        {
-            warn!(
-                "Log message missing source location data, filtering rules may not work: {log_msg:#?}"
-            );
-        }
 
-        // Check each configured rule - track which rule suppressed the alert
+            if !filter_matches {
-        // Note: we check all rules even if one already suppressed, to update all rate limiters
+                continue;
-        let mut suppressed_by_rule: Option<usize> = None;
+            }
-        for (idx, (rule, limiter)) in self.rate_limiters.iter().enumerate() {
+
-            if rule.filter.matches(log_msg) && !limiter.lock().await.evaluate(ts_sec) {
+            // Filter matched, evaluate the limiter set
-                suppressed_by_rule.get_or_insert(idx);
+            let result = limiter_set.lock().await.evaluate(log_msg, origin, ts_sec);
+
+            match result {
+                LimitResult::Passed => {
+                    // Remember the first route that passed
+                    if first_passed_destination.is_none() {
+                        first_passed_destination = Some(route.destination.clone());
+                    }
+                }
+                _ => {
+                    // Record the failure
+                    route_failures.push((idx, result));
+                }
             }
         }
-        if let Some(idx) = suppressed_by_rule {
+
-            return Some(SuppressionReason::Rule(idx));
+        // If no route passed, return the appropriate error
+        let first_destination = match first_passed_destination {
+            Some(dest) => dest,
+            None if route_failures.is_empty() => return Err(SuppressionReason::NoRoutes),
+            None => return Err(SuppressionReason::Routes(route_failures)),
+        };
+
+        // At least one route passed, now check overall limits
+        for (idx, limiter) in self.overall_limits.iter().enumerate() {
+            if !limiter.lock().await.evaluate(log_msg, ts_sec) {
+                return Err(SuppressionReason::Overall(idx, LimitResult::Limiter(0)));
+            }
         }
 
-        // Extract source location for per-location rate limiting
+        // All checks passed
-        let file = log_msg.file.as_deref().unwrap_or("?");
+        Ok(first_destination)
-        let line = log_msg.line.as_deref().unwrap_or("?");
-
-        if !self
-            .overall_limiter
-            .lock()
-            .await
-            .evaluate(file, line, ts_sec)
-        {
-            return Some(SuppressionReason::SourceLocation {
-                file: file.into(),
-                line: line.into(),
-            });
-        }
-
-        None // Alert should fire
     }
 }
 

signal-gateway/src/gateway/mod.rs

@@ -25,7 +25,7 @@ use std::{collections::HashMap, fmt::Write, net::SocketAddr, path::PathBuf};
 use tokio::{
     join,
     sync::{
-        Mutex, RwLock,
+        Mutex,
         mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel},
     },
 };
@@ -37,8 +37,11 @@ mod circular_buffer;
 mod log_handler;
 use log_handler::{LogHandler, LogHandlerConfig};
 
+mod route;
+pub use route::{Destination, Limit, RateThreshold, Route};
+
 mod rate_limiter;
-use rate_limiter::{MultiRateLimiter, RateThreshold, SourceLocationRateLimiter};
+pub use rate_limiter::{LimitResult, Limiter, LimiterSet};
 
 /// Configuration for the gateway.
 #[derive(Conf, Debug)]
@@ -186,8 +189,8 @@ pub struct Gateway {
     admin_mq_rx: Mutex<UnboundedReceiver<AdminMessage>>,
     token: CancellationToken,
     prometheus: Option<Prometheus>,
-    /// Log handlers keyed by origin (app + host). Lazily created when first message from an origin arrives.
+    /// Log handler for processing log messages from all origins.
-    log_handlers: RwLock<HashMap<Origin, LogHandler>>,
+    log_handler: LogHandler,
     /// Handler for admin messages that don't start with `/`
     message_handler: Option<Box<dyn MessageHandler>>,
 }
@@ -208,13 +211,15 @@ impl Gateway {
             .transpose()
             .expect("Invalid prometheus config");
 
+        let log_handler = LogHandler::new(config.log_handler.clone(), admin_mq_tx.clone());
+
         Self {
             config,
             admin_mq_tx,
             admin_mq_rx: Mutex::new(admin_mq_rx),
             token,
             prometheus,
-            log_handlers: RwLock::new(HashMap::new()),
+            log_handler,
             message_handler,
         }
     }
@@ -558,25 +563,10 @@ impl Gateway {
     async fn handle_gateway_command(&self, cmd: GatewayCommand) -> MessageHandlerResult {
         match cmd {
             GatewayCommand::Log { filter } => {
-                let handlers = self.log_handlers.read().await;
+                let text = self
-                if handlers.is_empty() {
+                    .log_handler
-                    return Ok(AdminMessageResponse::new("No log sources registered yet"));
+                    .format_logs(filter.as_deref())
-                }
+                    .await;
-                let mut text = String::new();
-                for (origin, handler) in handlers.iter() {
-                    // Apply filter if present
-                    if let Some(ref f) = filter
-                        && !origin.matches_filter(f)
-                    {
-                        continue;
-                    }
-                    writeln!(&mut text, "=== [{origin}] ===").unwrap();
-                    text.push_str(&handler.format_logs().await);
-                    text.push('\n');
-                }
-                if text.is_empty() {
-                    return Ok(AdminMessageResponse::new("No matching log sources"));
-                }
                 Ok(AdminMessageResponse::new(text))
             }
             GatewayCommand::Query { query } => {
@@ -826,24 +816,7 @@ impl Gateway {
     pub async fn handle_log_message(&self, log_msg: impl Into<LogMessage>) {
         let log_msg = log_msg.into();
         let origin = Origin::from(&log_msg);
-
+        self.log_handler.handle_log_message(log_msg, origin).await;
-        // Try to get existing handler with read lock first
-        {
-            let handlers = self.log_handlers.read().await;
-            if let Some(handler) = handlers.get(&origin) {
-                handler.handle_log_message(log_msg, origin).await;
-                return;
-            }
-        }
-
-        // Handler doesn't exist, need to create one with write lock
-        let mut handlers = self.log_handlers.write().await;
-        // Double-check in case another task created it while we were waiting for the write lock
-        let handler = handlers.entry(origin.clone()).or_insert_with(|| {
-            info!("Creating new log handler for origin: {origin}");
-            LogHandler::new(self.config.log_handler.clone(), self.admin_mq_tx.clone())
-        });
-        handler.handle_log_message(log_msg, origin).await;
     }
 }
 

signal-gateway/src/gateway/rate_limiter.rs

@@ -1,72 +1,113 @@
-use serde::Deserialize;
+use super::route::{Limit, RateThreshold};
+use crate::log_message::{LogMessage, Origin};
 use std::{
     collections::HashMap,
-    str::FromStr,
     sync::atomic::{AtomicI64, Ordering},
     time::Duration,
 };
 
-/// Represents a rate threshold, expressed as a string in the format:
+/// Maximum entries in a source-location rate limiter before triggering cleanup.
-///
+const SOURCE_LOCATION_MAX_ENTRIES: usize = 2000;
-/// * `1 / 10s`
+
-/// * `2 / 5m`
+/// A rate limiter that can be either a multi-rate limiter or a source-location limiter.
-/// * `3 / 1h`
+#[derive(Debug)]
-/// * `> 1 / 10s`
+pub enum Limiter {
-/// * `>= 2 / 10s`
+    /// Counts events regardless of source location.
-///
+    Multi(MultiRateLimiter),
-/// When the comparator is omitted, it is treated as `>=`
+    /// Tracks events independently per source location (file:line).
-#[derive(Clone, Copy, Debug, Deserialize)]
+    SourceLocation(SourceLocationRateLimiter),
-#[serde(try_from = "String")]
-pub struct RateThreshold {
-    pub times: usize,
-    pub duration: Duration,
 }
 
-impl FromStr for RateThreshold {
+/// Result of evaluating a limiter set.
-    type Err = String;
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum LimitResult {
+    /// The event passed all limits (not rate-limited).
+    Passed,
+    /// The event was blocked by a per-origin limiter at the given index.
+    Limiter(usize),
+    /// The event was blocked by a global limiter at the given index.
+    GlobalLimiter(usize),
+}
 
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
+/// A set of limiters for a route, containing both per-origin and global limiters.
-        let Some((first, second)) = s.trim().split_once('/') else {
+#[derive(Debug)]
-            return Err("missing '/' character in rate threshold".into());
+pub struct LimiterSet {
-        };
+    /// Limit configurations (used to create limiters for new origins).
+    limits: Vec<Limit>,
+    /// Per-origin rate limiters, keyed by origin. Lazily created.
+    limiters: HashMap<Origin, Vec<Limiter>>,
+    /// Global rate limiters (shared across all origins).
+    global_limiters: Vec<Limiter>,
+}
 
-        let duration = conf_extra::parse_duration(second.trim())?;
+impl LimiterSet {
+    /// Create a new limiter set from limit configurations.
+    pub fn new(limits: Vec<Limit>, global_limits: Vec<Limit>) -> Self {
+        Self {
+            limits,
+            limiters: HashMap::new(),
+            global_limiters: global_limits.iter().map(|l| l.make_limiter()).collect(),
+        }
+    }
 
-        let first = first.trim();
+    /// Evaluate whether an event should pass all rate limits.
-        let maybe_mid = first.as_bytes().iter().position(|b| b.is_ascii_digit());
+    ///
-        let (comparator, num) = if let Some(mid) = maybe_mid {
+    /// Returns [`LimitResult::Passed`] if the event passes all limits.
-            first.split_at(mid)
+    /// Returns [`LimitResult::Limiter(i)`] if blocked by per-origin limiter at index `i`.
-        } else {
+    /// Returns [`LimitResult::GlobalLimiter(i)`] if blocked by global limiter at index `i`.
-            ("", first)
+    pub fn evaluate(&mut self, log_msg: &LogMessage, origin: &Origin, ts_sec: i64) -> LimitResult {
-        };
+        // Get or create limiters for this origin
+        let origin_limiters = self
+            .limiters
+            .entry(origin.clone())
+            .or_insert_with(|| self.limits.iter().map(|l| l.make_limiter()).collect());
 
-        let is_greater_equal = match comparator.trim() {
+        for (i, limiter) in origin_limiters.iter_mut().enumerate() {
-            ">" => false,
+            if !limiter.evaluate(log_msg, ts_sec) {
-            ">=" | "=>" | "" => true,
+                return LimitResult::Limiter(i);
-            _ => return Err(format!("Unexpected comparator format: {comparator}")),
+            }
-        };
-
-        let num = num.trim();
-        let mut times: usize = num
-            .parse()
-            .map_err(|err| format!("invalid number {num}: {err}"))?;
-
-        if !is_greater_equal {
-            times += 1;
         }
 
-        if times == 0 {
+        for (i, limiter) in self.global_limiters.iter_mut().enumerate() {
-            return Err("Invalid threshold, times must be > 0".into());
+            if !limiter.evaluate(log_msg, ts_sec) {
+                return LimitResult::GlobalLimiter(i);
+            }
         }
 
-        Ok(RateThreshold { times, duration })
+        LimitResult::Passed
     }
 }
 
-impl TryFrom<String> for RateThreshold {
+impl Limiter {
-    type Error = <RateThreshold as FromStr>::Err;
+    /// Evaluate whether an event should pass the rate limit.
-    fn try_from(s: String) -> Result<Self, Self::Error> {
+    ///
-        RateThreshold::from_str(&s)
+    /// Returns `true` if the event should be allowed (not rate-limited),
+    /// `false` if it should be suppressed.
+    pub fn evaluate(&mut self, log_msg: &LogMessage, ts_sec: i64) -> bool {
+        match self {
+            Limiter::Multi(limiter) => limiter.evaluate(ts_sec),
+            Limiter::SourceLocation(limiter) => {
+                let file = log_msg.file.as_deref().unwrap_or("?");
+                let line = log_msg.line.as_deref().unwrap_or("?");
+                limiter.evaluate(file, line, ts_sec)
+            }
+        }
+    }
+
+    /// Create a multi-rate limiter from a threshold.
+    pub fn multi(threshold: RateThreshold) -> Self {
+        Limiter::Multi(MultiRateLimiter::from(threshold))
+    }
+
+    /// Create a source-location limiter from a threshold.
+    ///
+    /// Note: Only the duration is used; source-location limiters allow one event
+    /// per location per window.
+    pub fn source_location(threshold: RateThreshold) -> Self {
+        Limiter::SourceLocation(SourceLocationRateLimiter::new(
+            threshold.duration,
+            SOURCE_LOCATION_MAX_ENTRIES,
+        ))
     }
 }
 
@@ -210,6 +251,7 @@ impl From<RateThreshold> for MultiRateLimiter {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::str::FromStr;
 
     #[test]
     fn parse_rate_threshold() {

signal-gateway/src/gateway/route.rs

@@ -0,0 +1,170 @@
+//! Route configuration for log message handling.
+//!
+//! Routes define how log messages are processed based on filters, severity levels,
+//! and destination overrides.
+
+use crate::log_message::{Level, LogFilter};
+use serde::Deserialize;
+use std::{str::FromStr, time::Duration};
+
+/// Represents a rate threshold, expressed as a string in the format:
+///
+/// * `1 / 10s`
+/// * `2 / 5m`
+/// * `3 / 1h`
+/// * `> 1 / 10s`
+/// * `>= 2 / 10s`
+///
+/// When the comparator is omitted, it is treated as `>=`
+#[derive(Clone, Copy, Debug, Deserialize)]
+#[serde(try_from = "String")]
+pub struct RateThreshold {
+    /// Number of events required to trigger the threshold.
+    pub times: usize,
+    /// Time window for counting events.
+    pub duration: Duration,
+}
+
+impl FromStr for RateThreshold {
+    type Err = String;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let Some((first, second)) = s.trim().split_once('/') else {
+            return Err("missing '/' character in rate threshold".into());
+        };
+
+        let duration = conf_extra::parse_duration(second.trim())?;
+
+        let first = first.trim();
+        let maybe_mid = first.as_bytes().iter().position(|b| b.is_ascii_digit());
+        let (comparator, num) = if let Some(mid) = maybe_mid {
+            first.split_at(mid)
+        } else {
+            ("", first)
+        };
+
+        let is_greater_equal = match comparator.trim() {
+            ">" => false,
+            ">=" | "=>" | "" => true,
+            _ => return Err(format!("Unexpected comparator format: {comparator}")),
+        };
+
+        let num = num.trim();
+        let mut times: usize = num
+            .parse()
+            .map_err(|err| format!("invalid number {num}: {err}"))?;
+
+        if !is_greater_equal {
+            times += 1;
+        }
+
+        if times == 0 {
+            return Err("Invalid threshold, times must be > 0".into());
+        }
+
+        Ok(RateThreshold { times, duration })
+    }
+}
+
+impl TryFrom<String> for RateThreshold {
+    type Error = <RateThreshold as FromStr>::Err;
+    fn try_from(s: String) -> Result<Self, Self::Error> {
+        RateThreshold::from_str(&s)
+    }
+}
+
+/// A route configuration for processing log messages.
+///
+/// Routes match incoming log messages based on an optional filter, then apply
+/// the configured alert level threshold. Each route can optionally override
+/// the default destination and define rate limits.
+#[derive(Clone, Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Route {
+    /// Minimum severity level for messages to trigger an alert.
+    /// Messages at this level or higher (lower numeric value) will alert.
+    /// Default: ERROR
+    #[serde(default = "default_alert_level")]
+    pub alert_level: Level,
+
+    /// Optional filter to match log messages for this route.
+    /// If not specified, the route matches all messages.
+    #[serde(default)]
+    pub filter: Option<LogFilter>,
+
+    /// Optional destination override for alerts from this route.
+    /// If not specified, alerts go to the default admin destination.
+    #[serde(default)]
+    pub destination: Option<Destination>,
+
+    /// Rate limits applied per-origin for messages matching this route.
+    /// Each limit specifies a filter and threshold for suppressing repeated alerts.
+    #[serde(default)]
+    pub limit: Vec<Limit>,
+
+    /// Global rate limits applied across all origins for this route.
+    /// Each limit specifies a filter and threshold for suppressing repeated alerts.
+    #[serde(default)]
+    pub global_limit: Vec<Limit>,
+}
+
+fn default_alert_level() -> Level {
+    Level::ERROR
+}
+
+/// Destination override for alert messages.
+#[derive(Clone, Debug, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum Destination {
+    /// Send to specific recipient UUIDs.
+    Recipients(Vec<String>),
+    /// Send to a Signal group by group ID.
+    Group(String),
+}
+
+/// A rate limit rule for suppressing repeated alerts.
+///
+/// Combines a filter to match specific log messages with a rate threshold.
+#[derive(Clone, Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct Limit {
+    /// Filter criteria for messages this limit applies to.
+    #[serde(flatten)]
+    pub filter: LogFilter,
+    /// Rate threshold for suppressing alerts.
+    pub threshold: RateThreshold,
+    /// If true, rate limit independently per source location (file:line).
+    /// If false (default), count all matching events together.
+    #[serde(default)]
+    pub by_source_location: bool,
+}
+
+impl Limit {
+    /// Create the appropriate limiter for this limit configuration.
+    pub fn make_limiter(&self) -> super::rate_limiter::Limiter {
+        if self.by_source_location {
+            super::rate_limiter::Limiter::source_location(self.threshold)
+        } else {
+            super::rate_limiter::Limiter::multi(self.threshold)
+        }
+    }
+}
+
+impl Route {
+    /// Create a limiter set from this route's limit configurations.
+    pub fn make_limiter_set(&self) -> super::rate_limiter::LimiterSet {
+        super::rate_limiter::LimiterSet::new(self.limit.clone(), self.global_limit.clone())
+    }
+}
+
+impl Default for Route {
+    fn default() -> Self {
+        Self {
+            alert_level: default_alert_level(),
+            filter: None,
+            destination: None,
+            limit: Vec::new(),
+            global_limit: Vec::new(),
+        }
+    }
+}

signal-gateway/src/lib.rs

@@ -9,6 +9,7 @@ pub mod alertmanager;
 pub mod gateway;
 pub mod message_handler;
 
+pub(crate) mod concurrent_map;
 pub(crate) mod human_duration;
 pub(crate) mod jsonrpc;
 pub(crate) mod log_message;

signal-gateway/src/log_message.rs

@@ -1,6 +1,6 @@
 //! Log message schema and types.
 
-use serde::Deserialize;
+use serde::{Deserialize, de};
 
 /// Log severity level, following syslog conventions.
 ///
@@ -44,6 +44,48 @@ impl Level {
             Self::TRACE => "TRACE",
         }
     }
+
+    /// Parse a level from a string (case-insensitive).
+    ///
+    /// Accepts various common aliases:
+    /// - emergency, emerg
+    /// - alert
+    /// - critical, crit, fatal
+    /// - error, err
+    /// - warning, warn
+    /// - notice, note
+    /// - info, information
+    /// - debug
+    /// - trace
+    pub fn from_str(s: &str) -> Option<Self> {
+        match s.to_ascii_uppercase().as_str() {
+            "EMERGENCY" | "EMERG" => Some(Self::EMERGENCY),
+            "ALERT" => Some(Self::ALERT),
+            "CRITICAL" | "CRIT" | "FATAL" => Some(Self::CRITICAL),
+            "ERROR" | "ERR" => Some(Self::ERROR),
+            "WARNING" | "WARN" => Some(Self::WARNING),
+            "NOTICE" | "NOTE" => Some(Self::NOTICE),
+            "INFO" | "INFORMATION" => Some(Self::INFO),
+            "DEBUG" => Some(Self::DEBUG),
+            "TRACE" => Some(Self::TRACE),
+            _ => None,
+        }
+    }
+}
+
+impl<'de> Deserialize<'de> for Level {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: de::Deserializer<'de>,
+    {
+        let s = String::deserialize(deserializer)?;
+        Level::from_str(&s).ok_or_else(|| {
+            de::Error::custom(format!(
+                "unknown log level '{}', expected one of: emergency, alert, critical, error, warning, notice, info, debug, trace",
+                s
+            ))
+        })
+    }
 }
 
 /// A structured log message.