From 0dfadf35ca1cc3f33904243dabea436aa409110e Mon Sep 17 00:00:00 2001 From: Git'Fellow <12234510+solracsf@users.noreply.github.com> Date: Mon, 7 Sep 2026 10:02:44 +0200 Subject: [PATCH 1/2] fix: keep sending updates to users under a continuous stream of events The send queue only released a message once no update had landed in that slot for 100ms. That works for a short burst, but a user whose storages are being written to continuously (a bulk upload, a busy groupfolder) keeps refreshing `received` and the message is never handed out at all, so the client stops getting notified for as long as the activity lasts. Track when a message was first queued and release it once it has waited the full debounce time, regardless of whether the burst has settled. Signed-off-by: Git'Fellow <12234510+solracsf@users.noreply.github.com> --- src/message.rs | 53 +++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 46 insertions(+), 7 deletions(-) diff --git a/src/message.rs b/src/message.rs index 8aefb065..d15bdbdf 100644 --- a/src/message.rs +++ b/src/message.rs @@ -125,6 +125,9 @@ pub static DEBOUNCE_ENABLE: AtomicBool = AtomicBool::new(true); #[derive(Clone, Debug)] struct SendQueueItem { + /// when the currently held message was first queued + queued: Instant, + /// when the currently held message was last updated received: Instant, sent: Instant, message: Option, @@ -133,6 +136,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, @@ -177,6 +181,7 @@ impl SendQueue { None => return Some(message), }; + let first = item.message.is_none(); match &mut item.message { Some(queued) => { queued.merge(&message); @@ -185,6 +190,9 @@ impl SendQueue { *opt = Some(message); } }; + if first { + item.queued = time; + } item.received = time; None @@ -203,13 +211,17 @@ 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 - } + if now.duration_since(item.sent) <= debounce_time { + return None; + } + // hold a burst back briefly so related updates end up in a single message, + // but never longer than the debounce time itself: under a continuous stream + // of updates the message would otherwise never be sent at all + let burst_settled = now.duration_since(item.received) > Duration::from_millis(100); + let waited_full_debounce = now.duration_since(item.queued) > debounce_time; + if burst_settled || waited_full_debounce { + item.sent = now; + item.message.take() } else { None } @@ -348,3 +360,30 @@ fn test_send_queue_1() { .collect::>() ); } + +#[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" + ); +} From d8dc0e03ea598f615c123e3b46caf8a88b146ad8 Mon Sep 17 00:00:00 2001 From: Git'Fellow <12234510+solracsf@users.noreply.github.com> Date: Mon, 7 Sep 2026 12:37:28 +0200 Subject: [PATCH 2/2] refactor: flatten the drain condition and name the burst window `if let`/`else` on the slot removes the `first` flag and the catch-all arm that only existed to dodge a borrow. Early returns replace the nested condition, and the 100ms settle delay gets a name instead of appearing as a literal in the middle of a boolean. Signed-off-by: Git'Fellow <12234510+solracsf@users.noreply.github.com> --- src/message.rs | 37 ++++++++++++++++--------------------- 1 file changed, 16 insertions(+), 21 deletions(-) diff --git a/src/message.rs b/src/message.rs index d15bdbdf..db8aea63 100644 --- a/src/message.rs +++ b/src/message.rs @@ -123,11 +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 currently held message was first queued + /// when the held message was first queued, as opposed to last updated queued: Instant, - /// when the currently held message was last updated received: Instant, sent: Instant, message: Option, @@ -181,17 +183,12 @@ impl SendQueue { None => return Some(message), }; - let first = item.message.is_none(); 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; } - }; - if first { - item.queued = time; } item.received = time; @@ -214,17 +211,15 @@ impl SendQueue { if now.duration_since(item.sent) <= debounce_time { return None; } - // hold a burst back briefly so related updates end up in a single message, - // but never longer than the debounce time itself: under a continuous stream - // of updates the message would otherwise never be sent at all - let burst_settled = now.duration_since(item.received) > Duration::from_millis(100); - let waited_full_debounce = now.duration_since(item.queued) > debounce_time; - if burst_settled || waited_full_debounce { - item.sent = now; - item.message.take() - } else { - 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() }) } }