feat: replace notification batching with debouncing (#58)

Instead of batching, where notifications may be delayed by up to 24
hours, we now send notifications about check failures immediately,
avoiding sending multiple notifications within a 24-hour period to the
same member.

Author: kdkasad

Cargo.toml

@@ -39,7 +39,7 @@ serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.141"
 tera = { version = "1.20.0", default-features = false, features = [ "urlencode" ] }
 thiserror = { version = "2.0.12", default-features = false }
-tokio = { version = "1.47.0", features = ["macros", "rt-multi-thread", "test-util"] }
+tokio = { version = "1.47.0", features = ["macros", "rt-multi-thread"] }
 toml = { version = "0.9.2", default-features = false, features = ["preserve_order", "parse", "serde", "std"] }
 tower-http = { version = "0.6.6", features = ["cors", "fs", "catch-panic", "trace"] }
 tracing = "0.1.41"
@@ -52,3 +52,4 @@ httpmock = { version = "0.7.0", default-features = false }
 pretty_assertions = "1.4.1"
 tempfile = "3.20.0"
 tower = "0.5.2"
+tokio = { version = "1.47.0", features = ["macros", "rt-multi-thread", "test-util"] }

src/discord.rs

@@ -4,25 +4,19 @@
 //!
 //! [webhook-api-ref]: https://discord.com/developers/docs/resources/webhook#execute-webhook
 
-use std::{
-    collections::HashMap,
-    fmt::Display,
-    str::FromStr,
-    sync::{Mutex, MutexGuard},
-    time::Duration,
-};
+use std::{fmt::Display, str::FromStr, time::Duration};
 
-use eyre::{Context, bail};
+use eyre::{Context, eyre};
 use reqwest::{
     Client, Response, StatusCode, Url,
     header::{self, HeaderValue},
 };
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use tracing::{Instrument, error, info, info_span, instrument, warn};
+use tracing::{error, instrument, warn};
 
-/// Maximum number of attempts to send a message before giving up.
-const MAX_DELIVERY_ATTEMPTS: usize = 5;
+/// The period within which at most one notification can be sent to each member about their site.
+pub const NOTIFICATION_DEBOUNCE_PERIOD: Duration = Duration::from_secs(60 * 60 * 24); // 24 hours
 
 /// A user agent representing our program. [Required by Discord][api-doc-ua].
 ///
@@ -69,15 +63,12 @@ impl Display for Snowflake {
 }
 
 /// A notifier that sends messages to a Discord channel using a webhook URL.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct DiscordNotifier {
     /// The webhook URL to send messages to.
     webhook_url: reqwest::Url,
     /// HTTP client used to send requests.
     client: Client,
-    /// Set of pending messages to be sent.
-    /// Map from recipient -> (message, attempt count).
-    message_queue: Mutex<HashMap<Option<Snowflake>, (String, usize)>>,
 }
 
 impl DiscordNotifier {
@@ -87,102 +78,37 @@ impl DiscordNotifier {
         Self {
             webhook_url: webhook_url.to_owned(),
             client: Client::new(),
-            message_queue: Mutex::new(HashMap::new()),
-        }
-    }
-
-    /// Lock the message queue, recovering from a poisoned lock if necessary.
-    fn lock_message_queue(&self) -> MutexGuard<'_, HashMap<Option<Snowflake>, (String, usize)>> {
-        // If the lock is poisoned, we clear it and return a new lock.
-        self.message_queue.lock().unwrap_or_else(|mut err| {
-            error!("Discord notification queue was poisoned; some notifications may be lost");
-            **err.get_mut() = HashMap::new();
-            self.message_queue.clear_poison();
-            err.into_inner()
-        })
-    }
-
-    /// Enqueue a message to be sent to the Discord channel the next time notifications are
-    /// dispatched.
-    ///
-    /// Only one message is sent per user per notification dispatch, so if a message was already
-    /// enqueued with the same `ping` value, it will be replaced with the new message.
-    pub fn enqueue_message(&self, ping: Option<Snowflake>, message: String) {
-        self.lock_message_queue().insert(ping, (message, 0));
-    }
-
-    /// Dispatch all messages in the queue, sending them to Discord.
-    ///
-    /// Returns `(n_sent, n_failed)` where `n_sent` is the number of messages that were sent
-    /// successfully, and `n_failed` is the number of messages that failed to send. Errors are
-    /// logged but not returned.
-    #[instrument(name = "discord.dispatch_messages", skip(self))]
-    pub async fn dispatch_messages(&self) -> (usize, usize) {
-        // Take the map to avoid race conditions while sending messages.
-        let mut map = {
-            let mut lock = self.lock_message_queue();
-            std::mem::take(&mut *lock)
-        };
-        let mut sent = 0;
-        let mut failed = 0;
-        for (ping, (message, attempts)) in map.drain() {
-            let span = info_span!("discord.dispatch_message", ?ping, prev_attempts = attempts);
-            {
-                let _enter = span.enter();
-                if attempts >= MAX_DELIVERY_ATTEMPTS {
-                    error!(
-                        channel = %ping.map_or("channel".to_string(), |id| id.to_string()),
-                        attempts,
-                        "Failed to send message to channel after several attempts; giving up",
-                    );
-                    continue;
-                }
-            }
-            let result = self
-                .send_message(ping, &message)
-                .instrument(span.clone())
-                .await;
-            let _enter = span.enter();
-            if result.is_err() {
-                // If no new message was enqueued for the same recipient, retry this one.
-                let mut lock = self.lock_message_queue();
-                if lock.get(&ping).is_none() {
-                    lock.insert(ping, (message, attempts + 1));
-                }
-                failed += 1;
-            } else {
-                sent += 1;
-            }
-        }
-        if sent > 0 || failed > 0 {
-            info!(sent, failed, "Discord notifications dispatched");
         }
-        (sent, failed)
     }
 
     /// Send a message in the channel this notifier is registered to.
     #[instrument(name = "discord.send_message", skip(self, message), err(Display))]
