fix: keep sending notifications to users under a continuous stream of events

src/message.rs

@@ -123,8 +123,13 @@ pub enum MessageType {
 
 pub static DEBOUNCE_ENABLE: AtomicBool = AtomicBool::new(true);
 
+/// How long a slot stays quiet before its message is considered settled
+const BURST_WINDOW: Duration = Duration::from_millis(100);
+
 #[derive(Clone, Debug)]
 struct SendQueueItem {
+    /// when the held message was first queued, as opposed to last updated
+    queued: Instant,
     received: Instant,
     sent: Instant,
     message: Option<PushMessage>,
@@ -133,6 +138,7 @@ struct SendQueueItem {
 impl Default for SendQueueItem {
     fn default() -> Self {
         SendQueueItem {
+            queued: Instant::now() - Duration::from_secs(120),
             received: Instant::now() - Duration::from_secs(120),
             sent: Instant::now() - Duration::from_secs(120),
             message: None,
@@ -178,13 +184,12 @@ impl SendQueue {
         };
 
         match &mut item.message {
-            Some(queued) => {
-                queued.merge(&message);
-            }
-            opt => {
-                *opt = Some(message);
+            Some(queued) => queued.merge(&message),
+            None => {
+                item.message = Some(message);
+                item.queued = time;
             }
-        };
+        }
         item.received = time;
 
         None
@@ -203,16 +208,18 @@ impl SendQueue {
                 max_debounce_time,
                 debounce_factor,
             );
-            if now.duration_since(item.sent) > debounce_time {
-                if now.duration_since(item.received) > Duration::from_millis(100) {
-                    item.sent = now;
-                    item.message.take()
-                } else {
-                    None
-                }
-            } else {
-                None
+            if now.duration_since(item.sent) <= debounce_time {
+                return None;
+            }
+            // let a burst settle so related updates go out as one message, but never
+            // hold on past the debounce window or a continuous stream of updates would
+            // keep pushing the deadline out and nothing would ever be sent
+            let settled = now.duration_since(item.received) > BURST_WINDOW;
+            if !settled && now.duration_since(item.queued) <= debounce_time {
+                return None;
             }
+            item.sent = now;
+            item.message.take()
         })
     }
 }
@@ -348,3 +355,30 @@ fn test_send_queue_1() {
             .collect::<Vec<_>>()
     );
 }
+
+#[test]
+fn test_send_queue_sustained_updates() {
+    // a user whose storages are being written to continuously (a bulk upload, a busy
+    // groupfolder) receives an event more often than the drain interval.
+    // the queue must still hand out a message within the debounce window.
+    let base_time = Instant::now();
+    let mut queue = SendQueue::new(15, 1.0);
+
+    let mut sent = Vec::new();
+    // 30 seconds of updates arriving every 50ms, drained on the 500ms tick
+    for step in 0..600 {
+        let now = base_time + Duration::from_millis(step * 50);
+        queue.push(
+            PushMessage::File(UpdatedFiles::Known(vec![step].into())),
+            now,
+        );
+        if step % 10 == 0 {
+            sent.extend(queue.drain(now, 1));
+        }
+    }
+
+    assert!(
+        !sent.is_empty(),
+        "no message was sent during 30s of sustained updates"
+    );
+}