rs_matter/im/

subscriptions.rs

/*
 *
 *    Copyright (c) 2024-2026 Project CHIP Authors
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

use core::num::NonZeroU8;

use embassy_time::Instant;

#[cfg(feature = "persistent-subscriptions")]
use crate::error::Error;
use crate::fabric::MAX_FABRICS;
use crate::im::{AttrId, ClusterId, DeviceLoad, EndptId, EventId, EventNumber, IMBuffer, NodeId};
#[cfg(feature = "persistent-subscriptions")]
use crate::persist::{KvBlobStore, PERSISTENT_SUBSCRIPTIONS_START};
#[cfg(feature = "persistent-subscriptions")]
use crate::tlv::{FromTLV, OctetStr, Octets, TLVElement, TLVTag, ToTLV};
use crate::utils::cell::RefCell;
use crate::utils::init::{init, Init};
use crate::utils::storage::pooled::Buffers;
use crate::utils::storage::Vec;
#[cfg(feature = "persistent-subscriptions")]
use crate::utils::storage::WriteBuf;
use crate::utils::sync::blocking::Mutex;
use crate::utils::sync::{DynBase, Notification};

/// The maximum number of subscriptions that can be tracked at the same time by default.
///
/// According to the Matter spec, at least 3 subscriptions per fabric should be supported.
pub const DEFAULT_MAX_SUBSCRIPTIONS: usize = MAX_FABRICS * 3;

/// The maximum number of changed-attribute entries tracked simultaneously.
///
/// When the table is full, entries are coalesced ("promoted") to coarser-grained
/// wildcards so that new changes can always be recorded.
pub const MAX_CHANGED_ATTRS: usize = 16;

/// A struct for the RX buffers containing the read requests of the tracked subscriptions.
// NOTE: `SubscriptionsBuffers` is a thin wrapper around a second
// `Mutex<RefCell<Vec<..>>>` that is *always* locked in lockstep with
// `Subscriptions::state` (see `Subscriptions::with`). As long as that lock
// order is respected the pair is safe, but the two locks let someone (now or
// in the future) lock only one of them and violate the invariant that
// `subscriptions.len() == buffers.len()`. The cleanest fix is to move the
// `Vec<B::Buffer<'a>, N>` *into* `SubscriptionsInner` behind the same mutex
// so it cannot be locked independently. The current layout also forces all
// public APIs to thread an extra `&SubscriptionsBuffers` argument everywhere,
// which is why `remove` / `report` / `add` all grew a second ref parameter.
pub struct SubscriptionsBuffers<'a, B, const N: usize = DEFAULT_MAX_SUBSCRIPTIONS>
where
    B: Buffers<IMBuffer> + 'a,
{
    buffers: Mutex<RefCell<SubscriptionsBuffersInner<'a, B, N>>>,
}

impl<'a, B, const N: usize> SubscriptionsBuffers<'a, B, N>
where
    B: Buffers<IMBuffer> + 'a,
{
    /// Create the instance.
    pub const fn new() -> Self {
        Self {
            buffers: Mutex::new(RefCell::new(Vec::new())),
        }
    }

    /// Return an in-place initializer for the instance.
    pub fn init() -> impl Init<Self> {
        init!(Self {
            buffers <- Mutex::init(RefCell::init(Vec::init())),
        })
    }

    fn with<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&mut SubscriptionsBuffersInner<'a, B, N>) -> R,
    {
        self.buffers.lock(|buffers| f(&mut buffers.borrow_mut()))
    }
}

impl<'a, B, const N: usize> Default for SubscriptionsBuffers<'a, B, N>
where
    B: Buffers<IMBuffer> + 'a,
{
    fn default() -> Self {
        Self::new()
    }
}

/// A type alias for the inner buffer vector of `SubscriptionsBuffers`.
type SubscriptionsBuffersInner<'a, B, const N: usize> =
    Vec<<B as Buffers<IMBuffer>>::Buffer<'a>, N>;

/// A type for tracking subscriptions accepted by the data model.
///
/// The `N` type parameter specifies the maximum number of subscriptions that can be tracked at the same time.
/// Additional subscriptions are rejected by the data model with a "resource exhausted" IM status message.
pub struct Subscriptions<const N: usize = DEFAULT_MAX_SUBSCRIPTIONS> {
    state: Mutex<RefCell<SubscriptionsInner<N>>>,
    pub(crate) notification: Notification,
}

impl<const N: usize> Subscriptions<N> {
    /// Create the instance.
    #[inline(always)]
    pub const fn new() -> Self {
        Self {
            state: Mutex::new(RefCell::new(SubscriptionsInner::new())),
            notification: Notification::new(),
        }
    }

    /// Create an in-place initializer for the instance.
    pub fn init() -> impl Init<Self> {
        init!(Self {
            state <- Mutex::init(RefCell::init(SubscriptionsInner::init())),
            notification <- Notification::init(),
        })
    }

    /// Whether subscription persistence is compiled in.
    ///
    /// `true` only when the `persistent-subscriptions` feature is enabled; when it
    /// is off, the persist/resume machinery is compiled out entirely and this is a
    /// `const false` that lets the linker drop the callers.
    pub const fn persist_enabled(&self) -> bool {
        cfg!(feature = "persistent-subscriptions")
    }

    /// Notify the instance that the data of a specific attribute has changed and that it should re-evaluate the subscriptions
    /// and report on those that are interested in the changed data.
    ///
    /// This method is supposed to be called by the application code whenever it changes the data of an attribute.
    ///
    /// # Arguments
    /// - `endpoint_id`: The endpoint ID of the cluster that had changed.
    /// - `cluster_id`: The cluster ID of the cluster that had changed.
    /// - `attr_id`: The attribute ID of the attribute that changed.
    pub(crate) fn notify_attr_changed(
        &self,
        endpoint_id: EndptId,
        cluster_id: ClusterId,
        attr_id: AttrId,
    ) {
        self.state.lock(|internal| {
            internal
                .borrow_mut()
                .changed_attrs
                .record(endpoint_id, cluster_id, attr_id);
        });

        // The per-subscription decision of whether anything needs to be reported is
        // computed on-the-fly by `find_report_due` (and by the responder's filter)
        // by consulting the live `changed_attrs` table, so there is no per-sub flag
        // to flip here. We just wake the reporter task.
        self.notification.notify();
    }

    /// Record a cluster-wide change. Every attribute of `(endpoint_id,
    /// cluster_id)` is treated as changed for the purposes of subscription
    /// reporting.
    pub(crate) fn notify_cluster_changed(&self, endpoint_id: EndptId, cluster_id: ClusterId) {
        self.state.lock(|internal| {
            internal
                .borrow_mut()
                .changed_attrs
                .record_wildcard(Some(endpoint_id), Some(cluster_id));
        });

        self.notification.notify();
    }

    /// Record an endpoint-wide change. Every attribute on every cluster of
    /// `endpoint_id` is treated as changed for the purposes of subscription
    /// reporting.
    pub(crate) fn notify_endpoint_changed(&self, endpoint_id: EndptId) {
        self.state.lock(|internal| {
            internal
                .borrow_mut()
                .changed_attrs
                .record_wildcard(Some(endpoint_id), None);
        });

        self.notification.notify();
    }

    /// Whether a live subscription exists on `fab_idx` whose subscriber node
    /// matches `node_id`.
    ///
    /// This is the ICD "is this client subscribed?" predicate that decides
    /// whether a registered client is nudged with a Check-In. Persisted
    /// subscriptions are re-hydrated into this same table at startup
    /// (`resume_subscriptions`), so a resumed subscription counts here too — no
    /// separate "persisted subscription" lookup is needed.
    pub fn has_subscription_for(&self, fab_idx: NonZeroU8, node_id: NodeId) -> bool {
        self.state.lock(|internal| {
            internal
                .borrow()
                .subscriptions
                .iter()
                .any(|s| s.ids.fab_idx == fab_idx && s.ids.peer_node_id == node_id)
        })
    }

    /// A consistent snapshot of the subscription-related figures of
    /// `GeneralDiagnostics::DeviceLoadStatus`.
    ///
    /// The Interaction Model message counters are left at their defaults -
    /// this type knows nothing about them; see
    /// [`crate::im::InteractionModel::load_stats`].
    pub fn load_stats(&self, fab_idx: Option<NonZeroU8>) -> DeviceLoad {
        self.state.lock(|internal| {
            let internal = internal.borrow();

            DeviceLoad {
                // Counts the in-flight subscription too, if any.
                current_subscriptions: internal.subscriptions_count as _,
                // Walks the active list, so a subscription that is in-flight at
                // this instant is not counted - it carries no fabric index
                // while it sits in the `ReportContext`.
                current_subscriptions_for_fabric: fab_idx
                    .map(|fab_idx| {
                        internal
                            .subscriptions
                            .iter()
                            .filter(|s| s.ids.fab_idx == fab_idx)
                            .count() as _
                    })
                    .unwrap_or(0),
                // The ID counter is monotonic and hands out 1 first.
                total_subscriptions_established: internal.next_subscription_id.saturating_sub(1),
                ..Default::default()
            }
        })
    }

    /// Record a fully-global change. Every attribute on every cluster on
    /// every endpoint is treated as changed for the purposes of subscription
    /// reporting. Intended for coarse-grained reset / restart scenarios.
    pub(crate) fn notify_all_changed(&self) {
        self.state.lock(|internal| {
            internal
                .borrow_mut()
                .changed_attrs
                .record_wildcard(None, None);
        });

        self.notification.notify();
    }

    /// Notify the instance that a new event has been emitted and that it should
    /// re-evaluate the subscriptions and report on those that are interested in the new event.
    ///
    /// Public for the integration tests.
    pub fn notify_event_emitted(
        &self,
        _endpoint_id: EndptId,
        _cluster_id: ClusterId,
        _event_id: EventId,
    ) {
        // Events are filtered at report time by `min_event_number` + event path matching.
        // Whether a subscription is due to report because of new events is recomputed on
        // the fly in `find_report_due`, so here we only need to kick the reporter task.
        self.notification.notify();
    }

    /// Clear all subscriptions and pending changes.
    /// Used when initializing a new data model.
    pub(crate) fn clear(&self) {
        self.state.lock(|state| state.borrow_mut().clear());
    }

    /// Add a new subscription with the given parameters.
    /// Returns a context for the initial report if successful, or `None` if the subscription table is full.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn add<'a, 's, B>(
        &'s self,
        now: Instant,
        fabric_idx: NonZeroU8,
        peer_node_id: u64,
        min_int_secs: u16,
        max_int_secs: u16,
        event_numbers_watermark: EventNumber,
        buffer: B::Buffer<'a>,
        buffers: &'s SubscriptionsBuffers<'a, B, N>,
    ) -> Option<ReportContext<'a, 's, B, N>>
    where
        B: Buffers<IMBuffer> + 'a,
    {
        let (sub, buf, next_max_seen_attr_change_id) = self.with(buffers, |state, buffers| {
            let (sub, buf) = state.add::<B>(
                fabric_idx,
                peer_node_id,
                min_int_secs,
                max_int_secs,
                buffer,
                buffers,
            )?;

            // Mirror `report()`: commit the current watermark so the priming
            // report's `set_keep` does not regress the subscription's `since`
            // to 0 (which would cause every pre-`add` change to be replayed
            // on the first incremental report).
            Some((sub, buf, state.changed_attrs.watermark()))
        })?;

        Some(ReportContext {
            subscriptions: self,
            subscriptions_buffers: buffers,
            subscription: Some(sub),
            subscription_buffer: Some(buf),
            next_max_seen_attr_change_id,
            next_max_seen_event_number: event_numbers_watermark,
            next_reported_at: now,
            next_retry_at: Instant::MIN,
            next_fail_count: 0,
            keep: false,
        })
    }

    /// Remove every subscription for which `f` returns `Some(reason)`.
    ///
    /// A subscription that is currently being reported on has been moved out
    /// of `state.subscriptions` into its `ReportContext` (see
    /// [`SubscriptionsInner::report`]). To keep such an in-flight subscription
    /// observable, [`SubscriptionsInner::report`] also leaves a clone of it in
    /// `state.reporting`. If the predicate matches that clone, we flip
    /// `state.reporting_cancelled` so that [`SubscriptionsInner::report_complete`]
    /// drops the subscription on `Drop` of its `ReportContext` instead of
    /// re-inserting it. The count invariant is preserved: either the Vec path
    /// decrements `subscriptions_count` now, or `report_complete` does it
    /// later — never both for the same subscription.
    pub(crate) fn remove<B, F>(&self, buffers: &SubscriptionsBuffers<'_, B, N>, mut f: F) -> bool
    where
        B: Buffers<IMBuffer>,
        F: FnMut(&Subscription) -> Option<&'static str>,
    {
        let removed = self.with(buffers, |state, buffers| {
            let mut removed = false;

            loop {
                let next = state
                    .subscriptions
                    .iter()
                    .enumerate()
                    .filter_map(|(index, subscription)| {
                        f(subscription).map(|reason| (index, subscription.ids().clone(), reason))
                    })
                    .next();

                let Some((index, ids, reason)) = next else {
                    break;
                };

                state.subscriptions.swap_remove(index);
                buffers.swap_remove(index);

                state.subscriptions_count -= 1;

                info!("Removed subscription {:?}, reason: {}", ids, reason);

                removed = true;
            }

            // Consider the in-flight subscription (if any). It is not in
            // `state.subscriptions`; only a snapshot clone lives in
            // `state.reporting`. If the predicate matches and we have not
            // already flagged it for cancellation, request that
            // `report_complete` drop it.
            if state.reporting_cancelled.is_none() {
                if let Some(sub) = state.reporting.as_ref() {
                    if let Some(reason) = f(sub) {
                        info!(
                            "Marked in-flight subscription {:?} for removal, reason: {}",
                            sub.ids(),
                            reason
                        );
                        state.reporting_cancelled = Some(reason);
                        removed = true;
                    }
                }
            }

            removed
        });

        if removed {
            self.notification.notify();
        }

        removed
    }

    /// Begin a report for the subscription with the given parameters.
    /// Returns a context capturing the subscription's current state if successful, or `None`
    /// if no subscription is currently reportable.
    pub(crate) fn report<'a, 's, B>(
        &'s self,
        now: Instant,
        event_numbers_watermark: EventNumber,
        buffers: &'s SubscriptionsBuffers<'a, B, N>,
    ) -> Option<ReportContext<'a, 's, B, N>>
    where
        B: Buffers<IMBuffer> + 'a,
    {
        let (sub, buf, next_max_seen_attr_change_id) = self.with(buffers, |state, buffers| {
            let (sub, buf) = state.report::<B>(now, event_numbers_watermark, buffers)?;
            let attr_change_ids_watermark = state.changed_attrs.watermark();

            debug!("About to report on subscription {:?}, details: max_seen_attr_change_id: {}, max_seen_event_number: {}, attr_change_ids_watermark: {}, event_numbers_watermark: {}", sub.ids(), sub.max_seen_attr_change_id, sub.max_seen_event_number, attr_change_ids_watermark, event_numbers_watermark);

            Some((sub, buf, attr_change_ids_watermark))
        })?;

        Some(ReportContext {
            subscriptions: self,
            subscriptions_buffers: buffers,
            subscription: Some(sub),
            subscription_buffer: Some(buf),
            next_max_seen_attr_change_id,
            next_max_seen_event_number: event_numbers_watermark,
            next_reported_at: now,
            next_retry_at: Instant::MIN,
            next_fail_count: 0,
            keep: false,
        })
    }

    /// Earliest [`Instant`] at which any subscription will next need
    /// servicing, or [`Instant::MAX`] if there are no (primed) subscriptions
    /// and the reporter should simply wait to be notified.
    ///
    /// See [`Subscription::next_report_at`] for the per-subscription rule.
    pub(crate) fn next_report_at<'a, B>(
        &self,
        event_numbers_watermark: EventNumber,
        buffers: &SubscriptionsBuffers<'a, B, N>,
    ) -> Instant
    where
        B: Buffers<IMBuffer> + 'a,
    {
        self.with(buffers, |state, buffers| {
            state.next_report_at::<B>(event_numbers_watermark, buffers)
        })
    }

    /// Remove entries that every subscription has already reported on.
    pub(crate) fn purge_reported_changes(&self) {
        self.state
            .lock(|state| state.borrow_mut().purge_reported_changes())
    }

    /// Complete a report by updating the subscription's watermark and last-reported timestamp,
    /// and re-inserting it into the table if the `keep` flag is set on the context.
    fn report_complete<'a, B>(&self, report: &mut ReportContext<'a, '_, B, N>)
    where
        B: Buffers<IMBuffer> + 'a,
    {
        let mut sub = unwrap!(report.subscription.take());
        let buf = unwrap!(report.subscription_buffer.take());

        sub.max_seen_attr_change_id = report.next_max_seen_attr_change_id;
        sub.max_seen_event_number = report.next_max_seen_event_number;
        sub.reported_at = report.next_reported_at;
        sub.retry_at = report.next_retry_at;
        sub.fail_count = report.next_fail_count;

        let keep = report.keep;

        self.with(report.subscriptions_buffers, |state, buffers| {
            state.report_complete::<B>(sub, buf, buffers, keep)
        })
    }

    fn with<'a, B, F, R>(&self, buffers: &SubscriptionsBuffers<'a, B, N>, f: F) -> R
    where
        B: Buffers<IMBuffer> + 'a,
        F: FnOnce(&mut SubscriptionsInner<N>, &mut SubscriptionsBuffersInner<'a, B, N>) -> R,
    {
        self.state.lock(|state| {
            let mut state = state.borrow_mut();
            buffers.with(|buffers| f(&mut state, buffers))
        })
    }
}

/// Subscription persistence: the whole table is mirrored to (and resumed from)
/// the key-value store, one record per subscription. Gated as a unit so that a
/// device that does not want persistence drops it — and the TLV serialization
/// and key-value store traffic it pulls in — entirely.
#[cfg(feature = "persistent-subscriptions")]
impl<const N: usize> Subscriptions<N> {
    /// Compile-time proof that every slot of this table maps to a key inside
    /// the reserved persisted-subscription range - i.e. that persisting can
    /// never run past it and into the vendor keys.
    ///
    /// Evaluated by every method below that maps a slot to a key, so an
    /// oversized table is rejected at build time rather than corrupting
    /// whatever lives past the range. Being a check on a generic parameter, it
    /// is only evaluated once monomorphized - i.e. it surfaces on `cargo build`
    /// / `cargo test`, not on `cargo check`.
    // `::core::assert!` rather than the crate-wide `assert!`, which maps to
    // `defmt::assert!` under the `defmt` feature and is not const-callable.
    const SLOTS_FIT: () = ::core::assert!(
        N <= crate::persist::MAX_PERSISTED_SUBSCRIPTIONS,
        "the subscriptions table is larger than the reserved persisted-subscription key range"
    );

    /// Persist the whole subscription table to `kv`, one record per key.
    ///
    /// Each subscription is written under its own key
    /// (`PERSISTENT_SUBSCRIPTIONS_START + slot`) so that no single value grows
    /// beyond one subscribe request, and any keys past the current table length
    /// are removed. Rewriting the full range keeps the on-disk set an exact
    /// mirror of the live table without a per-subscription slot allocator;
    /// subscriptions change rarely, so the write amplification is negligible.
    ///
    /// Persistence is best-effort and spec-optional: a subscription whose record does
    /// not fit in `buf` is simply skipped (logged), not treated as an error.
    pub(crate) fn persist_all<'a, B, S>(
        &self,
        buffers: &SubscriptionsBuffers<'a, B, N>,
        mut kv: S,
        buf: &mut [u8],
    ) -> Result<(), Error>
    where
        B: Buffers<IMBuffer> + 'a,
        S: KvBlobStore,
    {
        // Force the slot-range assertion to be evaluated.
        let () = Self::SLOTS_FIT;

        self.with(buffers, |state, buffers| {
            for (slot, (sub, rx)) in state
                .subscriptions
                .iter()
                .zip(buffers.iter())
                .take(N)
                .enumerate()
            {
                let key = PERSISTENT_SUBSCRIPTIONS_START + slot as u16;

                let record = PersistedSubscription {
                    fab_idx: sub.ids.fab_idx,
                    peer_node_id: sub.ids.peer_node_id,
                    min_int_secs: sub.min_int_secs,
                    max_int_secs: sub.max_int_secs,
                    subscribe_req: Octets(rx.as_ref()),
                };

                // Serialize into the front of `buf`, leaving the tail as the store's
                // scratch (mirrors `Persist::store`). If it does not fit, skip it —
                // persisting is optional, so a too-large subscription is simply not
                // resumable across a reboot.
                let mut wb = WriteBuf::new(buf);
                if record.to_tlv(&TLVTag::Anonymous, &mut wb).is_err() {
                    warn!(
                        "Subscription {:?} too large to persist; skipping",
                        sub.ids()
                    );
                    continue;
                }

                let len = wb.get_tail();
                let (data, scratch) = buf.split_at_mut(len);
                kv.store(key, data, scratch)?;
            }

            // Drop the tail keys that a now-shorter table no longer occupies.
            for slot in state.subscriptions.len()..N {
                let key = PERSISTENT_SUBSCRIPTIONS_START + slot as u16;
                kv.remove(key, buf)?;
            }

            Ok(())
        })
    }

    /// Re-hydrate the subscription table from `kv`, replaying each persisted record
    /// through [`Self::add`] exactly as if the subscribe request had just arrived.
    ///
    /// Each reloaded subscription enters the table un-primed, so the reporter sends
    /// it a prompt priming report (establishing a session on demand) rather than
    /// waiting toward its max-interval deadline — see the priming note in the body
    /// for why that timing matters. A record that cannot be re-added (e.g. no free
    /// buffer) is skipped.
    pub(crate) fn load_persist<'a, 's, B, S>(
        &'s self,
        pool: &'a B,
        buffers: &'s SubscriptionsBuffers<'a, B, N>,
        mut kv: S,
        buf: &mut [u8],
        now: Instant,
        event_numbers_watermark: EventNumber,
    ) -> Result<(), Error>
    where
        'a: 's,
        B: Buffers<IMBuffer> + 'a,
        S: KvBlobStore,
    {
        // Force the slot-range assertion to be evaluated.
        let () = Self::SLOTS_FIT;

        for slot in 0..N {
            let key = PERSISTENT_SUBSCRIPTIONS_START + slot as u16;

            let Some(data) = kv.load(key, buf)? else {
                // The range is contiguous: the first empty slot ends the set.
                break;
            };

            let record = PersistedSubscription::from_tlv(&TLVElement::new(data))?;

            let Some(mut rx) = pool.get_immediate() else {
                warn!("No free buffer to resume a persisted subscription; skipping");
                continue;
            };
            rx.clear();
            if rx.extend_from_slice(record.subscribe_req.0).is_err() {
                warn!("Persisted subscription too large for an RX buffer; skipping");
                continue;
            }

            let added = self.add(
                now,
                record.fab_idx,
                record.peer_node_id,
                record.min_int_secs,
                record.max_int_secs,
                event_numbers_watermark,
                rx,
                buffers,
            );

            match added {
                Some(mut rctx) => {
                    // Commit it into the table as an un-primed subscription so the
                    // reporter sends a full priming report to the subscriber right
                    // away (establishing a session on demand), rather than staying
                    // silent until the max-interval liveness point.
                    //
                    // This is essential, not cosmetic: the *subscriber* measures its
                    // own liveness timeout from the last report IT received, not from
                    // our reboot. If we waited toward our own max-interval deadline,
                    // the subscriber — whose clock has been running the whole time we
                    // were down — could already have torn the subscription down. A
                    // prompt priming report resets the subscriber's liveness clock and
                    // re-syncs the state it missed while we were away.
                    //
                    // Leaving `reported_at` at the `Instant::MAX` priming sentinel
                    // (rather than stamping it with `now`) is what marks it un-primed.
                    rctx.next_reported_at = Instant::MAX;
                    rctx.set_keep();
                    info!(
                        "Resumed persisted subscription {:?}",
                        rctx.subscription().ids()
                    );
                }
                None => {
                    warn!("Subscription table full while resuming; dropping a persisted record")
                }
            }
        }

        Ok(())
    }

    /// Remove every persisted subscription record from `kv`.
    ///
    /// Used on a factory reset, and to clear the range before it is re-hydrated.
    pub(crate) fn reset_persist<S>(&self, mut kv: S, buf: &mut [u8]) -> Result<(), Error>
    where
        S: KvBlobStore,
    {
        // Force the slot-range assertion to be evaluated.
        let () = Self::SLOTS_FIT;

        for slot in 0..N {
            let key = PERSISTENT_SUBSCRIPTIONS_START + slot as u16;
            kv.remove(key, buf)?;
        }

        Ok(())
    }
}

impl<const N: usize> Default for Subscriptions<N> {
    fn default() -> Self {
        Self::new()
    }
}

impl<const N: usize> DynBase for Subscriptions<N> {}

/// The inner state of `Subscriptions`, protected by a mutex.
/// See `Subscriptions` for the public API and invariants.
struct SubscriptionsInner<const N: usize> {
    /// Monotonically increasing ID assigned to every accepted subscription.
    /// The first assigned ID is 1; `0` is reserved as the "no subscription" sentinel used by `reporting`.
    next_subscription_id: u32,
    /// The total number of accepted subscriptions, including any currently
    /// in-flight one (i.e. one whose `Subscription` has been moved into a
    /// `ReportContext` and is therefore temporarily not in `subscriptions`).
    /// Used to enforce the `N` capacity bound in `add`.
    subscriptions_count: usize,
    /// The active subscriptions. Does NOT include a subscription that is
    /// currently being reported on; see `reporting` for the snapshot of the
    /// in-flight one.
    subscriptions: Vec<Subscription, N>,
    /// The changed attributes that subscriptions are consulting to decide whether and what they need to report.
    changed_attrs: ChangedAttrs,
    /// Snapshot of the subscription currently being reported on (i.e. the
    /// one that has been `swap_remove`d into a `ReportContext`). `None` when
    /// no report is in flight. This is a frozen clone captured at `report()`
    /// time; mutations made by `ReportContext` (e.g. to
    /// `max_seen_event_number`) are NOT visible here. The slot exists so
    /// that `Subscriptions::remove` can still observe and cancel an
    /// in-flight subscription.
    reporting: Option<Subscription>,
    /// Set by `Subscriptions::remove` when its predicate matched
    /// `reporting`. Consumed by `report_complete`, which then drops the
    /// subscription (and decrements `subscriptions_count`) regardless of the
    /// `keep` flag on the `ReportContext`.
    reporting_cancelled: Option<&'static str>,
}

impl<const N: usize> SubscriptionsInner<N> {
    /// Create the instance.
    #[inline(always)]
    const fn new() -> Self {
        Self {
            next_subscription_id: 1,
            subscriptions_count: 0,
            subscriptions: Vec::new(),
            changed_attrs: ChangedAttrs::new(),
            reporting: None,
            reporting_cancelled: None,
        }
    }

    /// Create an in-place initializer for the instance.
    fn init() -> impl Init<Self> {
        init!(Self {
            next_subscription_id: 1,
            subscriptions_count: 0,
            subscriptions <- Vec::init(),
            changed_attrs <- ChangedAttrs::init(),
            reporting: None,
            reporting_cancelled: None,
        })
    }

    fn clear(&mut self) {
        self.subscriptions.clear();
        self.subscriptions_count = 0;
        // If a report is in flight, make sure `report_complete` drops it
        // rather than pushing it back into an otherwise-empty table.
        if self.reporting.is_some() {
            self.reporting_cancelled = Some("subscriptions cleared");
            // The in-flight subscription is still counted in
            // `subscriptions_count` until `report_complete` runs; restore
            // that so the decrement there balances.
            self.subscriptions_count = 1;
        }
    }

    /// Add a subscription with the given parameters.
    ///
    /// Returns the assigned subscription ID on success, or `None` if the subscription table is full.
    #[allow(clippy::too_many_arguments)]
    fn add<'a, B>(
        &mut self,
        fab_idx: NonZeroU8,
        peer_node_id: u64,
        min_int_secs: u16,
        max_int_secs: u16,
        buffer: B::Buffer<'a>,
        _buffers: &mut SubscriptionsBuffersInner<'a, B, N>,
    ) -> Option<(Subscription, B::Buffer<'a>)>
    where
        B: Buffers<IMBuffer> + 'a,
    {
        if self.subscriptions_count >= N {
            return None;
        }

        self.subscriptions_count += 1;

        let id = self.next_subscription_id;
        self.next_subscription_id += 1;

        // Start with the current watermark so that only changes happening AFTER the
        // subscription was accepted will be reported as incremental updates.
        let max_seen_attr_change_id = self.changed_attrs.watermark();

        let subscription = Subscription {
            ids: SubscriptionIds {
                id,
                fab_idx,
                peer_node_id,
            },
            min_int_secs,
            max_int_secs,
            reported_at: Instant::MAX,
            retry_at: Instant::MIN,
            fail_count: 0,
            max_seen_attr_change_id,
            // Start at 0 so the priming report delivers every event that was
            // already in the event buffer at subscribe time. The reader will
            // advance this via `update_max_seen_event_number` once the
            // priming report has consumed the events.
            max_seen_event_number: 0,
        };

        info!("Added subscription {:?}", subscription.ids());

        Some((subscription, buffer))
    }

    /// Begin a report for the subscription with the given ID.
    ///
    /// Returns a small [`ReportContext`] capturing the subscription's current
    /// `since` watermark and the watermark to commit via [`Self::mark_reported`]
    /// on success. Unlike a snapshot, the `changed_attrs` table itself is not
    /// copied; the report uses a [`SubAttrChangeFilter`] that consults the
    /// live table one attribute at a time.
    ///
    /// `priming = true` produces a context with filtering disabled; it is
    /// used for the initial ("priming") report delivered right after a
    /// subscription is accepted.
    fn report<'a, B>(
        &mut self,
        now: Instant,
        event_numbers_watermark: EventNumber,
        buffers: &mut SubscriptionsBuffersInner<'a, B, N>,
    ) -> Option<(Subscription, B::Buffer<'a>)>
    where
        B: Buffers<IMBuffer> + 'a,
    {
        // `reporting` must be vacant: callers only start a new report after
        // the previous `ReportContext` has been dropped (which clears the
        // slot via `report_complete`).
        debug_assert!(self.reporting.is_none());
        debug_assert!(self.reporting_cancelled.is_none());

        if let Some(index) = self.find_reportable::<B>(now, event_numbers_watermark, buffers) {
            let sub = self.subscriptions.swap_remove(index);
            let buf = buffers.swap_remove(index);

            debug!("About to report on subscription {:?}", sub.ids());

            // Leave a snapshot clone behind so that `Subscriptions::remove`
            // can still match and cancel this subscription while the report
            // is in flight.
            self.reporting = Some(sub.clone());

            Some((sub, buf))
        } else {
            None
        }
    }

    fn report_complete<'a, B>(
        &mut self,
        sub: Subscription,
        buffer: B::Buffer<'a>,
        buffers: &mut SubscriptionsBuffersInner<'a, B, N>,
        keep: bool,
    ) where
        B: Buffers<IMBuffer> + 'a,
    {
        // Always clear the reporting slot; it was populated in `report()`.
        self.reporting = None;
        let cancelled = self.reporting_cancelled.take();

        if let Some(reason) = cancelled {
            info!(
                "In-flight subscription {:?} cancelled during reporting: {}",
                sub.ids(),
                reason
            );
            self.subscriptions_count -= 1;
        } else if keep {
            debug!("Subscription {:?} kept after reporting; max-attr-change-id: {}, max-seen-event-number: {}", sub.ids(), sub.max_seen_attr_change_id, sub.max_seen_event_number);

            unwrap!(self.subscriptions.push(sub));
            unwrap!(buffers.push(buffer).map_err(|_| ()));
        } else {
            warn!("Subscription {:?} removed during reporting", sub.ids());
            self.subscriptions_count -= 1;
        }
    }

    fn find_reportable<'a, B>(
        &self,
        now: Instant,
        event_numbers_watermark: EventNumber,
        buffers: &SubscriptionsBuffersInner<'a, B, N>,
    ) -> Option<usize>
    where
        B: Buffers<IMBuffer> + 'a,
    {
        self.subscriptions
            .iter()
            .enumerate()
            .map(|(index, sub)| (sub, &buffers[index]))
            .position(|(sub, rx)| {
                sub.is_reportable(now, rx, &self.changed_attrs, event_numbers_watermark)
            })
    }

    /// Remove entries that every subscription has already reported on.
    fn purge_reported_changes(&mut self) {
        if let Some(min_seen_attr_change_id) = self
            .subscriptions
            .iter()
            .map(|s| s.max_seen_attr_change_id)
            .min()
        {
            self.changed_attrs.purge_up_to(min_seen_attr_change_id);
        } else {
            self.changed_attrs.clear();
        }
    }

    /// Earliest [`Instant`] at which any subscription will next need servicing,
    /// or [`Instant::MAX`] if none has a wake point (empty table or all
    /// not-yet-primed).
    fn next_report_at<'a, B>(
        &self,
        event_numbers_watermark: EventNumber,
        buffers: &SubscriptionsBuffersInner<'a, B, N>,
    ) -> Instant
    where
        B: Buffers<IMBuffer> + 'a,
    {
        self.subscriptions
            .iter()
            .enumerate()
            .map(|(index, sub)| {
                sub.next_report_at(
                    &buffers[index],
                    &self.changed_attrs,
                    event_numbers_watermark,
                )
            })
            .min()
            .unwrap_or(Instant::MAX)
    }
}

/// The IDs of a subscription, used to identify it across the system and to route reports to it.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub struct SubscriptionIds {
    /// The ID of the subscription. Uniquely identifies the subscription across all of them.
    pub id: u32,
    /// The fabric index of the subscriber. Used to route reports and to remove all subscriptions of a fabric when it gets removed.
    pub fab_idx: NonZeroU8,
    /// The node ID of the subscriber. Used to route reports and to remove all subscriptions of a peer when it gets removed.
    pub peer_node_id: NodeId,
}

/// The on-disk form of a single subscription.
///
/// One record is stored per subscription under its own key (see
/// [`PERSISTENT_SUBSCRIPTIONS_START`]), so the value never grows beyond a single
/// subscribe request. Everything the reporter needs to resume the subscription
/// is captured here: the routing `(fab_idx, peer_node_id)`, the negotiated
/// intervals, and the raw `SubscribeReq` TLV — the same bytes the live
/// subscription keeps in its RX buffer and re-parses on every report, so no
/// separate path list is serialized.
#[cfg(feature = "persistent-subscriptions")]
#[derive(Debug, FromTLV, ToTLV)]
#[tlvargs(lifetime = "'a")]
struct PersistedSubscription<'a> {
    fab_idx: NonZeroU8,
    peer_node_id: NodeId,
    min_int_secs: u16,
    max_int_secs: u16,
    /// The raw `SubscribeReq` TLV that created this subscription (its selected
    /// attribute/event paths and fabric-filtered flag live inside it).
    subscribe_req: OctetStr<'a>,
}

#[derive(Clone, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
pub struct Subscription {
    /// The IDs of the subscription
    ids: SubscriptionIds,
    /// The minimum interval in seconds. The subscription should not receive reports more frequently than this interval, but may receive them less frequently.
    /// We use u16 instead of embassy::Duration to save some storage
    min_int_secs: u16,
    /// The maximum interval in seconds. The subscription should receive reports at least this frequently, even if there are no changes to report (i.e. it is a liveness deadline).
    /// We use u16 instead of embassy::Duration to save some storage
    max_int_secs: u16,
    /// The timestamp of the last SUCCESSFUL report sent to this subscription. Used to decide when the next report is due based on the min/max intervals — and, crucially, when to give up: `is_expired` measures `max_int` from here, so a run of *failed* reports (which do NOT advance it) eventually expires the subscription rather than retrying forever.
    /// Set to `Instant::MAX` when the subscription is created to indicate that no report has been sent yet, so the first report is due immediately. After the first successful report, it is updated to the actual timestamp of that report.
    reported_at: Instant,
    /// Earliest instant at which a report may be attempted again after a *failed*
    /// send (see [`ReportContext::set_keep_retry`]). `Instant::MIN` means no retry
    /// is pending (normal operation). Gates the report-timing helpers so a peer we
    /// cannot reach is retried with a growing back-off instead of in a tight loop.
    retry_at: Instant,
    /// Number of consecutive failed report attempts, driving the retry back-off.
    /// Reset to `0` on any successful report.
    fail_count: u8,
    /// The largest attribute change ID from the [`ChangedAttributes`] table this subscription
    /// has already reported on. Entries with a larger change ID represent pending changes the subscription still needs to emit.
    max_seen_attr_change_id: u64,
    /// The largest event number this subscription has already reported on. Events with a larger event number represent pending events the subscription still needs to emit.
    max_seen_event_number: u64,
}

impl Subscription {
    /// Return the IDs of the subscription.
    pub const fn ids(&self) -> &SubscriptionIds {
        &self.ids
    }

    /// Return `true` if the subscription is expired and should be removed, or `false` if it is still active.
    pub fn is_expired(&self, now: Instant) -> bool {
        self.reported_at
            .checked_add(embassy_time::Duration::from_secs(self.max_int_secs as _))
            .map(|expiry| expiry <= now)
            .unwrap_or(false)
    }

    /// The back-off (in seconds) to wait before the `fail_count`-th consecutive
    /// retry of a failed report.
    ///
    /// Exponential — `BASE * 2^(fail_count - 1)` — capped at `max_int`: it never
    /// helps to wait longer than the point at which [`Self::is_expired`] gives up
    /// on the subscription. The cap also means the subscription is guaranteed to
    /// expire (from the pinned `reported_at`) rather than back off indefinitely.
    fn retry_backoff_secs(fail_count: u8, max_int_secs: u16) -> u16 {
        /// First retry delay. Small enough to recover quickly from a transient
        /// blip, large enough not to hammer an unreachable peer.
        const BASE_SECS: u16 = 2;

        let shift = fail_count.saturating_sub(1).min(15);
        let delay = (BASE_SECS as u32) << shift;

        delay.min(max_int_secs.max(BASE_SECS) as u32) as u16
    }

    /// Return `true` if the subscription is due for a report based on the given parameters, or `false` if it is not.
    fn is_reportable(
        &self,
        now: Instant,
        rx: &[u8],
        changed_attrs: &ChangedAttrs,
        event_numbers_watermark: EventNumber,
    ) -> bool {
        if !self.is_report_allowed(now) {
            return false;
        }

        self.is_report_due(now)
            || self.is_affected_by_attr_changes(rx, changed_attrs)
            || self.is_affected_by_new_events(rx, event_numbers_watermark)
    }

    /// Instant at which the min-interval quiet period ends — the earliest time
    /// a report is allowed ([`Self::is_report_allowed`] returns `true`).
    ///
    /// [`Instant::MIN`] when not yet primed (`reported_at == Instant::MAX`): a
    /// fresh subscription is allowed to report immediately (its priming report),
    /// and a point infinitely in the past reads correctly as "always allowed"
    /// for every consumer (the boolean gate below and `next_report_at`).
    fn report_allowed_at(&self) -> Instant {
        // A pending retry after a failed send is a hard floor on the next
        // attempt, even for a not-yet-primed subscription — otherwise a peer we
        // cannot reach would be retried in a tight loop.
        let min_int_gate = if self.reported_at == Instant::MAX {
            // Not yet primed: always allowed. This must be an explicit check, not
            // a reliance on `checked_add` saturating — with `min_int_secs == 0`
            // the add is `Instant::MAX + 0`, which does NOT overflow and would
            // wrongly yield `Instant::MAX` ("never allowed").
            Instant::MIN
        } else {
            self.reported_at
                .checked_add(embassy_time::Duration::from_secs(self.min_int_secs as _))
                .unwrap_or(Instant::MIN)
        };

        min_int_gate.max(self.retry_at)
    }

    /// Return `true` if the subscription is allowed to report based on the min interval, or `false` if it is still in the quiet period since the last report.
    fn is_report_allowed(&self, now: Instant) -> bool {
        self.report_allowed_at() <= now
    }

    /// Instant at which the max-interval liveness window opens — the earliest
    /// time [`Self::is_report_due`] returns `true`.
    ///
    /// `reported_at + max_int - max_int / 2`, i.e. the half-interval mark.
    /// Waking before the negotiated maximum interval is this implementation's
    /// margin for completing the report in time.
    ///
    /// [`Instant::MIN`] when not yet primed (`reported_at == Instant::MAX`): a
    /// fresh subscription is immediately due for its priming report (see
    /// [`Self::report_allowed_at`] for why `MIN` is the right sentinel).
    fn report_due_at(&self) -> Instant {
        // Not yet primed: always due (explicit, for the same reason as
        // `report_allowed_at`).
        if self.reported_at == Instant::MAX {
            return Instant::MIN;
        }

        self.reported_at
            .checked_add(embassy_time::Duration::from_secs(
                (self.max_int_secs - self.max_int_secs / 2) as _,
            ))
            .unwrap_or(Instant::MIN)
    }

    /// Return `true` if the subscription is due for a report based on the max interval, or `false` if it is not yet due.
    fn is_report_due(&self, now: Instant) -> bool {
        self.report_due_at() <= now
    }

    /// Return `true` if the subscription is affected by changes to the attribute triple `(endpoint, cluster, attr)` based on the subscription's RX and the given table of changed attributes, or `false` if it is not affected.
    fn is_affected_by_attr_changes(&self, _rx: &[u8], changes: &ChangedAttrs) -> bool {
        // NOTE: we could consult the subscription's RX here to skip the check if the subscription
        // is not interested in the changed path at all, but that would require parsing the RX at every report check,
        // which is anyway done later during reporting and the report is canceled if empty
        //
        // Therefore and for now do not to this here
        changes.any_since(self.max_seen_attr_change_id)
    }

    /// Return `true` if the subscription is affected by new events based on the subscription's RX and the given event numbers watermark, or `false` if it is not affected.
    fn is_affected_by_new_events(&self, _rx: &[u8], event_numbers_watermark: EventNumber) -> bool {
        // NOTE: we could consult the subscription's RX here to skip the check if the subscription
        // is not interested in events at all, but that would require parsing the RX at every report check,
        // which is anyway done later during reporting and the report is canceled if empty
        //
        // Therefore and for now do not to this here
        self.max_seen_event_number < event_numbers_watermark
    }

    /// Earliest [`Instant`] at which this subscription could next report.
    ///
    /// A not-yet-primed subscription (`reported_at == Instant::MAX`) yields
    /// [`Instant::MIN`] via both deadline helpers — "report now", so the reporter
    /// wakes immediately to deliver the priming report.
    ///
    /// Never earlier than the min-interval gate `reported_at + min_int`, before
    /// which a report SHALL NOT be sent (Matter spec). Subject to that
    /// gate, it is:
    /// - `reported_at + min_int` when a change or event is already pending, so
    ///   the wake lands at the end of the quiet period; otherwise
    /// - the liveness point ([`Self::report_due_at`]), when
    ///   [`Self::is_report_due`] flips — early enough for the report to be
    ///   received before `max_int` (the subscriber terminates otherwise).
    ///
    /// Clamping the liveness point up to the gate avoids a busy-spin when
    /// `min_int > max_int / 2`.
    fn next_report_at(
        &self,
        rx: &[u8],
        changed_attrs: &ChangedAttrs,
        event_numbers_watermark: EventNumber,
    ) -> Instant {
        let allowed_at = self.report_allowed_at();

        // Use the same `rx` the report path feeds `is_reportable`, so this
        // prediction stays faithful if these checks ever start consulting it.
        let pending = self.is_affected_by_attr_changes(rx, changed_attrs)
            || self.is_affected_by_new_events(rx, event_numbers_watermark);

        if pending {
            allowed_at
        } else {
            allowed_at.max(self.report_due_at())
        }
    }
}

/// A table of recently-changed attribute triples, each tagged with an
/// ever-increasing `change_id`.
///
/// Subscriptions consult this table to decide which attributes they should
/// re-emit on their next report: only attributes with a matching entry whose
/// `change_id` is strictly greater than the subscription's own watermark
/// (`last_change_id`) need to be reported.
///
/// The table has a fixed capacity of [`MAX_CHANGED_ATTRS`] entries. When it
/// fills up, existing entries are coalesced to coarser-grained wildcards
/// (`(endpoint, cluster, *)` → `(endpoint, *, *)` → `(*, *, *)`) so that a new change can always
/// be recorded. A wildcard entry over-covers and will therefore cause the
/// affected subscriptions to emit a slightly wider set of attributes on their
/// next report, but this is a bounded loss of precision that preserves
/// correctness.
pub(crate) struct ChangedAttrs {
    /// Monotonically increasing ID assigned to every recorded change.
    /// The first assigned ID is 1; `0` is reserved as the "no change seen yet"
    /// sentinel used by fresh subscriptions.
    next_change_id: u64,
    /// The actual table of recent changes, ordered from oldest to newest.
    /// The newest change has `change_id == next_change_id - 1`.
    entries: Vec<ChangedAttr, MAX_CHANGED_ATTRS>,
}

impl ChangedAttrs {
    /// Create the instance.
    #[inline(always)]
    const fn new() -> Self {
        Self {
            next_change_id: 1,
            entries: Vec::new(),
        }
    }

    /// Return an in-place initializer for the instance.
    fn init() -> impl Init<Self> {
        init!(Self {
            next_change_id: 1,
            entries <- Vec::init(),
        })
    }

    /// The largest change ID that has been assigned so far. A subscription
    /// whose max seen change ID is equal to the watermark has seen every change.
    #[inline]
    fn watermark(&self) -> u64 {
        self.next_change_id.wrapping_sub(1)
    }

    /// Record a change to the attribute triple `(endpoint, cluster, attr)`.
    /// Returns the newly assigned change ID.
    fn record(&mut self, endpoint: EndptId, cluster: ClusterId, attr: AttrId) -> u64 {
        self.record_raw(ChangedAttr::concrete(endpoint, cluster, attr, 0))
    }

    /// Record a cluster- or endpoint-wide wildcard change. `endpoint == None`
    /// together with `cluster == None` represents a global wildcard.
    /// Returns the newly assigned change ID.
    fn record_wildcard(&mut self, endpoint: Option<EndptId>, cluster: Option<ClusterId>) -> u64 {
        self.record_raw(ChangedAttr {
            endpoint: endpoint.unwrap_or(WILDCARD_ENDPOINT),
            cluster: cluster.unwrap_or(WILDCARD_CLUSTER),
            attr: WILDCARD_ATTR,
            change_id: 0,
        })
    }

    /// Insert `new` into the table. The caller is expected to leave `new.change_id`
    /// at any value - it is overwritten by a freshly-assigned ID.
    fn record_raw(&mut self, mut new: ChangedAttr) -> u64 {
        let change_id = self.next_change_id;
        self.next_change_id = self.next_change_id.wrapping_add(1).max(1);
        new.change_id = change_id;

        // If an existing entry already covers `new`, just refresh its change ID.
        if let Some(existing) = self.entries.iter_mut().find(|x| x.covers(&new)) {
            existing.change_id = change_id;
            return change_id;
        }

        // `new` may itself subsume existing concrete entries - drop those to
        // keep the table compact and avoid wasting slots on redundant paths.
        let mut i = 0;
        while i < self.entries.len() {
            if new.covers(&self.entries[i]) {
                self.entries.swap_remove(i);
            } else {
                i += 1;
            }
        }

        if let Err(new) = self.entries.push(new) {
            // The table is full - promote entries to coarser wildcards to free a slot.
            self.promote_and_insert(new);
        }

        change_id
    }

    /// Returns `true` if the table contains at least one entry covering
    /// `(endpoint, cluster, attr)` with `change_id > since`.
    fn contains_since(
        &self,
        endpoint: EndptId,
        cluster: ClusterId,
        attr: AttrId,
        since: u64,
    ) -> bool {
        self.entries
            .iter()
            .any(|x| x.change_id > since && x.matches(endpoint, cluster, attr))
    }

    /// Returns `true` if the table contains at least one entry with
    /// `change_id > since` (of any path).
    fn any_since(&self, since: u64) -> bool {
        self.entries.iter().any(|x| x.change_id > since)
    }

    /// Drop all entries with `change_id <= threshold`.
    fn purge_up_to(&mut self, threshold: u64) {
        if threshold == 0 {
            return;
        }

        let mut i = 0;
        while i < self.entries.len() {
            if self.entries[i].change_id <= threshold {
                self.entries.swap_remove(i);
            } else {
                i += 1;
            }
        }
    }

    /// Drop every recorded change. Used when no subscriptions exist.
    fn clear(&mut self) {
        self.entries.clear();
    }

    /// Coalesce existing entries to coarser wildcards so that `new` can be inserted.
    ///
    /// The strategy is to promote as little as possible: on each iteration we
    /// collapse the single largest collapsible group at the finest available
    /// level into one coarser wildcard entry, freeing at least one slot. Only
    /// once no fine-grained group of two or more entries exists do we escalate
    /// to the next level, and finally to a global wildcard as a last resort.
    fn promote_and_insert(&mut self, new: ChangedAttr) {
        loop {
            // If an existing (possibly just-promoted) entry already covers `new`,
            // refresh its change ID and we're done.
            if let Some(existing) = self.entries.iter_mut().find(|x| x.covers(&new)) {
                existing.change_id = new.change_id;
                return;
            }

            if self.entries.push(new.clone()).is_ok() {
                return;
            }

            // Full - promote exactly one group at the finest granularity that
            // actually yields compaction. Levels:
            // - 1: (endpoint, cluster, *)
            // - 2: (endpoint, *, *)
            if !self.promote_largest_group(1) && !self.promote_largest_group(2) {
                // No collapsible group at either level - last-ditch fallback:
                // collapse the whole table into a single global wildcard entry.
                self.entries.clear();

                unwrap!(self.entries.push(ChangedAttr {
                    endpoint: WILDCARD_ENDPOINT,
                    cluster: WILDCARD_CLUSTER,
                    attr: WILDCARD_ATTR,
                    change_id: new.change_id,
                }));

                return;
            }
        }
    }

    /// Find the largest group of entries (>= 2) that share the same key at the
    /// given promotion level, and collapse it into one coarser wildcard entry.
    ///
    /// Returns `true` if any promotion happened.
    fn promote_largest_group(&mut self, level: u8) -> bool {
        // Pick a pivot whose group is largest.
        let mut best_pivot: Option<ChangedAttr> = None;
        let mut best_count = 1usize;

        for i in 0..self.entries.len() {
            let pivot = &self.entries[i];
            let Some(coarsened) = pivot.coarsen(level) else {
                continue;
            };

            let count = self.entries.iter().filter(|e| coarsened.covers(e)).count();
            if count > best_count {
                best_count = count;
                best_pivot = Some(pivot.clone());
            }
        }

        let Some(pivot) = best_pivot else {
            return false;
        };
        // `coarsen` already returned `Some` above for this pivot.
        let mut coarsened = pivot.coarsen(level).unwrap();

        // Remove all entries covered by `coarsened`, keeping the largest
        // change_id to preserve recency.
        let mut max_change_id = 0u64;
        let mut i = 0;
        while i < self.entries.len() {
            if coarsened.covers(&self.entries[i]) {
                if self.entries[i].change_id > max_change_id {
                    max_change_id = self.entries[i].change_id;
                }
                self.entries.swap_remove(i);
            } else {
                i += 1;
            }
        }
        coarsened.change_id = max_change_id;
        // Safe: we just removed `best_count >= 2` entries, so there is room.
        unwrap!(self.entries.push(coarsened));
        true
    }
}

/// Sentinel value for "any endpoint" inside a [`ChangedAttr`] entry.
///
/// Matter endpoint ids are `u16`; the Matter Core Specification caps practical
/// endpoint numbering well below `0xFFFF`, and the CHIP reference SDK
/// (`kInvalidEndpointId` in `src/lib/core/DataModelTypes.h`) adopts the same
/// convention, so we can repurpose `u16::MAX` as an internal "wildcard" marker.
const WILDCARD_ENDPOINT: EndptId = EndptId::MAX;

/// Sentinel value for "any cluster" inside a [`ChangedAttr`] entry.
///
/// Matter cluster ids are Manufacturer Extensible Identifiers (MEIs, Core Spec):
/// `(vendor_prefix << 16) | suffix` with `0xFFFF` reserved as an
/// invalid vendor prefix. `0xFFFF_FFFF` therefore cannot be a legitimate
/// cluster id and is safe to use as an internal "wildcard" marker. The CHIP
/// reference SDK uses the same value as `kInvalidClusterId`.
const WILDCARD_CLUSTER: ClusterId = ClusterId::MAX;

/// Sentinel value for "any attribute" inside a [`ChangedAttr`] entry.
///
/// Same MEI argument as [`WILDCARD_CLUSTER`]: `0xFFFF_FFFF` cannot be a
/// legitimate attribute id and matches CHIP's `kInvalidAttributeId`.
const WILDCARD_ATTR: AttrId = AttrId::MAX;

/// A record of one recently changed attribute.
///
/// A field holding its corresponding `WILDCARD_*` sentinel acts as a wildcard
/// on that axis. Wildcards appear only as a result of "promotion" when the
/// `changed_attrs` table becomes full and several concrete entries need to be
/// coalesced into a coarser one.
///
/// Rust is free to reorder these fields under the default `repr(Rust)`, and
/// it does so to minimize size: on 64-bit targets `size_of::<ChangedAttr>()`
/// is 24 bytes (the `u64` change_id forces 8-byte alignment; the rest packs
/// into the remaining 16 bytes). The previous `Option<u16> / Option<u32> /
/// Option<u32>` encoding took 32 bytes per entry because `u16` / `u32` have
/// no niche for `Option`. See `changed_attr_size_is_compact`.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "defmt", derive(defmt::Format))]
struct ChangedAttr {
    endpoint: EndptId,
    cluster: ClusterId,
    attr: AttrId,
    change_id: u64,
}

impl ChangedAttr {
    /// Create a concrete (non-wildcard) entry with the given parameters and change ID.
    const fn concrete(endpoint: EndptId, cluster: ClusterId, attr: AttrId, change_id: u64) -> Self {
        Self {
            endpoint,
            cluster,
            attr,
            change_id,
        }
    }

    /// Return `true` if this entry is a wildcard on the endpoint axis, or `false` if it is concrete.
    #[inline]
    const fn is_endpoint_wildcard(&self) -> bool {
        self.endpoint == WILDCARD_ENDPOINT
    }

    /// Return `true` if this entry is a wildcard on the cluster axis, or `false` if it is concrete.
    #[inline]
    const fn is_cluster_wildcard(&self) -> bool {
        self.cluster == WILDCARD_CLUSTER
    }

    /// Return `true` if this entry is a wildcard on the attribute axis, or `false` if it is concrete.
    #[inline]
    const fn is_attr_wildcard(&self) -> bool {
        self.attr == WILDCARD_ATTR
    }

    /// Whether this record covers the concrete attribute triple
    /// `(endpoint, cluster, attr)`.
    fn matches(&self, endpoint: EndptId, cluster: ClusterId, attr: AttrId) -> bool {
        (self.is_endpoint_wildcard() || self.endpoint == endpoint)
            && (self.is_cluster_wildcard() || self.cluster == cluster)
            && (self.is_attr_wildcard() || self.attr == attr)
    }

    /// Whether `other` is semantically covered by `self` (i.e. `self` is as
    /// coarse as or coarser than `other` on every axis).
    fn covers(&self, other: &ChangedAttr) -> bool {
        #[inline]
        fn cov<T: Eq>(a: T, a_wild: bool, b: T, b_wild: bool) -> bool {
            if a_wild {
                true // self wildcard covers anything
            } else if b_wild {
                false // concrete doesn't cover wildcard
            } else {
                a == b
            }
        }
        cov(
            self.endpoint,
            self.is_endpoint_wildcard(),
            other.endpoint,
            other.is_endpoint_wildcard(),
        ) && cov(
            self.cluster,
            self.is_cluster_wildcard(),
            other.cluster,
            other.is_cluster_wildcard(),
        ) && cov(
            self.attr,
            self.is_attr_wildcard(),
            other.attr,
            other.is_attr_wildcard(),
        )
    }

    /// Build the coarsened wildcard entry representing `pivot`'s group at the
    /// given level. Returns `None` if `pivot` cannot be promoted at that level
    /// (e.g. its endpoint is already a wildcard for level 1 or 2).
    fn coarsen(&self, level: u8) -> Option<Self> {
        match level {
            1 => {
                if self.is_endpoint_wildcard() || self.is_cluster_wildcard() {
                    return None;
                }
                Some(Self {
                    change_id: 0,
                    cluster: self.cluster,
                    attr: WILDCARD_ATTR,
                    endpoint: self.endpoint,
                })
            }
            2 => {
                if self.is_endpoint_wildcard() {
                    return None;
                }
                Some(Self {
                    change_id: 0,
                    cluster: WILDCARD_CLUSTER,
                    attr: WILDCARD_ATTR,
                    endpoint: self.endpoint,
                })
            }
            _ => unreachable!(),
        }
    }
}

/// Per-subscription context for an in-progress report.
pub struct ReportContext<'a, 's, B, const N: usize>
where
    B: Buffers<IMBuffer> + 'a,
{
    /// A reference to the global subscriptions table, used to return the subscription on
    /// successful completion of the report
    subscriptions: &'s Subscriptions<N>,
    /// A reference to the global subscription buffers, used to return the subscription buffer on
    /// successful completion of the report
    subscriptions_buffers: &'s SubscriptionsBuffers<'a, B, N>,
    /// The subscription being reported on.
    subscription: Option<Subscription>,
    /// The RX buffer with report data associated with the subscription being reported on.
    subscription_buffer: Option<B::Buffer<'a>>,
    /// The next maximum seen attribute change ID for the subscription
    /// to be updated into it upon returning the subscription to the table.
    ///
    /// This is captured here because the subscription's own `max_seen_attr_change_id`
    /// is not updated until the report completes as it is until then still used.
    next_max_seen_attr_change_id: u64,
    /// The next maximum seen event number for the subscription
    /// to be updated into it upon returning the subscription to the table.
    ///
    /// This is captured here because the subscription's own `max_seen_event_number`
    /// is not updated until the report completes as it is until then still used.
    next_max_seen_event_number: EventNumber,
    /// The next reported timestamp for the subscription,
    /// to be updated into it upon returning the subscription to the table.
    ///
    /// This is captured here because the subscription's own `next_reported_at`
    /// is not updated until the report completes as it is until then still used.
    next_reported_at: Instant,
    /// The next retry-gate to commit onto the subscription on completion.
    /// `Instant::MIN` (the default) clears any pending retry; a failed send
    /// ([`Self::set_keep_retry`]) sets it to a backed-off future instant.
    next_retry_at: Instant,
    /// The next consecutive-failure count to commit. `0` (the default) clears the
    /// back-off; a failed send sets it to one more than the subscription's.
    next_fail_count: u8,
    /// Whether the subscription should be kept in the table after the report completes.
    /// Set by the report handler if the other peer acknowledges the data reported by the subscription.
    keep: bool,
}

impl<'a, 's, B, const N: usize> ReportContext<'a, 's, B, N>
where
    B: Buffers<IMBuffer> + 'a,
{
    /// Return a reference to the subscription being reported on.
    pub fn subscription(&self) -> &Subscription {
        unwrap!(self.subscription.as_ref())
    }

    /// Return a reference to the RX buffer associated with the subscription being reported on.
    pub fn rx(&self) -> &[u8] {
        unwrap!(self.subscription_buffer.as_ref()).as_ref()
    }

    /// Return `true` if the report should be sent even if it turns out to be empty
    /// (i.e. no attributes or events to report), or `false` if it can be skipped in that case.
    pub fn should_send_if_empty(&self) -> bool {
        // A fresh subscription has `reported_at == Instant::MAX`, which makes
        // `report_due_at` saturate to `Instant::MIN` and `is_report_due` return
        // `true`, so priming reports are delivered unconditionally without a
        // separate `priming` flag.
        unwrap!(self.subscription.as_ref()).is_report_due(self.next_reported_at)
    }

    /// Return `true` if the subscription should report the attribute
    /// identified by the given triple, or `false` if it can skip it.
    pub fn should_report_attr(
        &self,
        endpoint_id: EndptId,
        cluster_id: ClusterId,
        attr_id: AttrId,
    ) -> bool {
        let sub = self.subscription();

        // A fresh subscription (priming report) has never reported anything
        // yet; its `reported_at` sentinel doubles as the "priming" marker and
        // means every selected attribute must be delivered, regardless of
        // whether it appears in `changed_attrs`.
        if sub.reported_at == Instant::MAX {
            return true;
        }

        self.subscriptions.state.lock(|state| {
            state.borrow().changed_attrs.contains_since(
                endpoint_id,
                cluster_id,
                attr_id,
                sub.max_seen_attr_change_id,
            )
        })
    }

    /// Return the maximum event number the subscription has seen so far.
    pub fn max_seen_event_number(&self) -> EventNumber {
        unwrap!(self.subscription.as_ref()).max_seen_event_number
    }

    /// Return the next maximum event number to be updated into the subscription upon returning it to the table.
    pub fn next_max_seen_event_number(&self) -> EventNumber {
        self.next_max_seen_event_number
    }

    /// Mark the subscription to be kept in the table after the report completes,
    /// meaning the other peer acknowledged our report.
    pub fn set_keep(&mut self) {
        self.keep = true;
    }

    /// Keep the subscription in the table after a *failed* send to the peer, so it
    /// retries — with a back-off, and without advancing its watermarks or its
    /// last-success timestamp.
    ///
    /// Three things happen, each of which matters:
    /// - **Watermarks are preserved.** [`Self::set_keep`] is only correct after a
    ///   delivered report: `report()` captured the current change/event watermarks
    ///   as "what this report covers", and committing them would mark the
    ///   still-unsent changes/events as seen — they would then never be reported.
    ///   We restore the last actually-reported values so the pending data is
    ///   re-attempted.
    /// - **`reported_at` is preserved** (not advanced to now). `is_expired`
    ///   measures `max_int` from `reported_at`, so keeping it pinned to the last
    ///   *successful* report means a run of failures eventually expires the
    ///   subscription instead of retrying it forever.
    /// - **A back-off is scheduled.** `retry_at` is pushed to a growing (capped)
    ///   delay from now, so an unreachable peer is retried with exponential
    ///   back-off rather than in a tight loop that would flood the network.
    pub fn set_keep_retry(&mut self) {
        // Snapshot the values we need off the (borrowed) subscription first, so
        // the writes below don't conflict with the borrow.
        let sub = self.subscription();
        let (last_attr, last_event, last_reported_at, max_int_secs, fail_count) = (
            sub.max_seen_attr_change_id,
            sub.max_seen_event_number,
            sub.reported_at,
            sub.max_int_secs,
            sub.fail_count.saturating_add(1),
        );

        // The report context's "now" — the instant of *this* (failed) attempt —
        // was captured into `next_reported_at` at construction. Read it before we
        // overwrite that field with the preserved last-success timestamp below.
        let now = self.next_reported_at;

        // Preserve watermarks + last-success timestamp (so pending data is
        // re-attempted and `is_expired` still measures from the last success).
        self.next_max_seen_attr_change_id = last_attr;
        self.next_max_seen_event_number = last_event;
        self.next_reported_at = last_reported_at;

        // Grow the back-off from now. The delay is capped at the subscription's
        // max interval: it never makes sense to wait longer than the point at
        // which `is_expired` gives up on the subscription anyway.
        self.next_fail_count = fail_count;
        let backoff_secs = Subscription::retry_backoff_secs(fail_count, max_int_secs);
        self.next_retry_at = now
            .checked_add(embassy_time::Duration::from_secs(backoff_secs as _))
            .unwrap_or(Instant::MAX);

        self.keep = true;
    }
}

impl<'a, 's, B, const N: usize> Drop for ReportContext<'a, 's, B, N>
where
    B: Buffers<IMBuffer> + 'a,
{
    fn drop(&mut self) {
        self.subscriptions.report_complete(self);
    }
}

#[cfg(test)]
mod tests {
    use crate::utils::storage::pooled::PooledBuffers;

    use super::*;

    use embassy_time::Duration;

    type TestPool<const N: usize> = PooledBuffers<IMBuffer, N>;

    // ---------- ChangedAttributes ----------

    #[test]
    fn changed_attrs_starts_empty() {
        let attrs = ChangedAttrs::new();
        assert_eq!(attrs.watermark(), 0);
        assert!(!attrs.any_since(0));
        assert!(!attrs.contains_since(1, 2, 3, 0));
    }

    #[test]
    fn changed_attrs_record_assigns_monotonic_ids() {
        let mut attrs = ChangedAttrs::new();
        let id1 = attrs.record(1, 2, 3);
        let id2 = attrs.record(1, 2, 4);
        let id3 = attrs.record(2, 2, 3);
        assert_eq!(id1, 1);
        assert_eq!(id2, 2);
        assert_eq!(id3, 3);
        assert_eq!(attrs.watermark(), 3);
    }

    #[test]
    fn changed_attrs_contains_since_and_any_since() {
        let mut attrs = ChangedAttrs::new();
        attrs.record(1, 2, 3);
        attrs.record(1, 2, 4);

        assert!(attrs.any_since(0));
        assert!(attrs.any_since(1));
        assert!(!attrs.any_since(2));

        assert!(attrs.contains_since(1, 2, 3, 0));
        assert!(attrs.contains_since(1, 2, 4, 1));
        // After watermark 2 there are no more changes
        assert!(!attrs.contains_since(1, 2, 3, 2));
        // A never-recorded triple is not covered
        assert!(!attrs.contains_since(9, 9, 9, 0));
    }

    #[test]
    fn changed_attrs_duplicate_refreshes_change_id() {
        let mut attrs = ChangedAttrs::new();
        attrs.record(1, 2, 3);
        attrs.record(1, 2, 4);
        // Same triple as first record - should refresh, not add a new entry.
        let id3 = attrs.record(1, 2, 3);
        assert_eq!(id3, 3);
        assert_eq!(attrs.entries.len(), 2);
        // The (1, 2, 3) entry now has change_id 3, so it is visible from since=2
        assert!(attrs.contains_since(1, 2, 3, 2));
        // But it was originally at id=1, which is now lost - `since=0` still sees it
        // through the refreshed id.
        assert!(attrs.contains_since(1, 2, 3, 0));
    }

    #[test]
    fn changed_attrs_record_wildcard_cluster_covers_every_attr() {
        let mut attrs = ChangedAttrs::new();
        let id = attrs.record_wildcard(Some(7), Some(42));

        // Any concrete attribute on that (endpoint, cluster) is now covered.
        assert!(attrs.contains_since(7, 42, 0, 0));
        assert!(attrs.contains_since(7, 42, 1, 0));
        assert!(attrs.contains_since(7, 42, u32::MAX, 0));
        // Unrelated clusters / endpoints are not.
        assert!(!attrs.contains_since(7, 99, 0, 0));
        assert!(!attrs.contains_since(8, 42, 0, 0));
        assert_eq!(id, attrs.watermark());
    }

    #[test]
    fn changed_attrs_record_wildcard_endpoint_covers_every_cluster() {
        let mut attrs = ChangedAttrs::new();
        attrs.record_wildcard(Some(5), None);

        assert!(attrs.contains_since(5, 1, 1, 0));
        assert!(attrs.contains_since(5, 1000, 1000, 0));
        assert!(!attrs.contains_since(6, 1, 1, 0));
    }

    #[test]
    fn changed_attrs_record_wildcard_absorbs_existing_concrete_entries() {
        let mut attrs = ChangedAttrs::new();
        // Seed three concrete attrs on (1, 2).
        attrs.record(1, 2, 10);
        attrs.record(1, 2, 11);
        attrs.record(1, 2, 12);
        // And one concrete on a different cluster - should survive.
        attrs.record(1, 3, 20);
        assert_eq!(attrs.entries.len(), 4);

        // Recording a cluster-wide wildcard for (1, 2) must collapse the three
        // concrete (1, 2, *) entries into the single wildcard.
        attrs.record_wildcard(Some(1), Some(2));

        assert_eq!(attrs.entries.len(), 2);
        assert!(attrs
            .entries
            .iter()
            .any(|e| e.endpoint == 1 && e.cluster == 2 && e.is_attr_wildcard()));
        assert!(attrs.contains_since(1, 3, 20, 0));
    }

    #[test]
    fn changed_attrs_record_wildcard_is_refreshed_when_already_covered() {
        let mut attrs = ChangedAttrs::new();
        // Endpoint-wide wildcard covers any cluster on that endpoint.
        attrs.record_wildcard(Some(1), None);
        let before_len = attrs.entries.len();

        // A cluster-wide wildcard for the same endpoint is already covered
        // by the endpoint-wide one - it must not grow the table and must
        // refresh the existing entry's change id.
        let id = attrs.record_wildcard(Some(1), Some(2));
        assert_eq!(attrs.entries.len(), before_len);
        assert_eq!(attrs.watermark(), id);
    }

    #[test]
    fn changed_attrs_purge_up_to_removes_old_entries() {
        let mut attrs = ChangedAttrs::new();
        attrs.record(1, 2, 3); // id 1
        attrs.record(1, 2, 4); // id 2
        attrs.record(2, 2, 3); // id 3

        attrs.purge_up_to(2);

        assert!(!attrs.contains_since(1, 2, 3, 0));
        assert!(!attrs.contains_since(1, 2, 4, 0));
        assert!(attrs.contains_since(2, 2, 3, 0));

        // Purging with 0 is a no-op.
        attrs.purge_up_to(0);
        assert!(attrs.contains_since(2, 2, 3, 0));
    }

    #[test]
    fn changed_attrs_clear_empties_table_but_keeps_watermark() {
        let mut attrs = ChangedAttrs::new();
        attrs.record(1, 2, 3);
        attrs.record(1, 2, 4);
        let wm_before = attrs.watermark();
        attrs.clear();
        assert!(!attrs.any_since(0));
        // Watermark is preserved so subsequent records remain strictly monotonic.
        assert_eq!(attrs.watermark(), wm_before);
        let id = attrs.record(5, 5, 5);
        assert_eq!(id, wm_before + 1);
    }

    #[test]
    fn changed_attrs_promotion_on_overflow_same_cluster() {
        let mut attrs = ChangedAttrs::new();
        // Fill the table with distinct concrete entries on the same (endpoint, cluster).
        for attr in 0..MAX_CHANGED_ATTRS as u32 {
            attrs.record(1, 2, attr);
        }
        assert_eq!(attrs.entries.len(), MAX_CHANGED_ATTRS);

        // One more record must still succeed - the existing entries get promoted.
        let overflow_id = attrs.record(1, 2, 9999);
        assert_eq!(overflow_id as usize, MAX_CHANGED_ATTRS + 1);

        // The table must never overflow its capacity.
        assert!(attrs.entries.len() <= MAX_CHANGED_ATTRS);

        // Every originally-recorded concrete attribute must still be reported as
        // "changed" when queried from since=0 (possibly via a coarser wildcard).
        for attr in 0..MAX_CHANGED_ATTRS as u32 {
            assert!(
                attrs.contains_since(1, 2, attr, 0),
                "attr {} lost after promotion",
                attr
            );
        }
        assert!(attrs.contains_since(1, 2, 9999, 0));

        // The new overflow entry is visible from the previous watermark.
        assert!(attrs.contains_since(1, 2, 9999, MAX_CHANGED_ATTRS as u64));
    }

    #[test]
    fn changed_attrs_promotion_to_global_wildcard() {
        let mut attrs = ChangedAttrs::new();
        // Entries spread across many endpoints/clusters/attrs to force promotion
        // past the (endpoint, cluster, *) and (endpoint, *, *) levels.
        for i in 0..(MAX_CHANGED_ATTRS as u16 + 5) {
            attrs.record(i, i as u32, i as u32);
        }
        assert!(attrs.entries.len() <= MAX_CHANGED_ATTRS);
        // All previously-recorded triples must still report as changed.
        for i in 0..(MAX_CHANGED_ATTRS as u16 + 5) {
            assert!(attrs.contains_since(i, i as u32, i as u32, 0));
        }
        // And an arbitrary never-recorded triple may or may not be covered
        // (over-reporting is allowed), but `any_since(0)` must be true.
        assert!(attrs.any_since(0));
    }

    #[test]
    fn promotion_prefers_largest_level_1_group() {
        // 10 entries on (1, 1, *) and 5 singletons on (1, k, 0) for k=2..=6
        // (= 15 entries total). One extra record fills the table, then an
        // overflowing record forces exactly ONE level-1 promotion which must
        // collapse the big (1, 1, *) group while leaving singletons concrete.
        let mut attrs = ChangedAttrs::new();
        for attr in 0..10u32 {
            attrs.record(1, 1, attr);
        }
        for cluster in 2..=6u32 {
            attrs.record(1, cluster, 0);
        }
        // Fill exactly to capacity without overflow.
        attrs.record(1, 1, 100);
        assert_eq!(attrs.entries.len(), MAX_CHANGED_ATTRS);

        // Now overflow to trigger promotion.
        attrs.record(2, 2, 2);
        assert!(attrs.entries.len() <= MAX_CHANGED_ATTRS);

        // The big (1, 1, *) group became exactly one wildcard entry.
        let wild_11 = attrs
            .entries
            .iter()
            .filter(|e| e.endpoint == 1 && e.cluster == 1 && e.is_attr_wildcard())
            .count();
        assert_eq!(wild_11, 1);
        // No concrete (1, 1, _) entries survived.
        let concrete_11 = attrs
            .entries
            .iter()
            .filter(|e| e.endpoint == 1 && e.cluster == 1 && !e.is_attr_wildcard())
            .count();
        assert_eq!(concrete_11, 0);
        // Singletons on (1, k, 0) for k=2..=6 remain concrete.
        for cluster in 2..=6u32 {
            let n = attrs
                .entries
                .iter()
                .filter(|e| e.endpoint == 1 && e.cluster == cluster && e.attr == 0)
                .count();
            assert_eq!(n, 1, "singleton (1, {}, 0) should remain concrete", cluster);
        }
        // The new (2, 2, 2) entry is present as a concrete entry.
        assert!(attrs
            .entries
            .iter()
            .any(|e| e.endpoint == 2 && e.cluster == 2 && e.attr == 2));

        // All original triples still report as changed.
        for attr in 0..10u32 {
            assert!(attrs.contains_since(1, 1, attr, 0));
        }
        for cluster in 2..=6u32 {
            assert!(attrs.contains_since(1, cluster, 0, 0));
        }
        assert!(attrs.contains_since(1, 1, 100, 0));
        assert!(attrs.contains_since(2, 2, 2, 0));
    }

    #[test]
    fn promotion_is_minimal_only_one_group_collapsed_per_overflow() {
        // Two big level-1 groups of equal size. A single overflow must collapse
        // only ONE of them, not both (minimal promotion).
        let mut attrs = ChangedAttrs::new();
        // Group A: (1, 1, 0..8) = 8 entries
        for attr in 0..8u32 {
            attrs.record(1, 1, attr);
        }
        // Group B: (2, 2, 0..8) = 8 entries
        for attr in 0..8u32 {
            attrs.record(2, 2, attr);
        }
        assert_eq!(attrs.entries.len(), MAX_CHANGED_ATTRS);

        // Overflow with an unrelated entry.
        attrs.record(9, 9, 9);

        // Exactly one of the groups got collapsed into a wildcard.
        let a_wild = attrs
            .entries
            .iter()
            .any(|e| e.endpoint == 1 && e.cluster == 1 && e.is_attr_wildcard());
        let b_wild = attrs
            .entries
            .iter()
            .any(|e| e.endpoint == 2 && e.cluster == 2 && e.is_attr_wildcard());
        assert!(
            a_wild ^ b_wild,
            "expected exactly one of the groups to be collapsed (A: {}, B: {})",
            a_wild,
            b_wild
        );
        // The un-collapsed group still has all 8 concrete entries.
        let a_concrete = attrs
            .entries
            .iter()
            .filter(|e| e.endpoint == 1 && e.cluster == 1 && !e.is_attr_wildcard())
            .count();
        let b_concrete = attrs
            .entries
            .iter()
            .filter(|e| e.endpoint == 2 && e.cluster == 2 && !e.is_attr_wildcard())
            .count();
        assert!(
            (a_wild && a_concrete == 0 && b_concrete == 8)
                || (b_wild && b_concrete == 0 && a_concrete == 8)
        );
    }

    #[test]
    fn promotion_falls_back_to_level_2_when_no_level_1_group() {
        // All (endpoint, cluster) pairs are unique (level-1 groups are all
        // singletons) but endpoints repeat, so level-2 groups are non-trivial.
        let mut attrs = ChangedAttrs::new();
        for cluster in 0..8u32 {
            attrs.record(1, cluster, 0);
        }
        for cluster in 0..8u32 {
            attrs.record(2, cluster, 0);
        }
        assert_eq!(attrs.entries.len(), MAX_CHANGED_ATTRS);

        attrs.record(3, 9, 9);
        assert!(attrs.entries.len() <= MAX_CHANGED_ATTRS);

        // No level-1 wildcard (endpoint, cluster, *) was produced.
        let lvl1_wild = attrs
            .entries
            .iter()
            .filter(|e| {
                !e.is_endpoint_wildcard() && !e.is_cluster_wildcard() && e.is_attr_wildcard()
            })
            .count();
        assert_eq!(lvl1_wild, 0);
        // Exactly one level-2 wildcard on endpoint 1 or 2 was produced.
        let ep1_wild = attrs
            .entries
            .iter()
            .any(|e| e.endpoint == 1 && e.is_cluster_wildcard() && e.is_attr_wildcard());
        let ep2_wild = attrs
            .entries
            .iter()
            .any(|e| e.endpoint == 2 && e.is_cluster_wildcard() && e.is_attr_wildcard());
        assert!(ep1_wild ^ ep2_wild);
        // No global wildcard was produced either.
        assert!(!attrs
            .entries
            .iter()
            .any(|e| e.is_endpoint_wildcard() && e.is_cluster_wildcard() && e.is_attr_wildcard()));

        // All originals still visible.
        for cluster in 0..8u32 {
            assert!(attrs.contains_since(1, cluster, 0, 0));
            assert!(attrs.contains_since(2, cluster, 0, 0));
        }
        assert!(attrs.contains_since(3, 9, 9, 0));
    }

    #[test]
    fn promotion_falls_back_to_global_only_when_no_lower_group() {
        // All-distinct endpoints AND (endpoint, cluster) pairs: no level-1 or
        // level-2 group has >=2 entries. Overflow must collapse everything to
        // a single global wildcard.
        let mut attrs = ChangedAttrs::new();
        for i in 0..MAX_CHANGED_ATTRS as u16 {
            attrs.record(i, i as u32, i as u32);
        }
        assert_eq!(attrs.entries.len(), MAX_CHANGED_ATTRS);

        attrs.record(100, 200, 300);
        assert_eq!(attrs.entries.len(), 1);
        let only = &attrs.entries[0];
        assert!(
            only.is_endpoint_wildcard() && only.is_cluster_wildcard() && only.is_attr_wildcard()
        );

        // Every previously-recorded triple is still covered.
        for i in 0..MAX_CHANGED_ATTRS as u16 {
            assert!(attrs.contains_since(i, i as u32, i as u32, 0));
        }
        assert!(attrs.contains_since(100, 200, 300, 0));
    }

    #[test]
    fn promotion_preserves_max_change_id_in_coarsened_entry() {
        // After collapsing a (1, 1, *) group, the resulting wildcard's
        // change_id must equal the max change_id of the collapsed entries.
        let mut attrs = ChangedAttrs::new();
        for attr in 0..MAX_CHANGED_ATTRS as u32 {
            attrs.record(1, 1, attr);
        }
        let max_before = attrs.watermark();

        attrs.record(2, 2, 2);
        let wild = attrs
            .entries
            .iter()
            .find(|e| e.endpoint == 1 && e.cluster == 1 && e.is_attr_wildcard())
            .expect("(1, 1, *) wildcard was produced");
        assert_eq!(wild.change_id, max_before);

        // contains_since respects that watermark exactly.
        assert!(attrs.contains_since(1, 1, 0, max_before - 1));
        assert!(!attrs.contains_since(1, 1, 0, max_before));
    }

    #[test]
    fn promotion_with_existing_wildcard_refreshes_instead_of_promoting_again() {
        // Build a state where (1, 1, *) wildcard already exists via a forced
        // promotion. Recording another (1, 1, k) must refresh that wildcard's
        // change_id without producing any new entry.
        let mut attrs = ChangedAttrs::new();
        for attr in 0..MAX_CHANGED_ATTRS as u32 {
            attrs.record(1, 1, attr);
        }
        attrs.record(2, 2, 2); // forces (1, 1, *) promotion

        // Now the table has 2 entries: (1, 1, *) and (2, 2, 2).
        assert_eq!(attrs.entries.len(), 2);
        let wm_after_promo = attrs.watermark();

        let new_id = attrs.record(1, 1, 42);
        // No new entry: still 2 entries. Wildcard's change_id advanced.
        assert_eq!(attrs.entries.len(), 2);
        assert_eq!(new_id, wm_after_promo + 1);
        let wild = attrs
            .entries
            .iter()
            .find(|e| e.endpoint == 1 && e.cluster == 1 && e.is_attr_wildcard())
            .unwrap();
        assert_eq!(wild.change_id, new_id);
    }

    #[test]
    fn promotion_capacity_invariant_under_sustained_churn() {
        // Sustained mixed churn must never let the table exceed its capacity,
        // and every freshly-recorded triple must remain visible immediately
        // after recording.
        let mut attrs = ChangedAttrs::new();
        for i in 0..1000u32 {
            let endpoint = (i % 7) as u16;
            let cluster = i % 13;
            let attr = i;
            attrs.record(endpoint, cluster, attr);
            assert!(
                attrs.entries.len() <= MAX_CHANGED_ATTRS,
                "capacity exceeded at i={}",
                i
            );
            assert!(
                attrs.contains_since(endpoint, cluster, attr, 0),
                "just-recorded triple lost at i={}",
                i
            );
        }
    }

    #[test]
    fn promotion_iterated_into_same_existing_wildcard() {
        // Once (1, 1, *) exists, repeated inserts on that group must never
        // grow the table, and never trigger further promotion.
        let mut attrs = ChangedAttrs::new();
        for attr in 0..MAX_CHANGED_ATTRS as u32 {
            attrs.record(1, 1, attr);
        }
        attrs.record(2, 2, 2); // -> [(1,1,*), (2,2,2)]
        assert_eq!(attrs.entries.len(), 2);

        for attr in 100..200u32 {
            attrs.record(1, 1, attr);
            assert_eq!(attrs.entries.len(), 2);
        }
    }

    #[test]
    fn promotion_escalates_when_level_1_group_still_insufficient() {
        // Pathological case: a single level-1 group of size 2 exists, the rest
        // are singletons. After the first overflow, that group collapses
        // (freeing 1 slot), but the table is still full once the new record
        // tries to be inserted on a fresh singleton location. Subsequent
        // overflows must escalate to level-2 / global.
        let mut attrs = ChangedAttrs::new();
        // 2 entries sharing (1, 1, *) -- a single level-1 group of size 2.
        attrs.record(1, 1, 0);
        attrs.record(1, 1, 1);
        // Fill the rest with unique (endpoint, cluster) pairs.
        for i in 0..(MAX_CHANGED_ATTRS as u16 - 2) {
            attrs.record(10 + i, 100 + i as u32, i as u32);
        }
        assert_eq!(attrs.entries.len(), MAX_CHANGED_ATTRS);

        // First overflow: the only level-1 group collapses; then the new entry
        // gets inserted.
        attrs.record(50, 50, 50);
        assert!(attrs.entries.len() <= MAX_CHANGED_ATTRS);
        // The (1, 1, *) wildcard is present.
        assert!(attrs
            .entries
            .iter()
            .any(|e| e.endpoint == 1 && e.cluster == 1 && e.is_attr_wildcard()));

        // Keep feeding: eventually we must fall back to level-2 or global
        // without breaking correctness.
        for i in 0..200u32 {
            let endpoint = 200 + (i % 5) as u16;
            let cluster = 300 + (i % 3);
            let attr = i;
            attrs.record(endpoint, cluster, attr);
            assert!(attrs.entries.len() <= MAX_CHANGED_ATTRS);
            assert!(attrs.contains_since(endpoint, cluster, attr, 0));
        }
        // Historical triples still covered.
        assert!(attrs.contains_since(1, 1, 0, 0));
        assert!(attrs.contains_since(1, 1, 1, 0));
        assert!(attrs.contains_since(50, 50, 50, 0));
    }

    #[test]
    fn changed_attr_covers_wildcards() {
        let concrete = ChangedAttr::concrete(1, 2, 3, 1);
        let any_attr = ChangedAttr {
            endpoint: 1,
            cluster: 2,
            attr: WILDCARD_ATTR,
            change_id: 1,
        };
        let any_cluster = ChangedAttr {
            endpoint: 1,
            cluster: WILDCARD_CLUSTER,
            attr: WILDCARD_ATTR,
            change_id: 1,
        };
        let global = ChangedAttr {
            endpoint: WILDCARD_ENDPOINT,
            cluster: WILDCARD_CLUSTER,
            attr: WILDCARD_ATTR,
            change_id: 1,
        };

        assert!(any_attr.covers(&concrete));
        assert!(any_cluster.covers(&concrete));
        assert!(global.covers(&concrete));
        // Concrete does not cover wildcards.
        assert!(!concrete.covers(&any_attr));
        assert!(!concrete.covers(&global));
        // Concrete matches itself.
        assert!(concrete.matches(1, 2, 3));
        assert!(!concrete.matches(1, 2, 4));
        // Wildcards match any concrete triple on the wildcarded axis.
        assert!(any_attr.matches(1, 2, 99));
        assert!(!any_attr.matches(1, 9, 99));
        assert!(global.matches(99, 99, 99));
    }

    #[test]
    fn changed_attr_size_is_compact() {
        // `ChangedAttr` must stay at 24 bytes on 64-bit targets: `u64` change_id
        // forces 8-byte alignment, and the `(u32, u32, u16)` path tuple fits in
        // the remaining 16 bytes (4 + 4 + 2 + 6 padding). Regressing back to an
        // `Option<u16> / Option<u32> / Option<u32>` encoding would bump this to
        // 32 bytes per entry, i.e. +128 bytes per `Subscriptions` table.
        assert_eq!(core::mem::size_of::<ChangedAttr>(), 24);
    }

    // ---------- Subscriptions ----------

    fn fab(i: u8) -> NonZeroU8 {
        NonZeroU8::new(i).unwrap()
    }

    #[test]
    fn add_returns_monotonic_ids_and_rejects_when_full() {
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();

        let rctx1 = subs
            .add(
                now,
                fab(1),
                10,
                1,
                60,
                0,
                pool.get_immediate().unwrap(),
                &subs_bufs,
            )
            .unwrap();
        let rctx2 = subs
            .add(
                now,
                fab(1),
                10,
                1,
                60,
                0,
                pool.get_immediate().unwrap(),
                &subs_bufs,
            )
            .unwrap();
        assert_eq!(rctx1.subscription().ids().id, 1);
        assert_eq!(rctx2.subscription().ids().id, 2);

        // Third add exceeds N=2.
        assert!(subs
            .add(
                now,
                fab(1),
                10,
                1,
                60,
                0,
                pool.get_immediate().unwrap(),
                &subs_bufs
            )
            .is_none());
    }

    #[test]
    fn begin_report_snapshots_watermark_and_pending() {
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();

        subs.notify_attr_changed(1, 2, 3);
        {
            let mut rctx = subs
                .add(
                    now,
                    fab(1),
                    10,
                    1,
                    60,
                    0,
                    pool.get_immediate().unwrap(),
                    &subs_bufs,
                )
                .unwrap();

            // The priming report is un-filtered: every attribute is reported and
            // `should_send_if_empty` is true so that the snapshot is delivered
            // unconditionally.
            assert!(rctx.should_send_if_empty());
            assert!(rctx.should_report_attr(1, 2, 3));
            assert!(rctx.should_report_attr(42, 55555, 1234556677));

            rctx.set_keep();
        }

        // A new change bumps the watermark and becomes pending.
        subs.notify_attr_changed(1, 2, 4);
        // `min_int` = 1s has not elapsed at `now`, so the subscription is not
        // yet report-allowed; step past it.
        let later = now + Duration::from_secs(2);
        let rctx = subs.report(later, 0, &subs_bufs).unwrap();
        assert!(!rctx.should_send_if_empty());
        // The priming commit advanced the sub's `since` past the (1, 2, 3)
        // change, so only the new (1, 2, 4) is pending.
        assert!(!rctx.should_report_attr(1, 2, 3));
        assert!(rctx.should_report_attr(1, 2, 4));
    }

    /// A failed report (`set_failed`) keeps the subscription but must NOT advance
    /// its watermark, so the changes it was carrying stay pending and are
    /// re-attempted on the next report. Contrast with `set_keep`, which commits
    /// the advanced watermark (correct only after a *delivered* report).
    #[test]
    fn failed_report_preserves_pending_changes() {
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();

        // Prime the subscription (delivered) so it is a normal, non-priming entry.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }

        // A change becomes pending.
        subs.notify_attr_changed(1, 2, 3);

        let later = now + Duration::from_secs(2);

        // First report attempt FAILS: the report was carrying (1, 2, 3) but never
        // reached the subscriber.
        {
            let mut rctx = subs.report(later, 0, &subs_bufs).unwrap();
            assert!(rctx.should_report_attr(1, 2, 3), "change is pending");
            rctx.set_keep_retry();
        }

        // Because the failed report did not advance the watermark, (1, 2, 3) is
        // STILL pending on the next report — no silent data loss.
        let later2 = later + Duration::from_secs(2);
        {
            let rctx = subs.report(later2, 0, &subs_bufs).unwrap();
            assert!(
                rctx.should_report_attr(1, 2, 3),
                "a failed report must not consume the pending change"
            );
        }
    }

    /// The success counterpart: `set_keep` after a delivered report DOES advance
    /// the watermark, so the same change is not reported again.
    #[test]
    fn delivered_report_consumes_pending_changes() {
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();

        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }

        subs.notify_attr_changed(1, 2, 3);

        let later = now + Duration::from_secs(2);
        {
            let mut rctx = subs.report(later, 0, &subs_bufs).unwrap();
            assert!(rctx.should_report_attr(1, 2, 3));
            rctx.set_keep(); // delivered
        }

        // The change was consumed by the delivered report and the liveness
        // deadline is not yet reached, so there is nothing left to report.
        let later2 = later + Duration::from_secs(2);
        assert!(
            subs.report(later2, 0, &subs_bufs).is_none(),
            "a delivered report consumes the pending change"
        );
    }

    /// A failed report must not busy-loop: even with a change pending and
    /// `min_int` elapsed, the subscription is held back until its back-off
    /// (`retry_at`) elapses, and the reporter's wake deadline reflects that.
    #[test]
    fn failed_report_backs_off_instead_of_busy_looping() {
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();

        // Prime (delivered), then make a change pending, all with min_int = 1s.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 600);
            rctx.set_keep();
        }
        subs.notify_attr_changed(1, 2, 3);

        // min_int has elapsed and the change is pending — the report is due, but
        // it FAILS.
        let t1 = now + Duration::from_secs(2);
        {
            let mut rctx = subs.report(t1, 0, &subs_bufs).unwrap();
            rctx.set_keep_retry();
        }

        // Immediately after the failure the subscription is NOT reportable again,
        // even though the change is still pending and min_int is long past: the
        // back-off gates it. This is the anti-busy-loop guarantee.
        assert!(
            subs.report(t1, 0, &subs_bufs).is_none(),
            "a failed report must not be retried in the same instant"
        );
        // Still gated a moment later (BASE back-off is 2s).
        assert!(subs
            .report(t1 + Duration::from_millis(500), 0, &subs_bufs)
            .is_none());

        // The reporter's wake deadline is pushed to the back-off point, so the
        // loop sleeps rather than spinning.
        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            t1 + Duration::from_secs(2),
            "wake is scheduled at the back-off point, not immediately"
        );

        // Once the back-off elapses the pending change is retried.
        let t2 = t1 + Duration::from_secs(2);
        {
            let rctx = subs.report(t2, 0, &subs_bufs).unwrap();
            assert!(
                rctx.should_report_attr(1, 2, 3),
                "pending change is retried"
            );
        }
    }

    /// The back-off grows with consecutive failures (exponential from BASE),
    /// observable through the reporter's wake deadline.
    #[test]
    fn repeated_failures_grow_the_backoff() {
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        // Large max_int so the cap never clips the small back-offs under test.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 600);
            rctx.set_keep();
        }
        subs.notify_attr_changed(1, 2, 3);

        // BASE = 2s, doubling each consecutive failure: 2s, 4s, 8s.
        let mut t = now + Duration::from_secs(2);
        for expected_backoff in [2u64, 4, 8] {
            let mut rctx = subs.report(t, 0, &subs_bufs).unwrap();
            rctx.set_keep_retry();
            drop(rctx);

            assert_eq!(
                subs.next_report_at(0, &subs_bufs),
                t + Duration::from_secs(expected_backoff),
                "back-off after this failure",
            );

            // Advance exactly to the back-off point for the next failed attempt.
            t += Duration::from_secs(expected_backoff);
        }
    }

    /// The back-off never exceeds `max_int`: it makes no sense to wait past the
    /// point at which the subscription expires anyway.
    #[test]
    fn backoff_is_capped_at_max_int() {
        // With max_int = 5s, even a high fail count is clamped to 5s.
        assert_eq!(Subscription::retry_backoff_secs(1, 5), 2);
        assert_eq!(Subscription::retry_backoff_secs(3, 5), 5); // 8 -> capped to 5
        assert_eq!(Subscription::retry_backoff_secs(20, 5), 5); // huge -> capped
                                                                // With a generous max_int the exponential shows through, then saturates.
        assert_eq!(Subscription::retry_backoff_secs(1, 600), 2);
        assert_eq!(Subscription::retry_backoff_secs(4, 600), 16);
        assert_eq!(Subscription::retry_backoff_secs(100, 600), 600);
    }

    /// `is_expired` measures `max_int` from the last *successful* report, not
    /// from the last attempt: a run of failures does NOT keep the subscription
    /// alive forever — it expires `max_int` after the last success.
    #[test]
    fn failures_do_not_postpone_expiry() {
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let base = Instant::now();
        // max_int = 10s.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, base, 1, 10, 1, 10);
            rctx.set_keep(); // last SUCCESS is at `base`
        }
        subs.notify_attr_changed(1, 2, 3);

        // A failed report at base+2s must NOT move the expiry deadline.
        {
            let mut rctx = subs
                .report(base + Duration::from_secs(2), 0, &subs_bufs)
                .unwrap();
            rctx.set_keep_retry();
        }

        // Just short of max_int-since-success: still alive.
        let before = base + Duration::from_secs(9);
        assert!(!subs.remove(&subs_bufs, |sub| sub
            .is_expired(before)
            .then_some("expired")));

        // At max_int-since-success (measured from `base`, not from the failed
        // attempt at base+2s): expired.
        let after = base + Duration::from_secs(10);
        assert!(
            subs.remove(&subs_bufs, |sub| sub.is_expired(after).then_some("expired")),
            "expiry is measured from the last success, not the last attempt"
        );
    }

    /// A delivered report after a run of failures clears the back-off: the next
    /// wake returns to the normal liveness schedule rather than a back-off floor.
    #[test]
    fn success_clears_the_backoff() {
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }
        subs.notify_attr_changed(1, 2, 3);

        // Two failures build up a back-off.
        let t1 = now + Duration::from_secs(2);
        {
            let mut rctx = subs.report(t1, 0, &subs_bufs).unwrap();
            rctx.set_keep_retry();
        }
        let t2 = t1 + Duration::from_secs(2);
        {
            let mut rctx = subs.report(t2, 0, &subs_bufs).unwrap();
            rctx.set_keep_retry();
        }

        // Now a delivered report at t3.
        let t3 = t2 + Duration::from_secs(4);
        {
            let mut rctx = subs.report(t3, 0, &subs_bufs).unwrap();
            assert!(rctx.should_report_attr(1, 2, 3));
            rctx.set_keep(); // delivered — resets fail_count and retry_at
        }

        // The back-off is gone: the next wake is the plain liveness point
        // (reported_at + max_int - max_int/2 = t3 + 30s), with no retry floor.
        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            t3 + Duration::from_secs(30),
            "a delivered report clears the back-off"
        );
    }

    // The following tests cover the public API of `Subscriptions` /
    // `SubscriptionsBuffers` / `ReportContext` post-refactor. A few of the
    // pre-refactor tests had no meaningful successor and were deleted:
    //
    //   * `sub_attr_change_filter_honors_since_watermark` — `SubAttrChangeFilter`
    //     is now dead code (see REVIEW above); the `since`-watermark logic is
    //     already covered by `changed_attrs_contains_since_and_any_since`.
    //   * `find_report_due_events_pending_receives_subscription_watermark` —
    //     the old `events_pending` callback no longer exists; the
    //     subscription's `max_seen_event_number` is now compared directly
    //     against the `event_numbers_watermark` passed to
    //     `Subscriptions::report`.
    //   * `find_removed_session_matches_predicate` — `session_id` tracking
    //     was dropped in the refactor (see REVIEW on `SubscriptionsInner::add`).
    //     Predicate-based removal is covered by `remove_invokes_predicate_*`.

    /// Helper: add a subscription with sensible defaults and return its `ReportContext`.
    #[allow(clippy::too_many_arguments)]
    fn add_sub<'a, 's, const N: usize, const B: usize>(
        subs: &'s Subscriptions<N>,
        subs_bufs: &'s SubscriptionsBuffers<'a, TestPool<B>, N>,
        pool: &'a TestPool<B>,
        now: Instant,
        fab_idx: u8,
        peer_node_id: u64,
        min_int: u16,
        max_int: u16,
    ) -> ReportContext<'a, 's, TestPool<B>, N>
    where
        'a: 's,
    {
        subs.add(
            now,
            fab(fab_idx),
            peer_node_id,
            min_int,
            max_int,
            /* event_numbers_watermark */ 0,
            pool.get_immediate().unwrap(),
            subs_bufs,
        )
        .unwrap()
    }

    #[test]
    fn priming_report_context_is_report_due_and_keeps_sub() {
        // A subscription returned from `add` is the "priming" report: it must be
        // report-due regardless of time (so the initial report is delivered
        // unconditionally) and, when dropped with `set_keep`, must survive in
        // the subscription table for subsequent incremental reports.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            assert!(rctx.should_send_if_empty());
            assert_eq!(rctx.max_seen_event_number(), 0);
            rctx.set_keep();
        }

        // After priming, a zero-delta report at the same instant finds nothing
        // pending (no attr changes, no new events, min_int not elapsed).
        assert!(subs.report(now, 0, &subs_bufs).is_none());
    }

    #[test]
    fn report_without_keep_frees_the_slot() {
        // Dropping a `ReportContext` *without* `set_keep` must remove the
        // subscription from the table (and free its buffer), so a new
        // subscription can take its place up to the `N` capacity.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();

        // Add then drop without keep.
        drop(add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60));

        // The slot is free again: a second add succeeds even with N=1.
        let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 11, 1, 60);
        // IDs are still strictly monotonic across add/remove cycles.
        assert_eq!(rctx.subscription().ids().id, 2);
        rctx.set_keep();
    }

    #[test]
    fn report_with_keep_advances_reported_at_and_watermark() {
        // After a "kept" report, the subscription must not be picked up again
        // at the same instant unless new changes arrive.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();

        // Prime and keep.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }

        // Record one attribute change — watermark advances to 1.
        subs.notify_attr_changed(1, 2, 3);

        // At the same instant, min_int (1s) has NOT elapsed so the sub is not
        // report-allowed: even though there is a pending change, `report()`
        // returns None.
        assert!(subs.report(now, 0, &subs_bufs).is_none());

        // Past min_int: the pending change makes the sub reportable.
        let later = now + Duration::from_secs(2);
        {
            let mut rctx = subs.report(later, 0, &subs_bufs).unwrap();
            assert!(rctx.should_report_attr(1, 2, 3));
            // A fresh (never recorded) triple is NOT in the table and must
            // not be spuriously reported.
            assert!(!rctx.should_report_attr(9, 9, 9));
            rctx.set_keep();
        }

        // Watermark has been committed — another call at `later` with no new
        // activity finds nothing.
        assert!(subs.report(later, 0, &subs_bufs).is_none());
    }

    #[test]
    fn report_triggered_by_new_events() {
        // A bump in `event_numbers_watermark` (i.e. a newly emitted event)
        // makes the subscription reportable even without attribute changes.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }

        // Same instant, no new events (watermark = 0 same as sub's
        // max_seen), min_int not elapsed → nothing to report.
        assert!(subs.report(now, 0, &subs_bufs).is_none());

        let later = now + Duration::from_secs(2);

        // Still no new events at `later` (min_int elapsed though).
        assert!(subs.report(later, 0, &subs_bufs).is_none());

        // A new event bumps the watermark → sub is reportable. The captured
        // `next_max_seen_event_number` mirrors the watermark and is the
        // value that will be committed on `set_keep`.
        {
            let mut rctx = subs.report(later, 5, &subs_bufs).unwrap();
            assert_eq!(rctx.max_seen_event_number(), 0);
            assert_eq!(rctx.next_max_seen_event_number(), 5);
            rctx.set_keep();
        }

        // After reporting, watermark=5 is no longer "new" for this sub.
        assert!(subs.report(later, 5, &subs_bufs).is_none());
        // But a further bump does trigger again (past min_int is needed).
        let even_later = later + Duration::from_secs(2);
        {
            let mut rctx = subs.report(even_later, 6, &subs_bufs).unwrap();
            assert_eq!(rctx.max_seen_event_number(), 5);
            assert_eq!(rctx.next_max_seen_event_number(), 6);
            rctx.set_keep();
        }
    }

    #[test]
    fn report_triggered_by_liveness_deadline() {
        // With no changes at all, a subscription still becomes reportable once
        // it enters the "liveness" window (within half of `max_int` of the
        // deadline).
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        // max_int = 20s → half of max_int = 10s → becomes report-due at now+10s.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 20);
            rctx.set_keep();
        }

        // Short of the liveness window: not due.
        let short = now + Duration::from_secs(5);
        assert!(subs.report(short, 0, &subs_bufs).is_none());

        // At the liveness window: due even without any attr/event change.
        let long = now + Duration::from_secs(11);
        {
            let mut rctx = subs.report(long, 0, &subs_bufs).unwrap();
            assert!(rctx.should_send_if_empty());
            rctx.set_keep();
        }
    }

    #[test]
    fn next_report_at_max_when_empty() {
        // No subscriptions → no deadline (`Instant::MAX`); the timer never fires
        // and the reporter waits to be notified.
        let subs: Subscriptions<1> = Subscriptions::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();
        assert_eq!(subs.next_report_at(0, &subs_bufs), Instant::MAX);
    }

    #[test]
    fn next_report_at_liveness_when_idle() {
        // No pending data → wake at the liveness point
        // `reported_at + max_int - max_int/2` (when `is_report_due` flips).
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        // min_int = 1s, max_int = 60s → 60 - 30 = 30s.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }

        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            now + Duration::from_secs(30)
        );
    }

    #[test]
    fn next_report_at_quiet_period_for_pending_attr_change() {
        // A change recorded inside the quiet period must schedule the wake at
        // the end of that period (`reported_at + min_int`), not the far-off
        // liveness point.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        // min_int = 5s, max_int = 60s (liveness would otherwise be at now+30s).
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 5, 60);
            rctx.set_keep();
        }

        subs.notify_attr_changed(1, 2, 3);

        // Held back by the quiet period, so still not reportable now...
        assert!(subs.report(now, 0, &subs_bufs).is_none());
        // ...but the wake is scheduled at min_int, not liveness.
        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            now + Duration::from_secs(5)
        );
    }

    #[test]
    fn next_report_at_quiet_period_for_pending_event() {
        // Same as above, driven by a new event (watermark past the sub's
        // max_seen = 0) rather than an attribute change.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 5, 60);
            rctx.set_keep();
        }

        assert_eq!(
            subs.next_report_at(7, &subs_bufs),
            now + Duration::from_secs(5)
        );
    }

    #[test]
    fn next_report_at_clamps_liveness_to_min_interval() {
        // Regression guard for the busy-spin: when `min_int > max_int/2`, the
        // liveness point (`reported_at + max_int - max_int/2`) precedes the
        // min-interval gate at which a report is first allowed. The wake MUST
        // be clamped to the gate; otherwise the reporter wakes early, finds the
        // sub still gated, re-arms the same past deadline, and spins.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();
        // min_int = 25s, max_int = 40s → liveness at now+20s, gate at now+25s.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 25, 40);
            rctx.set_keep();
        }

        // Clamped to the gate (now+25), not the earlier liveness point (now+20).
        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            now + Duration::from_secs(25)
        );
        // The scheduled instant matches actual reportability: gated before it,
        // reportable at it.
        assert!(subs
            .report(now + Duration::from_secs(24), 0, &subs_bufs)
            .is_none());
        assert!(subs
            .report(now + Duration::from_secs(25), 0, &subs_bufs)
            .is_some());
    }

    #[test]
    fn next_report_at_returns_earliest_across_subs() {
        // The reporter must wake for whichever subscription is due first.
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();
        // Sub A: max_int = 60s → liveness now+30s.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }
        // Sub B: max_int = 40s → liveness now+20s (the earliest).
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 2, 11, 1, 40);
            rctx.set_keep();
        }

        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            now + Duration::from_secs(20)
        );
    }

    #[test]
    fn subscription_added_notification_wakes_reporter_to_recompute_deadline() {
        use core::pin::pin;
        use embassy_futures::select::{select, Either};

        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            rctx.set_keep();
        }

        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            now + Duration::from_secs(30)
        );

        let waiter = pin!(subs.notification.wait());

        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 11, 1, 10);
            rctx.set_keep();
        }
        subs.notification.notify();

        let notified = embassy_futures::block_on(async {
            match select(waiter, pin!(core::future::ready(()))).await {
                Either::First(_) => true,
                Either::Second(_) => false,
            }
        });

        assert!(notified);
        assert_eq!(
            subs.next_report_at(0, &subs_bufs),
            now + Duration::from_secs(5)
        );
    }

    #[test]
    fn is_expired_uses_max_int() {
        // `Subscription::is_expired` returns true once `max_int` has elapsed
        // since the last reported_at.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let base = Instant::now();
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, base, 1, 10, 1, 5);
            rctx.set_keep();
        }

        // Before max_int: not expired. Use the `remove` predicate as a probe
        // because we have no other way to observe per-sub `is_expired` through
        // the public API.
        let before = base + Duration::from_secs(2);
        assert!(!subs.remove(&subs_bufs, |sub| sub
            .is_expired(before)
            .then_some("expired")));

        // Past max_int: expired — removal fires.
        let after = base + Duration::from_secs(10);
        assert!(subs.remove(&subs_bufs, |sub| sub.is_expired(after).then_some("expired")));
    }

    #[test]
    fn remove_invokes_predicate_and_frees_slots() {
        // `Subscriptions::remove` drains every matching entry (not just one),
        // returns whether anything was removed, and frees the slots so that
        // subsequent `add` calls succeed up to the capacity `N`.
        let subs: Subscriptions<3> = Subscriptions::new();
        let pool = TestPool::<4>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<4>, 3> = SubscriptionsBuffers::new();

        let now = Instant::now();
        for peer in [100_u64, 101, 102] {
            let fab_idx = if peer == 102 { 2 } else { 1 };
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, fab_idx, peer, 1, 60);
            rctx.set_keep();
        }
        // Table is full: a 4th add must be rejected.
        assert!(subs
            .add(
                now,
                fab(1),
                200,
                1,
                60,
                0,
                pool.get_immediate().unwrap(),
                &subs_bufs
            )
            .is_none());

        // Remove every fab(1) subscription (2 of them).
        let mut seen_peers: std::vec::Vec<u64> = std::vec::Vec::new();
        let removed = subs.remove(&subs_bufs, |sub| {
            if sub.ids().fab_idx == fab(1) {
                seen_peers.push(sub.ids().peer_node_id);
                Some("fabric 1 removed")
            } else {
                None
            }
        });
        assert!(removed);
        seen_peers.sort();
        assert_eq!(seen_peers, std::vec![100_u64, 101]);

        // A second identical remove is a no-op and returns false.
        assert!(!subs.remove(&subs_bufs, |sub| (sub.ids().fab_idx == fab(1))
            .then_some("fabric 1 removed")));

        // Two slots were freed: we can add two more subs.
        {
            let mut r1 = add_sub(&subs, &subs_bufs, &pool, now, 3, 300, 1, 60);
            r1.set_keep();
            let mut r2 = add_sub(&subs, &subs_bufs, &pool, now, 3, 301, 1, 60);
            r2.set_keep();
        }
        // And a third add is rejected again (back at capacity).
        assert!(subs
            .add(
                now,
                fab(3),
                302,
                1,
                60,
                0,
                pool.get_immediate().unwrap(),
                &subs_bufs
            )
            .is_none());
    }

    #[test]
    fn remove_on_empty_table_returns_false() {
        let subs: Subscriptions<2> = Subscriptions::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 2> = SubscriptionsBuffers::new();
        assert!(!subs.remove(&subs_bufs, |_| Some("never called on empty")));
    }

    #[test]
    fn remove_cancels_in_flight_subscription() {
        // A subscription that has been moved into a `ReportContext` is still
        // observable to `remove` via `SubscriptionsInner::reporting`. Matching
        // it must cause `report_complete` to drop the subscription on Drop
        // rather than re-inserting it, even when `set_keep` was called.
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();

        // Prime a subscription so it lives in the table.
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 100, 1, 60);
            rctx.set_keep();
        }
        assert_eq!(subs.state.lock(|s| s.borrow().subscriptions_count), 1);

        // Start an incremental report and, while it is "in flight", issue
        // a `remove` that matches the in-flight subscription. Also flip
        // `set_keep` to verify the cancel flag wins over `keep`.
        subs.notify_attr_changed(1, 2, 3);
        let later = now + Duration::from_secs(2);
        {
            let mut rctx = subs.report(later, 0, &subs_bufs).unwrap();

            // The in-flight sub is currently absent from `state.subscriptions`
            // but must still be visible to `remove` through the `reporting`
            // slot.
            let mut matched_peers: std::vec::Vec<u64> = std::vec::Vec::new();
            let removed = subs.remove(&subs_bufs, |sub| {
                matched_peers.push(sub.ids().peer_node_id);
                (sub.ids().peer_node_id == 100).then_some("test-cancel")
            });
            assert!(removed);
            assert!(matched_peers.contains(&100));

            // Even though we ask to keep, the cancel flag must force a drop.
            rctx.set_keep();
        }

        // After `ReportContext::drop` the subscription must be gone and the
        // slot freed.
        subs.state.lock(|s| {
            let s = s.borrow();
            assert_eq!(s.subscriptions_count, 0);
            assert!(s.subscriptions.is_empty());
            assert!(s.reporting.is_none());
            assert!(s.reporting_cancelled.is_none());
        });

        // Slot is free: a new sub can be added.
        let mut r = add_sub(&subs, &subs_bufs, &pool, now, 1, 101, 1, 60);
        r.set_keep();
    }

    #[test]
    fn remove_not_matching_in_flight_leaves_it_intact() {
        // If `remove`'s predicate matches neither the in-flight subscription
        // nor anything in the table, the in-flight subscription must still
        // be re-inserted on `ReportContext::drop` when `set_keep` is called.
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let now = Instant::now();
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 100, 1, 60);
            rctx.set_keep();
        }

        subs.notify_attr_changed(1, 2, 3);
        let later = now + Duration::from_secs(2);
        {
            let mut rctx = subs.report(later, 0, &subs_bufs).unwrap();
            let removed = subs.remove(&subs_bufs, |sub| {
                (sub.ids().peer_node_id == 999).then_some("no-match")
            });
            assert!(!removed);
            rctx.set_keep();
        }

        subs.state.lock(|s| {
            let s = s.borrow();
            assert_eq!(s.subscriptions_count, 1);
            assert_eq!(s.subscriptions.len(), 1);
            assert!(s.reporting.is_none());
            assert!(s.reporting_cancelled.is_none());
        });
    }

    #[test]
    fn purge_reported_changes_keeps_entries_until_all_subs_catch_up() {
        // `purge_reported_changes` must only drop table entries every
        // subscription has already reported on: the slowest subscriber's
        // `max_seen_attr_change_id` acts as a floor.
        let subs: Subscriptions<2> = Subscriptions::new();
        let pool = TestPool::<3>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<3>, 2> = SubscriptionsBuffers::new();

        let base = Instant::now();

        // Two priming adds. Both start at watermark = 0 (no changes yet).
        // `ReportContext::next_max_seen_attr_change_id` is captured as 0 by
        // `add`, so dropping either rctx with keep commits max_seen = 0.
        {
            let mut r1 = add_sub(&subs, &subs_bufs, &pool, base, 1, 100, 1, 60);
            r1.set_keep();
            let mut r2 = add_sub(&subs, &subs_bufs, &pool, base, 1, 101, 1, 60);
            r2.set_keep();
        }

        // Record two changes. Watermark becomes 2.
        subs.notify_attr_changed(1, 2, 3); // id 1
        subs.notify_attr_changed(1, 2, 4); // id 2

        // Advance both subs to watermark 2 via two `report` + keep cycles.
        let later = base + Duration::from_secs(2);
        for _ in 0..2 {
            let mut rctx = subs.report(later, 0, &subs_bufs).unwrap();
            assert!(rctx.should_report_attr(1, 2, 3));
            assert!(rctx.should_report_attr(1, 2, 4));
            rctx.set_keep();
        }

        // Both subs have max_seen = 2; purge is safe and removes the stale
        // entries. The next report should now find nothing pending (same
        // instant, no new changes, min_int elapsed but not half of max_int).
        subs.purge_reported_changes();
        assert!(subs.report(later, 0, &subs_bufs).is_none());

        // A brand new change becomes pending again post-purge.
        subs.notify_attr_changed(5, 6, 7);
        let even_later = later + Duration::from_secs(2);
        {
            let mut rctx = subs.report(even_later, 0, &subs_bufs).unwrap();
            assert!(rctx.should_report_attr(5, 6, 7));
            // Previously-purged entries are no longer visible through the
            // sub's filter either.
            assert!(!rctx.should_report_attr(1, 2, 3));
            rctx.set_keep();
        }
    }

    #[test]
    fn next_max_seen_event_number_captured_at_report_time() {
        // The captured `next_max_seen_event_number` reflects the
        // `event_numbers_watermark` passed to `add` / `report` and is what
        // gets committed to the subscription on `set_keep`. The committed
        // value advances even if no events were actually emitted during the
        // report — this is what prevents the "endless reporting loop" for
        // subscriptions that are not interested in events but receive an
        // event-triggered report.
        let subs: Subscriptions<1> = Subscriptions::new();
        let pool = TestPool::<2>::new();
        let subs_bufs: SubscriptionsBuffers<TestPool<2>, 1> = SubscriptionsBuffers::new();

        let now = Instant::now();

        // Priming report sees the watermark passed to `add` (0 here).
        {
            let mut rctx = add_sub(&subs, &subs_bufs, &pool, now, 1, 10, 1, 60);
            assert_eq!(rctx.max_seen_event_number(), 0);
            assert_eq!(rctx.next_max_seen_event_number(), 0);
            rctx.set_keep();
        }

        let later = now + Duration::from_secs(2);

        // First incremental report at watermark=7: the captured "next" is 7,
        // and the previous watermark (the sub's `max_seen_event_number`) is
        // still 0 until commit.
        {
            let mut rctx = subs.report(later, 7, &subs_bufs).unwrap();
            assert_eq!(rctx.max_seen_event_number(), 0);
            assert_eq!(rctx.next_max_seen_event_number(), 7);
            rctx.set_keep();
        }

        // After commit the sub's `max_seen_event_number` has advanced to 7
        // — even though we never recorded a single emitted event during
        // this report. A second call at the same watermark is therefore a
        // no-op (no new events to deliver).
        assert!(subs.report(later, 7, &subs_bufs).is_none());

        let even_later = later + Duration::from_secs(2);

        // Bumping the watermark to 42 makes the sub reportable again; the
        // previous max-seen is the 7 we just committed, the captured next
        // is the new watermark.
        {
            let mut rctx = subs.report(even_later, 42, &subs_bufs).unwrap();
            assert_eq!(rctx.max_seen_event_number(), 7);
            assert_eq!(rctx.next_max_seen_event_number(), 42);
            rctx.set_keep();
        }
    }

    // ---------- persistence ----------

    // Gated as a unit: the persist/resume machinery under test only exists when
    // the `persistent-subscriptions` feature is enabled.
    #[cfg(feature = "persistent-subscriptions")]
    mod persistence {
        use super::*;

        /// A minimal multi-key in-memory store, enough to test the
        /// one-record-per-key subscription persist/reload roundtrip.
        #[derive(Default)]
        struct MemKv {
            blobs: std::collections::HashMap<u16, std::vec::Vec<u8>>,
        }

        impl KvBlobStore for &mut MemKv {
            fn load<'a>(&mut self, key: u16, buf: &'a mut [u8]) -> Result<Option<&'a [u8]>, Error> {
                Ok(self.blobs.get(&key).map(|v| {
                    buf[..v.len()].copy_from_slice(v);
                    &buf[..v.len()]
                }))
            }

            fn store(&mut self, key: u16, data: &[u8], _buf: &mut [u8]) -> Result<(), Error> {
                self.blobs.insert(key, data.to_vec());
                Ok(())
            }

            fn remove(&mut self, key: u16, _buf: &mut [u8]) -> Result<(), Error> {
                self.blobs.remove(&key);
                Ok(())
            }
        }

        /// Add a subscription whose RX buffer carries `req` (standing in for a raw
        /// `SubscribeReq` TLV), and commit it into the table.
        #[allow(clippy::too_many_arguments)]
        fn add_sub_with_req<'a, 's, const N: usize, const B: usize>(
            subs: &'s Subscriptions<N>,
            subs_bufs: &'s SubscriptionsBuffers<'a, TestPool<B>, N>,
            pool: &'a TestPool<B>,
            now: Instant,
            fab_idx: u8,
            peer_node_id: u64,
            min_int: u16,
            max_int: u16,
            req: &[u8],
        ) where
            'a: 's,
        {
            let mut rx = pool.get_immediate().unwrap();
            rx.clear();
            rx.extend_from_slice(req).unwrap();

            let mut rctx = subs
                .add(
                    now,
                    fab(fab_idx),
                    peer_node_id,
                    min_int,
                    max_int,
                    0,
                    rx,
                    subs_bufs,
                )
                .unwrap();
            // Commit it as a live subscription (`add` already stamped `reported_at`
            // with `now`, so this is a non-priming entry once kept).
            rctx.set_keep();
        }

        /// A subscription persisted by one `Subscriptions` instance is faithfully
        /// re-hydrated — routing IDs, intervals, and the raw request bytes — into a
        /// fresh instance, exactly as a reboot would.
        #[test]
        fn persist_reload_roundtrip() {
            let mut kv = MemKv::default();
            let now = Instant::now();

            // --- First "boot": build a table and persist it. ---
            {
                let subs: Subscriptions<4> = Subscriptions::new();
                let pool = TestPool::<5>::new();
                let subs_bufs: SubscriptionsBuffers<TestPool<5>, 4> = SubscriptionsBuffers::new();

                add_sub_with_req(
                    &subs,
                    &subs_bufs,
                    &pool,
                    now,
                    1,
                    0xAABB,
                    1,
                    60,
                    &[1, 2, 3, 4],
                );
                add_sub_with_req(&subs, &subs_bufs, &pool, now, 2, 0xCCDD, 0, 120, &[9, 8, 7]);

                let mut buf = [0u8; 512];
                subs.persist_all(&subs_bufs, &mut kv, &mut buf).unwrap();
            }

            // Two records written; the rest of the reserved range is empty.
            assert!(kv.blobs.contains_key(&PERSISTENT_SUBSCRIPTIONS_START));
            assert!(kv.blobs.contains_key(&(PERSISTENT_SUBSCRIPTIONS_START + 1)));
            assert!(!kv.blobs.contains_key(&(PERSISTENT_SUBSCRIPTIONS_START + 2)));

            // --- Second "boot": a fresh, empty table reloads from the same store. ---
            let subs2: Subscriptions<4> = Subscriptions::new();
            let pool2 = TestPool::<5>::new();
            let subs_bufs2: SubscriptionsBuffers<TestPool<5>, 4> = SubscriptionsBuffers::new();

            let mut buf = [0u8; 512];
            subs2
                .load_persist(&pool2, &subs_bufs2, &mut kv, &mut buf, now, 0)
                .unwrap();

            // Both subscriptions are back, with their routing + intervals intact...
            subs2.state.lock(|s| {
                let s = s.borrow();
                assert_eq!(s.subscriptions.len(), 2);

                let a = s
                    .subscriptions
                    .iter()
                    .find(|x| x.ids.peer_node_id == 0xAABB)
                    .expect("first sub resumed");
                assert_eq!(a.ids.fab_idx, fab(1));
                assert_eq!(a.min_int_secs, 1);
                assert_eq!(a.max_int_secs, 60);

                let b = s
                    .subscriptions
                    .iter()
                    .find(|x| x.ids.peer_node_id == 0xCCDD)
                    .expect("second sub resumed");
                assert_eq!(b.ids.fab_idx, fab(2));
                assert_eq!(b.min_int_secs, 0);
                assert_eq!(b.max_int_secs, 120);

                // Resumed subscriptions must be un-primed so the reporter sends a
                // prompt priming report (resetting the subscriber's own liveness
                // clock before it can time the subscription out), NOT wait toward
                // our max-interval deadline.
                for sub in s.subscriptions.iter() {
                    assert!(
                        sub.is_report_due(now),
                        "resumed sub {:?} should be immediately report-due (primed)",
                        sub.ids()
                    );
                    assert!(sub.is_report_due(now + Duration::from_secs(1)));
                }
            });

            // ...and the raw request bytes survived (they live in the parallel buffer
            // pool, indexed alongside the subscriptions).
            subs_bufs2.with(|bufs| {
                let mut reqs: std::vec::Vec<std::vec::Vec<u8>> =
                    bufs.iter().map(|b| b[..].to_vec()).collect();
                reqs.sort();
                assert_eq!(reqs, std::vec![std::vec![1, 2, 3, 4], std::vec![9, 8, 7]]);
            });

            // The ICD predicate now sees the resumed subscriptions.
            assert!(subs2.has_subscription_for(fab(1), 0xAABB));
            assert!(subs2.has_subscription_for(fab(2), 0xCCDD));
            assert!(!subs2.has_subscription_for(fab(1), 0x9999));
        }

        /// Removing a subscription and re-persisting drops exactly its record from
        /// the store (the table stays an exact mirror of what is on disk).
        #[test]
        fn persist_mirrors_removal() {
            let mut kv = MemKv::default();
            let now = Instant::now();

            let subs: Subscriptions<4> = Subscriptions::new();
            let pool = TestPool::<5>::new();
            let subs_bufs: SubscriptionsBuffers<TestPool<5>, 4> = SubscriptionsBuffers::new();

            add_sub_with_req(&subs, &subs_bufs, &pool, now, 1, 100, 1, 60, &[1]);
            add_sub_with_req(&subs, &subs_bufs, &pool, now, 1, 101, 1, 60, &[2]);
            add_sub_with_req(&subs, &subs_bufs, &pool, now, 2, 102, 1, 60, &[3]);

            let mut buf = [0u8; 512];
            subs.persist_all(&subs_bufs, &mut kv, &mut buf).unwrap();
            assert_eq!(kv.blobs.len(), 3);

            // Drop the two fab(1) subscriptions and re-persist.
            subs.remove(&subs_bufs, |sub| {
                (sub.ids().fab_idx == fab(1)).then_some("fabric 1 removed")
            });
            subs.persist_all(&subs_bufs, &mut kv, &mut buf).unwrap();

            // Only the single surviving record remains on disk.
            assert_eq!(kv.blobs.len(), 1);

            // And a fresh reload sees exactly that one.
            let subs2: Subscriptions<4> = Subscriptions::new();
            let pool2 = TestPool::<5>::new();
            let subs_bufs2: SubscriptionsBuffers<TestPool<5>, 4> = SubscriptionsBuffers::new();
            subs2
                .load_persist(&pool2, &subs_bufs2, &mut kv, &mut buf, now, 0)
                .unwrap();
            subs2.state.lock(|s| {
                let s = s.borrow();
                assert_eq!(s.subscriptions.len(), 1);
                assert_eq!(s.subscriptions[0].ids.peer_node_id, 102);
            });
        }

        /// `reset_persist` wipes the whole reserved range.
        #[test]
        fn reset_persist_clears_all_records() {
            let mut kv = MemKv::default();
            let now = Instant::now();

            let subs: Subscriptions<4> = Subscriptions::new();
            let pool = TestPool::<5>::new();
            let subs_bufs: SubscriptionsBuffers<TestPool<5>, 4> = SubscriptionsBuffers::new();

            add_sub_with_req(&subs, &subs_bufs, &pool, now, 1, 100, 1, 60, &[1]);
            add_sub_with_req(&subs, &subs_bufs, &pool, now, 2, 101, 1, 60, &[2]);

            let mut buf = [0u8; 512];
            subs.persist_all(&subs_bufs, &mut kv, &mut buf).unwrap();
            assert_eq!(kv.blobs.len(), 2);

            subs.reset_persist(&mut kv, &mut buf).unwrap();
            assert!(kv.blobs.is_empty());
        }
    }
}