-    async fn send_message(&self, ping: Option<Snowflake>, message: &str) -> eyre::Result<()> {
+    pub async fn send_message(&self, ping: Option<Snowflake>, message: &str) -> eyre::Result<()> {
+        let result;
         loop {
             let response = self.send_single_message(ping, message).await?;
             // If we get rate limited, try again.
             // See https://discord.com/developers/docs/topics/rate-limits.
             if response.status() == StatusCode::TOO_MANY_REQUESTS {
-                match response.headers().get(header::RETRY_AFTER) {
-                    Some(value) => {
-                        warn!("Retry-After" = ?value, "Hit Discord API rate limit; retrying after delay");
-                        sleep_from_retry_after(value).await?;
-                        continue;
-                    }
-                    None => bail!("Got rate-limited but response contains no Retry-After header"),
+                if let Some(value) = response.headers().get(header::RETRY_AFTER) {
+                    warn!("Retry-After" = ?value, "Hit Discord API rate limit; retrying after delay");
+                    sleep_from_retry_after(value).await?;
+                    continue;
                 }
+                result = Err(eyre!(
+                    "Got rate-limited but response contains no Retry-After header"
+                ));
+            } else {
+                result = response
+                    .error_for_status()
+                    .wrap_err("Discord webhook returned error");
             }
-            response
-                .error_for_status()
-                .wrap_err("Discord webhook returned error")?;
             break;
         }
-        Ok(())
+        if let Err(err) = &result {
+            error!(%err, "Failed to send Discord message");
+        }
+        result.map(|_| ())
     }
 
     /// Sends a single message to the Discord webhook.
@@ -238,10 +164,9 @@ async fn sleep_from_retry_after(header_val: &HeaderValue) -> eyre::Result<()> {
 
 #[cfg(test)]
 mod tests {
-    use httpmock::{Method, MockServer};
     use reqwest::Url;
 
-    use super::{DiscordNotifier, MAX_DELIVERY_ATTEMPTS};
+    use super::DiscordNotifier;
 
     /// Tests sending a message in Discord.
     ///
@@ -265,59 +190,4 @@ mod tests {
             .await
             .unwrap();
     }
-
-    /// Tests that messages destined for the same user are deduplicated
-    #[tokio::test]
-    async fn test_dispatch_deduplicate() {
-        let server = MockServer::start_async().await;
-        let mock = server.mock(|when, then| {
-            when.method(Method::POST);
-            then.status(200);
-        });
-        let notifier = DiscordNotifier::new(&Url::parse(&server.base_url()).unwrap());
-
-        notifier.enqueue_message(Some(1234.into()), "first message".to_string());
-        // Next message should replace the first one
-        notifier.enqueue_message(Some(1234.into()), "second message".to_string());
-        notifier.enqueue_message(Some(4321.into()), "message for other user".to_string());
-        notifier.enqueue_message(None, "message for no user".to_string());
-
-        let (sent, failed) = notifier.dispatch_messages().await;
-
-        assert_eq!(3, sent);
-        assert_eq!(0, failed);
-        assert_eq!(3, mock.hits_async().await);
-
-        // Ensure sent messages were removed
-        let (sent, failed) = notifier.dispatch_messages().await;
-        assert_eq!(0, sent);
-        assert_eq!(0, failed);
-        assert_eq!(3, mock.hits_async().await); // 3 were sent last round
-    }
-
-    #[tokio::test]
-    async fn test_give_up_after_max_attempts() {
-        let server = MockServer::start_async().await;
-        let mock = server.mock(|when, then| {
-            when.method(Method::POST);
-            then.status(500);
-        });
-        let notifier = DiscordNotifier::new(&Url::parse(&server.base_url()).unwrap());
-
-        notifier.enqueue_message(Some(1234.into()), "test message".to_string());
-
-        // Expect MAX_DELIVERY_ATTEMPTS attempts
-        for i in 0..MAX_DELIVERY_ATTEMPTS {
-            let (sent, failed) = notifier.dispatch_messages().await;
-            assert_eq!(0, sent);
-            assert_eq!(1, failed);
-            assert_eq!(i + 1, mock.hits_async().await);
-        }
-
-        // Expect no more attempts
-        let (sent, failed) = notifier.dispatch_messages().await;
-        assert_eq!(0, sent);
-        assert_eq!(0, failed);
-        assert_eq!(MAX_DELIVERY_ATTEMPTS, mock.hits_async().await);
-    }
 }

src/main.rs

@@ -99,18 +99,6 @@ async fn main() -> ExitCode {
         });
     }
 
-    // Send Discord notifications every 24 hours
-    if let Some(notifier) = webring.notifier() {
-        const DISCORD_NOTIFICATION_INTERVAL: Duration = Duration::from_secs(60 * 60 * 24);
-        let notifier = Arc::clone(notifier);
-        tokio::spawn(async move {
-            loop {
-                notifier.dispatch_messages().await;
-                tokio::time::sleep(DISCORD_NOTIFICATION_INTERVAL).await;
-            }
-        });
-    }
-
     webring.enable_ip_pruning(chrono::Duration::hours(1));
     if let Err(err) = webring.enable_reloading(&cli.config_file) {
         error!(%err, "Failed to watch configuration files for changes");