diff --git a/src/message.rs b/src/message.rs index 8aefb065..db8aea63 100644 --- a/src/message.rs +++ b/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, @@ -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::>() ); } + +#[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" + ); +}