dm: wake subscription reporter at the next deadline, not every 4 s

The reporter loop slept on a fixed `Timer::after(4 s)`, with the only
early-wake signal being `Subscriptions::notification`. When an event
(or attribute change) was emitted while a subscription was still inside
its `min_int` quiet period, the notification fired, the loop woke,
`is_report_allowed` was false, the notification was consumed without
sending a report, and the next wake-up had to come from the 4 s
fallback timer. A 1 s `MinIntervalFloor` therefore observed as ~4 s
between the first and second report in a burst — visible on Generic
Switch endpoints as ~4 s between `InitialPress` and `ShortRelease`.

Add `Subscriptions::next_wake`, which returns the earliest instant at
which any subscription will become reportable: end of `min_int` if it
has pending changes/events, or start of the `max_int / 2` liveness
window otherwise. The reporter sleeps until that instant, clamped to
a 4 s safety ceiling.

Author: snabb

rs-matter/src/dm.rs

@@ -604,9 +604,29 @@ where
     /// Process all valid subscriptions in an endless loop, checking for changes
     /// and reporting them to the peers.
     async fn process_subscriptions(&self, matter: &Matter<'_>) -> Result<(), Error> {
+        // Hard ceiling on how long the reporter can sleep without
+        // re-evaluating. Bounds the latency of late notifications and acts
+        // as the fallback when no subscription has an earlier deadline.
+        const SLEEP_CEILING: embassy_time::Duration = embassy_time::Duration::from_secs(4);
+
         loop {
-            // TODO: Un-hardcode these 4 seconds of waiting when the more precise change detection logic is implemented
-            let mut timeout = pin!(Timer::after(embassy_time::Duration::from_secs(4)));
+            // Wake at the soonest of: a subscription's `min_int` quiet
+            // period ending while it has pending data, a subscription's
+            // `max_int / 2` liveness deadline, or the safety ceiling.
+            // Without this, events emitted during a sub's quiet period
+            // would `notify()` once, see `is_report_allowed == false`, and
+            // then have to wait the full ceiling before being looked at
+            // again — turning a 1 s min-interval into a ~4 s observed gap.
+            let now = Instant::now();
+            let event_numbers_watermark = self.events.watermark();
+            let ceiling_at = now + SLEEP_CEILING;
+            let deadline = self
+                .subscriptions
+                .next_wake(event_numbers_watermark)
+                .map(|d| if d < ceiling_at { d } else { ceiling_at })
+                .unwrap_or(ceiling_at);
+
+            let mut timeout = pin!(Timer::at(deadline));
             let mut notification = pin!(self.subscriptions.notification.wait());
             let mut session_removed = pin!(matter.session_removed.wait());
 

rs-matter/src/dm/subscriptions.rs

@@ -370,6 +370,57 @@ impl<const N: usize> Subscriptions<N> {
         })
     }
 
+    /// Earliest [`Instant`] at which the reporter loop next needs to wake to
+    /// make forward progress, or `None` if no subscriptions exist.
+    ///
+    /// Considered, per subscription:
+    /// - If the subscription has pending changed attributes or new events,
+    ///   the end of its min-interval quiet period (`reported_at + min_int`).
+    /// - The start of the max-interval liveness window
+    ///   (`reported_at + max_int / 2`), which is when `is_report_due` first
+    ///   flips true.
+    ///
+    /// Callers should still clamp the returned instant against their own
+    /// safety ceiling (e.g. the historical 4 s fallback) before sleeping.
+    pub(crate) fn next_wake(&self, event_numbers_watermark: EventNumber) -> Option<Instant> {
+        self.state.lock(|state| {
+            let state = state.borrow();
+            let mut earliest: Option<Instant> = None;
+
+            let mut consider = |inst: Instant| {
+                earliest = Some(match earliest {
+                    Some(e) if e <= inst => e,
+                    _ => inst,
+                });
+            };
+
+            for sub in state.subscriptions.iter() {
+                let has_pending = sub.is_affected_by_attr_changes(&[], &state.changed_attrs)
+                    || sub.is_affected_by_new_events(&[], event_numbers_watermark);
+
+                if has_pending {
+                    if let Some(next_allowed) = sub
+                        .reported_at
+                        .checked_add(embassy_time::Duration::from_secs(sub.min_int_secs as _))
+                    {
+                        consider(next_allowed);
+                    }
+                }
+
+                if let Some(due_at) =
+                    sub.reported_at
+                        .checked_add(embassy_time::Duration::from_secs(
+                            (sub.max_int_secs / 2) as _,
+                        ))
+                {
+                    consider(due_at);
+                }
+            }
+
+            earliest
+        })
+    }
+
     /// Remove entries that every subscription has already reported on.
     pub(crate) fn purge_reported_changes(&self) {
         self.state
@@ -2141,6 +2192,40 @@ mod tests {
         }
     }
 
+    #[test]
+    fn next_wake_returns_min_int_when_events_pending() {
+        // With a pending event during the min-interval quiet period,
+        // `next_wake` returns the end of that quiet period — not the
+        // (much later) liveness deadline.
+        let subs: Subscriptions<1> = Subscriptions::new();
+        let pool = TestPool::<2>::new(0);
+        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();
+
+        let now = Instant::now();
+        // min_int = 1 s, max_int = 60 s.
+        {
+            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
+            rctx.set_keep();
+        }
+
+        // No pending events → next wake is the liveness window
+        // (now + max_int/2 = now + 30 s).
+        assert_eq!(subs.next_wake(0), Some(now + Duration::from_secs(30)),);
+
+        // Pending event (watermark bumped past the sub's max_seen=0) →
+        // next wake collapses to the end of the min-interval quiet period
+        // (now + min_int = now + 1 s), which is much sooner than the
+        // liveness window. Regression guard for the historical 4 s gap
+        // between InitialPress and the next event report.
+        assert_eq!(subs.next_wake(7), Some(now + Duration::from_secs(1)),);
+    }
+
+    #[test]
+    fn next_wake_is_none_when_no_subscriptions() {
+        let subs: Subscriptions<1> = Subscriptions::new();
+        assert_eq!(subs.next_wake(0), None);
+    }
+
     #[test]
     fn is_expired_uses_max_int() {
         // `Subscription::is_expired` returns true once `max_int` has elapsed