armdb 0.11.2

use std::cell::Cell;
use std::collections::VecDeque;
use std::future::Future;
use std::marker::PhantomData;
use std::num::NonZeroUsize;
use std::pin::Pin;
use std::ptr;
use std::sync::Arc;
use std::sync::atomic::{AtomicPtr, AtomicUsize, Ordering};
use std::task::{Context, Poll, Waker};
use std::time::{Duration, Instant};

use seize::Guard;

use crate::sync::{Condvar, Mutex, cv_wait, cv_wait_timeout, lock};
use crate::tree_meta::CollectionMeta;
use crate::{DbResult, Key};

/// Marker traits used to prevent external collection implementations from
/// claiming watch support without forwarding the runtime channel.
pub(crate) mod sealed {
    pub trait Spec {}
    pub trait Enabled {}
}

/// Decode a raw watch payload on the receiving side.
pub trait WatchDecoder<T>: Send + Sync {
    /// Decode one stored value payload.
    fn decode(&self, bytes: &[u8]) -> DbResult<T>;
}

/// A watch decoder backed by the collection's [`crate::Codec`].
pub struct CodecWatchDecoder<C, T> {
    codec: C,
    _value: PhantomData<fn() -> T>,
}
impl<C: Clone, T> Clone for CodecWatchDecoder<C, T> {
    fn clone(&self) -> Self {
        Self {
            codec: self.codec.clone(),
            _value: PhantomData,
        }
    }
}
impl<C: crate::codec::Codec<T>, T> WatchDecoder<T> for CodecWatchDecoder<C, T> {
    fn decode(&self, bytes: &[u8]) -> DbResult<T> {
        self.codec.decode_from(bytes)
    }
}
impl<C, T> CodecWatchDecoder<C, T> {
    pub(crate) fn new(codec: C) -> Self {
        Self {
            codec,
            _value: PhantomData,
        }
    }
}

/// The collection identity and decoder used by a watch receiver.
pub trait WatchSpec: sealed::Spec {
    /// Collection key carried by each event.
    type Key: Key;
    /// Decoded collection value returned by [`WatchEvent`].
    type Value;
    /// Decoder bound to each receiver at subscription time.
    type Decoder: WatchDecoder<Self::Value> + Clone;
}

/// Runtime carrier that adds subscriptions around an existing collection hook.
pub struct WatchHook<T: CollectionMeta, H, const OLD: bool> {
    pub(crate) channel: Arc<WatchChannel<T::SelfId>>,
    pub inner: H,
}

impl<T: CollectionMeta, H> WatchHook<T, H, true> {
    /// Subscribe with both old and new values retained by future publishers.
    pub fn with_old(inner: H) -> Self {
        Self {
            channel: WatchChannel::new(),
            inner,
        }
    }
}

impl<T: CollectionMeta, H> WatchHook<T, H, false> {
    /// Subscribe with new values only; future updates are reported as upserts.
    pub fn new_only(inner: H) -> Self {
        Self {
            channel: WatchChannel::new(),
            inner,
        }
    }
}

impl<T: CollectionMeta, H, const OLD: bool> Drop for WatchHook<T, H, OLD> {
    fn drop(&mut self) {
        self.channel.close_terminal();
    }
}

impl<T, H, const OLD: bool> crate::hook::TypedWriteHook<T::SelfId, T> for WatchHook<T, H, OLD>
where
    T: CollectionMeta,
    H: crate::hook::TypedWriteHook<T::SelfId, T>,
{
    const WATCHES: bool = true;
    const NEEDS_OLD_VALUE: bool = OLD || H::NEEDS_OLD_VALUE;
    const NEEDS_INIT: bool = H::NEEDS_INIT;
    const NEEDS_WRITE: bool = H::NEEDS_WRITE;
    const EXPIRES: bool = H::EXPIRES;
    const TTL_REVISION: u32 = H::TTL_REVISION;
    const PROJECTS: bool = H::PROJECTS;
    const COUNTS: bool = H::COUNTS;
    const CLAIMS: bool = H::CLAIMS;
    #[cfg(feature = "armour")]
    const EMITS: bool = H::EMITS;

    fn watch_channel(&self) -> Option<&Arc<WatchChannel<T::SelfId>>> {
        Some(&self.channel)
    }
    fn watch_old(&self) -> bool {
        OLD
    }
    fn deadline(&self, key: &T::SelfId, value: &T) -> Option<u64> {
        self.inner.deadline(key, value)
    }
    fn on_write(&self, key: &T::SelfId, old: Option<&T>, new: Option<&T>) {
        self.inner.on_write(key, old, new)
    }
    fn on_init(&self, key: &T::SelfId, value: &T) {
        self.inner.on_init(key, value)
    }
    fn project(
        &self,
        key: &T::SelfId,
        value: &T,
        sink: &mut crate::projection::ProjectionSink<'_>,
    ) -> DbResult<()> {
        self.inner.project(key, value, sink)
    }
    fn projection_targets(&self) -> Option<&crate::projection::IndexTargets> {
        self.inner.projection_targets()
    }
    fn projection_limits(&self) -> (usize, usize) {
        self.inner.projection_limits()
    }
    fn counter_project(
        &self,
        key: &T::SelfId,
        value: &T,
        sink: &mut crate::counter::CounterSink<'_>,
    ) -> DbResult<()> {
        self.inner.counter_project(key, value, sink)
    }
    fn counter_targets(&self) -> Option<&crate::counter::CounterTargets> {
        self.inner.counter_targets()
    }
    fn claim_project(
        &self,
        key: &T::SelfId,
        value: &T,
        sink: &mut crate::unique::ClaimSink<'_>,
    ) -> DbResult<()> {
        self.inner.claim_project(key, value, sink)
    }
    fn claim_targets(&self) -> Option<&crate::unique::UniqueTargets> {
        self.inner.claim_targets()
    }
    #[cfg(feature = "armour")]
    fn stream_project(
        &self,
        key: &T::SelfId,
        old: Option<&T>,
        new: Option<&T>,
        sink: &mut crate::EmitSink<'_>,
    ) -> DbResult<()> {
        self.inner.stream_project(key, old, new, sink)
    }
    #[cfg(feature = "armour")]
    fn stream_targets(&self) -> Option<&crate::QueueTargets> {
        self.inner.stream_targets()
    }
}

impl<T, H, const OLD: bool> sealed::Enabled for WatchHook<T, H, OLD>
where
    T: CollectionMeta,
    H: crate::hook::TypedWriteHook<T::SelfId, T>,
{
}

impl<T, H> sealed::Enabled for crate::ttl::TtlHook<T, H>
where
    T: CollectionMeta,
    H: sealed::Enabled,
{
}

impl<K, T, H, E> sealed::Enabled for crate::projection::Projected<K, T, H, E> where
    H: sealed::Enabled
{
}

#[cfg(feature = "typed-tree")]
impl<K, T, H> sealed::Enabled for crate::hook::ZeroHookAdapter<K, T, H> where H: sealed::Enabled {}

#[cfg(all(feature = "typed-tree", feature = "var-collections"))]
impl<K, T, C, H> sealed::Enabled for crate::hook::VarTypedHookAdapter<K, T, C, H> where
    H: sealed::Enabled
{
}

/// Decoder for fixed-size zero-copy collection values.
pub struct ZeroWatchDecoder<T>(PhantomData<fn() -> T>);

impl<T> Clone for ZeroWatchDecoder<T> {
    fn clone(&self) -> Self {
        Self(PhantomData)
    }
}
impl<T> ZeroWatchDecoder<T> {
    pub(crate) fn new() -> Self {
        Self(PhantomData)
    }
}

impl<T: Copy + zerocopy::FromBytes> WatchDecoder<T> for ZeroWatchDecoder<T> {
    fn decode(&self, bytes: &[u8]) -> DbResult<T> {
        zerocopy::FromBytes::read_from_bytes(bytes)
            .map_err(|_| crate::error::DbError::CorruptedEntry { offset: 0 })
    }
}

/// Delivery detail available from a raw mutation.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WatchKind {
    /// The key was previously absent and now has a value with old-value mode.
    Insert,
    /// The key's old and new values are both retained.
    Update,
    /// A value is present but no old value was retained.
    Upsert,
    /// The key was removed.
    Delete,
}

/// Receiver outcomes that are not decoded-value errors.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WatchError {
    /// The receiver ring is currently empty.
    Empty,
    /// A timed receive reached its deadline without an event.
    Timeout,
    /// The receiver fell behind and lost this many events.
    Lagged(u64),
    /// The collection's terminal owner was dropped or shut down.
    Closed,
}

/// Per-subscription bounded-ring configuration.
#[derive(Clone, Copy, Debug)]
pub struct WatchOptions {
    /// Per-receiver ring capacity.  A full ring reports [`WatchError::Lagged`].
    pub capacity: NonZeroUsize,
}

impl Default for WatchOptions {
    fn default() -> Self {
        Self {
            capacity: NonZeroUsize::new(1024).expect("1024 is non-zero"),
        }
    }
}

/// A decoded view of one shared raw watch event.
pub struct WatchEvent<S: WatchSpec> {
    inner: Arc<WatchEventInner<S::Key>>,
    decoder: S::Decoder,
    _spec: PhantomData<fn() -> S>,
}

impl<S: WatchSpec> std::fmt::Debug for WatchEvent<S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("WatchEvent")
            .field("kind", &self.kind())
            .finish_non_exhaustive()
    }
}

impl<S: WatchSpec> PartialEq for WatchEvent<S> {
    fn eq(&self, other: &Self) -> bool {
        Arc::ptr_eq(&self.inner, &other.inner)
    }
}

impl<S: WatchSpec> Eq for WatchEvent<S> {}

impl<S: WatchSpec> WatchEvent<S> {
    /// Event key.
    pub fn key(&self) -> &S::Key {
        &self.inner.key
    }

    /// Event kind derived from the retained old/new payloads.
    pub fn kind(&self) -> WatchKind {
        self.inner.kind
    }

    #[allow(clippy::new_ret_no_self, clippy::wrong_self_convention)]
    /// Decode the new value, when the mutation has one.
    pub fn new(&self) -> DbResult<Option<S::Value>> {
        self.inner
            .new
            .as_deref()
            .map(|bytes| self.decoder.decode(bytes))
            .transpose()
    }

    /// Decode the old value, when `with_old` retained one.
    pub fn old(&self) -> DbResult<Option<S::Value>> {
        self.inner
            .old
            .as_deref()
            .map(|bytes| self.decoder.decode(bytes))
            .transpose()
    }

    /// Borrow the stored bytes for the new value without decoding.
    pub fn new_bytes(&self) -> Option<&[u8]> {
        self.inner.new.as_deref()
    }

    /// Borrow the stored bytes for the old value without decoding.
    pub fn old_bytes(&self) -> Option<&[u8]> {
        self.inner.old.as_deref()
    }
}

pub(crate) struct WatchEventInner<K> {
    key: K,
    old: Option<Box<[u8]>>,
    new: Option<Box<[u8]>>,
    kind: WatchKind,
}

impl<K> WatchEventInner<K> {
    /// Only the test publish helper builds an event outside a mutation.
    #[cfg(test)]
    pub(crate) fn new(key: K, old: Option<Box<[u8]>>, new: Option<Box<[u8]>>) -> Self {
        let existed = old.is_some();
        Self::from_write(key, old, new, false, existed)
    }

    /// Build an event from a watched write.
    ///
    /// `retains_old` distinguishes an insertion observed in `with_old` mode
    /// from a new-only upsert. `old_existed` is a fact about the mutation, not
    /// about this event's payload: a row that was overwritten is an `Update`
    /// even when its previous bytes could not be materialized. Deriving the
    /// kind from `old.is_some()` instead would let a failed materialization
    /// silently report an `Insert` over a live row.
    pub(crate) fn from_write(
        key: K,
        old: Option<Box<[u8]>>,
        new: Option<Box<[u8]>>,
        retains_old: bool,
        old_existed: bool,
    ) -> Self {
        let kind = match (&new, retains_old, old_existed) {
            (Some(_), true, true) => WatchKind::Update,
            (Some(_), true, false) => WatchKind::Insert,
            (Some(_), false, _) => WatchKind::Upsert,
            (None, _, _) => WatchKind::Delete,
        };
        Self {
            key,
            old,
            new,
            kind,
        }
    }
}

/// Wakers taken under subscriber locks and invoked after the caller releases
/// every lock relevant to a mutation.
pub(crate) struct WatchWakeBatch {
    inline: [Option<Waker>; 4],
    used: usize,
    spill: Vec<Waker>,
}

impl Default for WatchWakeBatch {
    fn default() -> Self {
        Self {
            inline: std::array::from_fn(|_| None),
            used: 0,
            spill: Vec::new(),
        }
    }
}

impl WatchWakeBatch {
    pub(crate) fn extend(&mut self, other: Self) {
        for waker in other.inline.into_iter().take(other.used).flatten() {
            self.push(waker);
        }
        self.spill.extend(other.spill);
    }

    fn push(&mut self, waker: Waker) {
        if self.used < self.inline.len() {
            self.inline[self.used] = Some(waker);
            self.used += 1;
        } else {
            self.spill.push(waker);
        }
    }

    pub(crate) fn drain(mut self) {
        for waker in self.inline[..self.used].iter_mut().filter_map(Option::take) {
            waker.wake();
        }
        for waker in self.spill.drain(..) {
            waker.wake();
        }
    }
}

/// Shared bounded fan-out state owned by a [`WatchHook`].
pub struct WatchChannel<K: Key> {
    head: AtomicPtr<SubscriberNode<K>>,
    registry: Mutex<()>,
    collector: seize::Collector,
    subscribers: AtomicUsize,
    gate: ProducerPhases,
    #[cfg(test)]
    publish_pause: std::sync::Mutex<Option<test_support::ParkPause>>,
    #[cfg(test)]
    close_pause: std::sync::Mutex<Option<test_support::ParkPause>>,
    #[cfg(test)]
    traversal_pause: std::sync::Mutex<Option<test_support::ParkPause>>,
    #[cfg(test)]
    ctx_lock_pause: std::sync::Mutex<Option<test_support::ParkPause>>,
    #[cfg(test)]
    gate_pause: std::sync::Mutex<Option<test_support::ParkPause>>,
    #[cfg(test)]
    subscribe_pause: std::sync::Mutex<Option<test_support::ParkPause>>,
}

impl<K: Key> WatchChannel<K> {
    pub(crate) fn new() -> Arc<Self> {
        Arc::new(Self {
            head: AtomicPtr::new(ptr::null_mut()),
            registry: Mutex::new(()),
            collector: seize::Collector::new(),
            subscribers: AtomicUsize::new(0),
            gate: ProducerPhases::new(),
            #[cfg(test)]
            publish_pause: std::sync::Mutex::new(None),
            #[cfg(test)]
            close_pause: std::sync::Mutex::new(None),
            #[cfg(test)]
            traversal_pause: std::sync::Mutex::new(None),
            #[cfg(test)]
            ctx_lock_pause: std::sync::Mutex::new(None),
            #[cfg(test)]
            gate_pause: std::sync::Mutex::new(None),
            #[cfg(test)]
            subscribe_pause: std::sync::Mutex::new(None),
        })
    }

    /// Sample the subscriber gate. Callers must already hold a
    /// [`WatchProducer`]: the guard is what makes a `false` answer safe, by
    /// forcing a receiver that registers afterwards to wait for this producer
    /// to make its row visible (so the receiver's own scan covers it).
    pub(crate) fn has_subscribers(&self) -> bool {
        !self.gate.is_closed() && self.subscribers.load(Ordering::Acquire) != 0
    }

    /// Join the current producer phase. `None` once the channel is terminal.
    pub(crate) fn enter_producer(&self) -> Option<WatchProducer<'_>> {
        let phase = self.gate.enter()?;
        Some(WatchProducer {
            gate: Some((&self.gate, phase)),
            active: self.has_subscribers(),
        })
    }

    pub(crate) fn subscribe<S: WatchSpec<Key = K>>(
        self: &Arc<Self>,
        decoder: S::Decoder,
        options: WatchOptions,
    ) -> WatchReceiver<S> {
        let subscriber = Arc::new(Subscriber::new(options.capacity));
        if self.is_closed() {
            subscriber.close(&mut WatchWakeBatch::default());
            return WatchReceiver {
                subscriber,
                channel: Arc::clone(self),
                node: ptr::null_mut(),
                decoder,
                _spec: PhantomData,
            };
        }
        let node = Box::into_raw(Box::new(SubscriberNode {
            subscriber: Arc::clone(&subscriber),
            next: AtomicPtr::new(ptr::null_mut()),
        }));
        {
            let _registry = lock(&self.registry);
            if self.is_closed() {
                // SAFETY: the node was never published into the intrusive list.
                unsafe { drop(Box::from_raw(node)) };
                subscriber.close(&mut WatchWakeBatch::default());
                return WatchReceiver {
                    subscriber,
                    channel: Arc::clone(self),
                    node: ptr::null_mut(),
                    decoder,
                    _spec: PhantomData,
                };
            }
            // SAFETY: `node` is freshly allocated and not visible until the
            // following release store publishes it.
            unsafe {
                (*node)
                    .next
                    .store(self.head.load(Ordering::Relaxed), Ordering::Relaxed);
            }
            self.head.store(node, Ordering::Release);
            self.subscribers.fetch_add(1, Ordering::Release);
        }
        #[cfg(test)]
        test_support::wait_pause(&self.subscribe_pause);
        // Every producer that may still observe the pre-registration count has
        // to finish before `watch()` returns. Those producers publish nothing
        // to this receiver, so the caller's catch-up scan is what must cover
        // them — and a scan only covers a row that is already visible, which is
        // exactly what leaving the producer phase guarantees.
        self.gate.flip_and_drain();
        WatchReceiver {
            subscriber,
            channel: Arc::clone(self),
            node,
            decoder,
            _spec: PhantomData,
        }
    }

    /// Fan one event out to every linked subscriber.
    ///
    /// The caller must hold the [`WatchProducer`] it sampled the gate under —
    /// that guard, not a check here, is what orders publication against both a
    /// concurrent registration and a terminal close. An event built under the
    /// guard is still delivered while `close_terminal` waits, and the drain
    /// there is what gets it to the receiver before `Closed`.
    pub(crate) fn publish(&self, event: Arc<WatchEventInner<K>>, wakes: &mut WatchWakeBatch) {
        #[cfg(test)]
        test_support::wait_pause(&self.publish_pause);
        if self.subscribers.load(Ordering::Acquire) != 0 {
            let guard = self.collector.enter();
            let mut current = guard.protect(&self.head, Ordering::Acquire);
            while let Some(node) = unsafe { current.as_ref() } {
                #[cfg(test)]
                test_support::wait_pause(&self.traversal_pause);
                node.subscriber.push(Arc::clone(&event), wakes);
                current = guard.protect(&node.next, Ordering::Acquire);
            }
        }
    }

    /// Tell every subscriber that a mutation happened whose event could not be
    /// built. Delivery is at-most-once, but a loss is never silent: it is
    /// reported as the same `Lagged` gap a ring overflow produces, so the
    /// consumer's existing resync path covers it.
    pub(crate) fn publish_lost(&self, wakes: &mut WatchWakeBatch) {
        if self.subscribers.load(Ordering::Acquire) == 0 {
            return;
        }
        let guard = self.collector.enter();
        let mut current = guard.protect(&self.head, Ordering::Acquire);
        while let Some(node) = unsafe { current.as_ref() } {
            node.subscriber.record_gap(wakes);
            current = guard.protect(&node.next, Ordering::Acquire);
        }
    }

    pub(crate) fn close_terminal(&self) {
        if !self.gate.close() {
            return;
        }
        #[cfg(test)]
        test_support::wait_pause(&self.close_pause);
        self.gate.drain_all();
        let mut wakes = WatchWakeBatch::default();
        let _registry = lock(&self.registry);
        let guard = self.collector.enter();
        let mut current = guard.protect(&self.head, Ordering::Acquire);
        while let Some(node) = unsafe { current.as_ref() } {
            node.subscriber.close(&mut wakes);
            current = guard.protect(&node.next, Ordering::Acquire);
        }
        drop(guard);
        drop(_registry);
        wakes.drain();
    }

    fn is_closed(&self) -> bool {
        self.gate.is_closed()
    }

    fn unsubscribe(&self, node: *mut SubscriberNode<K>) {
        self.subscribers.fetch_sub(1, Ordering::Release);
        let _registry = lock(&self.registry);
        let mut link = &self.head;
        loop {
            let current = link.load(Ordering::Acquire);
            if current.is_null() {
                return;
            }
            if current == node {
                // SAFETY: registry mutations are serialized; `current` remains
                // linked until this release store makes it unreachable.
                let next = unsafe { (*current).next.load(Ordering::Acquire) };
                link.store(next, Ordering::Release);
                // SAFETY: the node is unlinked and producer traversals protect
                // every pointer load with this collector.
                unsafe {
                    self.collector
                        .retire(current, seize::reclaim::boxed::<SubscriberNode<K>>);
                }
                return;
            }
            // SAFETY: a linked node cannot be retired while registry lock is held.
            link = unsafe { &(*current).next };
        }
    }
}

const CLOSED: usize = 1usize << (usize::BITS - 1);
const PHASE: usize = 1;

/// Producer-participation gate.
///
/// A producer samples the subscriber count *before* it makes its row visible,
/// so a bare count cannot be trusted on its own: a receiver registering between
/// that sample and the mutation would find the row in neither the stream (the
/// producer decided not to publish) nor its own catch-up scan (the row was not
/// visible yet). The gate closes that window by making the receiver wait for
/// those producers to finish.
///
/// Waiting on a single in-flight counter would starve under continuous writes,
/// so participation is split across two phases. A registering receiver flips
/// the phase new producers join, then waits only for the phase it left behind —
/// a set that is fixed at the moment of the flip and therefore drains.
struct ProducerPhases {
    /// [`PHASE`] selects the phase new producers join; [`CLOSED`] is terminal.
    phase: AtomicUsize,
    counts: [AtomicUsize; 2],
}

impl ProducerPhases {
    fn new() -> Self {
        Self {
            phase: AtomicUsize::new(0),
            counts: [AtomicUsize::new(0), AtomicUsize::new(0)],
        }
    }

    fn is_closed(&self) -> bool {
        self.phase.load(Ordering::Acquire) & CLOSED != 0
    }

    /// Join the current phase, or fail once the channel is terminal.
    ///
    /// The re-check after the increment is what keeps "no push lands after
    /// close" true: `close` sets the flag and only then waits for the counts,
    /// so a producer that read the phase just before the flag must withdraw
    /// itself rather than rely on having been seen.
    fn enter(&self) -> Option<usize> {
        let observed = self.phase.load(Ordering::Acquire);
        if observed & CLOSED != 0 {
            return None;
        }
        let idx = observed & PHASE;
        self.counts[idx].fetch_add(1, Ordering::AcqRel);
        if self.is_closed() {
            self.counts[idx].fetch_sub(1, Ordering::Release);
            return None;
        }
        Some(idx)
    }

    fn leave(&self, idx: usize) {
        self.counts[idx].fetch_sub(1, Ordering::Release);
    }

    /// Send new producers to the other phase, then wait out this one.
    ///
    /// The flip is `AcqRel` and the caller published its registration with a
    /// `Release` store beforehand, so any producer that joins the new phase is
    /// guaranteed to observe that registration.
    fn flip_and_drain(&self) {
        let previous = self.phase.fetch_xor(PHASE, Ordering::AcqRel) & PHASE;
        while self.counts[previous].load(Ordering::Acquire) != 0 {
            std::thread::yield_now();
        }
    }

    /// Mark terminal. `false` when another caller already did.
    fn close(&self) -> bool {
        let mut observed = self.phase.load(Ordering::Acquire);
        loop {
            if observed & CLOSED != 0 {
                return false;
            }
            match self.phase.compare_exchange_weak(
                observed,
                observed | CLOSED,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => return true,
                Err(seen) => observed = seen,
            }
        }
    }

    fn drain_all(&self) {
        while self.counts[0].load(Ordering::Acquire) != 0
            || self.counts[1].load(Ordering::Acquire) != 0
        {
            std::thread::yield_now();
        }
    }
}

/// Participation in a channel's current producer phase, held across a mutation.
///
/// Entered before the subscriber gate is sampled and dropped only after the
/// event has been published, so the interval covers the whole window in which a
/// stale `active == false` could otherwise lose a row. [`Self::inactive`] is the
/// branch-free form for a family that carries no watch.
pub(crate) struct WatchProducer<'a> {
    gate: Option<(&'a ProducerPhases, usize)>,
    active: bool,
}

impl WatchProducer<'_> {
    #[inline]
    pub(crate) fn inactive() -> Self {
        Self {
            gate: None,
            active: false,
        }
    }

    /// Whether a receiver was linked when this producer joined its phase.
    #[inline]
    pub(crate) fn active(&self) -> bool {
        self.active
    }
}

impl Drop for WatchProducer<'_> {
    #[inline]
    fn drop(&mut self) {
        if let Some((gate, idx)) = self.gate {
            gate.leave(idx);
        }
    }
}

thread_local! {
    static CONTEXT_ACTIVE: Cell<bool> = const { Cell::new(false) };
}

/// Whether the current context captured an active receiver at its producer
/// linearization point.  Collection internals use this only to avoid building
/// watch payloads; public callbacks must use their explicit wake path.
#[inline]
pub(crate) fn context_active() -> bool {
    CONTEXT_ACTIVE.with(Cell::get)
}

pub(crate) struct ContextActiveGuard(bool);

impl ContextActiveGuard {
    #[inline]
    pub(crate) fn set(active: bool) -> Self {
        Self(CONTEXT_ACTIVE.with(|state| state.replace(active)))
    }
}

impl Drop for ContextActiveGuard {
    fn drop(&mut self) {
        CONTEXT_ACTIVE.with(|state| state.set(self.0));
    }
}

struct SubscriberNode<K: Key> {
    subscriber: Arc<Subscriber<K>>,
    next: AtomicPtr<SubscriberNode<K>>,
}

struct Subscriber<K: Key> {
    state: Mutex<SubscriberState<K>>,
    changed: Condvar,
    capacity: usize,
}

struct SubscriberState<K: Key> {
    events: VecDeque<Arc<WatchEventInner<K>>>,
    lagged: u64,
    gap_at: Option<usize>,
    waker: Option<Waker>,
    terminal: bool,
}

impl<K: Key> Subscriber<K> {
    fn new(capacity: NonZeroUsize) -> Self {
        Self {
            state: Mutex::new(SubscriberState {
                events: VecDeque::with_capacity(capacity.get()),
                lagged: 0,
                gap_at: None,
                waker: None,
                terminal: false,
            }),
            changed: Condvar::new(),
            capacity: capacity.get(),
        }
    }

    fn push(&self, event: Arc<WatchEventInner<K>>, wakes: &mut WatchWakeBatch) {
        let mut state = lock(&self.state);
        if state.terminal {
            return;
        }
        if state.events.len() == self.capacity {
            let _ = state.events.pop_front();
            state.lagged = state.lagged.saturating_add(1);
            state.gap_at = Some(match state.gap_at {
                Some(gap_at) => gap_at.saturating_sub(1),
                None => 0,
            });
        }
        state.events.push_back(event);
        self.changed.notify_one();
        if let Some(waker) = state.waker.take() {
            wakes.push(waker);
        }
    }

    /// Record an event that was produced but could not be delivered.
    ///
    /// The gap belongs after everything already queued — that is where the loss
    /// happened — so the consumer drains what it has and only then sees
    /// `Lagged`. An earlier, still-unreported gap keeps its position.
    fn record_gap(&self, wakes: &mut WatchWakeBatch) {
        let mut state = lock(&self.state);
        if state.terminal {
            return;
        }
        state.lagged = state.lagged.saturating_add(1);
        let at = state.events.len();
        state.gap_at = Some(state.gap_at.map_or(at, |existing| existing.min(at)));
        self.changed.notify_one();
        if let Some(waker) = state.waker.take() {
            wakes.push(waker);
        }
    }

    fn close(&self, wakes: &mut WatchWakeBatch) {
        let mut state = lock(&self.state);
        state.terminal = true;
        self.changed.notify_all();
        if let Some(waker) = state.waker.take() {
            wakes.push(waker);
        }
    }
}

/// A single-consumer subscription receiver.
pub struct WatchReceiver<S: WatchSpec> {
    subscriber: Arc<Subscriber<S::Key>>,
    channel: Arc<WatchChannel<S::Key>>,
    node: *mut SubscriberNode<S::Key>,
    decoder: S::Decoder,
    _spec: PhantomData<fn() -> S>,
}

impl<S: WatchSpec> WatchReceiver<S> {
    pub(crate) fn subscribe(
        channel: &Arc<WatchChannel<S::Key>>,
        decoder: S::Decoder,
        options: WatchOptions,
    ) -> Self {
        channel.subscribe(decoder, options)
    }
}

impl<S: WatchSpec> WatchReceiver<S> {
    pub fn try_recv(&mut self) -> Result<WatchEvent<S>, WatchError> {
        let mut state = lock(&self.subscriber.state);
        Self::take_next(&mut state, &self.decoder).unwrap_or(Err(WatchError::Empty))
    }

    pub fn recv_blocking(&mut self) -> Result<WatchEvent<S>, WatchError> {
        loop {
            {
                let mut state = lock(&self.subscriber.state);
                if let Some(next) = Self::take_next(&mut state, &self.decoder) {
                    return next;
                }
            }
            before_condvar_park();
            let mut state = lock(&self.subscriber.state);
            if let Some(next) = Self::take_next(&mut state, &self.decoder) {
                return next;
            }
            drop(cv_wait(&self.subscriber.changed, state));
        }
    }

    pub fn recv_timeout(&mut self, timeout: Duration) -> Result<WatchEvent<S>, WatchError> {
        let deadline = Instant::now() + timeout;
        loop {
            {
                let mut state = lock(&self.subscriber.state);
                if let Some(next) = Self::take_next(&mut state, &self.decoder) {
                    return next;
                }
            }
            if Instant::now() >= deadline {
                return Err(WatchError::Timeout);
            }
            let mut state = lock(&self.subscriber.state);
            if let Some(next) = Self::take_next(&mut state, &self.decoder) {
                return next;
            }
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return Err(WatchError::Timeout);
            }
            drop(cv_wait_timeout(
                &self.subscriber.changed,
                &self.subscriber.state,
                state,
                remaining,
            ));
        }
    }

    pub fn recv(&mut self) -> RecvFuture<'_, S> {
        RecvFuture {
            receiver: self,
            registered: None,
        }
    }

    pub fn recv_many(
        &mut self,
        out: &mut Vec<WatchEvent<S>>,
        max: usize,
    ) -> Result<usize, WatchError> {
        if max == 0 {
            return Ok(0);
        }
        let mut state = lock(&self.subscriber.state);
        let mut received = 0;
        while received < max {
            if state.gap_at == Some(0) {
                return if received == 0 {
                    let lagged = std::mem::take(&mut state.lagged);
                    state.gap_at = None;
                    Err(WatchError::Lagged(lagged))
                } else {
                    Ok(received)
                };
            }
            if let Some(inner) = state.events.pop_front() {
                if let Some(gap_at) = &mut state.gap_at {
                    *gap_at -= 1;
                }
                out.push(WatchEvent {
                    inner,
                    decoder: self.decoder.clone(),
                    _spec: PhantomData,
                });
                received += 1;
            } else {
                return if state.terminal && received == 0 {
                    Err(WatchError::Closed)
                } else {
                    Ok(received)
                };
            }
        }
        Ok(received)
    }

    fn take_next(
        state: &mut SubscriberState<S::Key>,
        decoder: &S::Decoder,
    ) -> Option<Result<WatchEvent<S>, WatchError>> {
        if state.gap_at == Some(0) {
            state.gap_at = None;
            let lagged = std::mem::take(&mut state.lagged);
            return Some(Err(WatchError::Lagged(lagged)));
        }
        if let Some(inner) = state.events.pop_front() {
            if let Some(gap_at) = &mut state.gap_at {
                *gap_at -= 1;
            }
            return Some(Ok(WatchEvent {
                inner,
                decoder: decoder.clone(),
                _spec: PhantomData,
            }));
        }
        if state.terminal {
            return Some(Err(WatchError::Closed));
        }
        None
    }
}

/// Future returned by [`WatchReceiver::recv`].
pub struct RecvFuture<'a, S: WatchSpec> {
    receiver: &'a mut WatchReceiver<S>,
    registered: Option<Waker>,
}

impl<S: WatchSpec> Future for RecvFuture<'_, S> {
    type Output = Result<WatchEvent<S>, WatchError>;

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.as_mut().get_mut();
        let mut state = lock(&this.receiver.subscriber.state);
        if let Some(next) = WatchReceiver::<S>::take_next(&mut state, &this.receiver.decoder) {
            return Poll::Ready(next);
        }
        let replace = state
            .waker
            .as_ref()
            .is_none_or(|registered| !registered.will_wake(cx.waker()));
        if replace {
            state.waker = Some(cx.waker().clone());
        }
        this.registered = Some(cx.waker().clone());
        Poll::Pending
    }
}

impl<S: WatchSpec> Drop for RecvFuture<'_, S> {
    fn drop(&mut self) {
        let Some(registered) = self.registered.take() else {
            return;
        };
        let mut state = lock(&self.receiver.subscriber.state);
        if state
            .waker
            .as_ref()
            .is_some_and(|waker| waker.will_wake(&registered))
        {
            state.waker = None;
        }
    }
}

impl<S: WatchSpec> Drop for WatchReceiver<S> {
    fn drop(&mut self) {
        if !self.node.is_null() {
            self.channel.unsubscribe(self.node);
        }
    }
}

// The raw pointer is only an intrusive-list identity. All dereferences are
// guarded by the channel registry or its seize collector.
unsafe impl<S: WatchSpec> Send for WatchReceiver<S> {}

#[cfg(test)]
fn before_condvar_park() {
    test_support::wait_before_condvar_park();
}

#[cfg(not(test))]
fn before_condvar_park() {}

#[cfg(test)]
fn test_channel<K: Key, T>(_capacity: NonZeroUsize) -> (Arc<WatchChannel<K>>, WatchReceiver<T>)
where
    T: WatchSpec<Key = K>,
    T::Decoder: Default,
{
    let channel = WatchChannel::new();
    let receiver = channel.subscribe::<T>(
        T::Decoder::default(),
        WatchOptions {
            capacity: _capacity,
        },
    );
    (channel, receiver)
}

#[cfg(test)]
impl<K: Key> WatchChannel<K> {
    fn publish_for_test(&self, key: K, old: Option<Box<[u8]>>, new: Option<Box<[u8]>>) {
        let mut wakes = WatchWakeBatch::default();
        // Mirror the production seam: the participation guard is entered by the
        // mutation context, and publication happens under it.
        if let Some(_producer) = self.enter_producer() {
            self.publish(Arc::new(WatchEventInner::new(key, old, new)), &mut wakes);
        }
        wakes.drain();
    }

    fn debug_registered_wakers(&self) -> usize {
        let guard = self.collector.enter();
        let mut count = 0;
        let mut current = guard.protect(&self.head, Ordering::Acquire);
        while let Some(node) = unsafe { current.as_ref() } {
            count += usize::from(lock(&node.subscriber.state).waker.is_some());
            current = guard.protect(&node.next, Ordering::Acquire);
        }
        count
    }

    fn arm_before_publish(&self) -> test_support::ParkPause {
        test_support::arm_pause(&self.publish_pause)
    }

    fn arm_after_terminal_mark(&self) -> test_support::ParkPause {
        test_support::arm_pause(&self.close_pause)
    }

    fn arm_after_traversal_starts(&self) -> test_support::ParkPause {
        test_support::arm_pause(&self.traversal_pause)
    }

    fn arm_before_ctx_lock(&self) -> test_support::ParkPause {
        test_support::arm_pause(&self.ctx_lock_pause)
    }

    pub(crate) fn wait_before_ctx_lock(&self) {
        test_support::wait_pause(&self.ctx_lock_pause);
    }

    /// Park a producer between its gate sample and its mutation — the window
    /// the participation guard exists to cover.
    fn arm_after_watch_gate(&self) -> test_support::ParkPause {
        test_support::arm_pause(&self.gate_pause)
    }

    pub(crate) fn wait_after_watch_gate(&self) {
        test_support::wait_pause(&self.gate_pause);
    }

    /// Park a registering receiver after it is linked and counted, but before
    /// it waits out the producers that may have missed it.
    fn arm_after_subscribe_link(&self) -> test_support::ParkPause {
        test_support::arm_pause(&self.subscribe_pause)
    }

    fn debug_subscriber_count(&self) -> usize {
        self.subscribers.load(Ordering::Acquire)
    }
}

#[cfg(test)]
mod test_support {
    use std::sync::{Arc, Barrier, LazyLock, Mutex};

    pub(super) type ParkPause = (Arc<Barrier>, Arc<Barrier>);

    static PARK_PAUSE: LazyLock<Mutex<Option<ParkPause>>> = LazyLock::new(|| Mutex::new(None));
    pub(super) fn arm_before_condvar_park() -> ParkPause {
        let pause = (Arc::new(Barrier::new(2)), Arc::new(Barrier::new(2)));
        *PARK_PAUSE.lock().expect("watch park pause lock") = Some(pause.clone());
        pause
    }

    pub(super) fn wait_before_condvar_park() {
        let pause = PARK_PAUSE
            .lock()
            .expect("watch park pause lock")
            .as_ref()
            .cloned();
        if let Some((checked_empty, release)) = pause {
            checked_empty.wait();
            release.wait();
        }
    }

    pub(super) fn arm_pause(pause_slot: &Mutex<Option<ParkPause>>) -> ParkPause {
        let pause = (Arc::new(Barrier::new(2)), Arc::new(Barrier::new(2)));
        *pause_slot.lock().expect("watch pause lock") = Some(pause.clone());
        pause
    }

    pub(super) fn wait_pause(pause_slot: &Mutex<Option<ParkPause>>) {
        let pause = pause_slot.lock().expect("watch pause lock").take();
        if let Some((checked, release)) = pause {
            checked.wait();
            release.wait();
        }
    }
}

#[cfg(test)]
mod tests {
    use std::future::Future;
    // Only the follower test needs a socket, and it is gated on `replication`;
    // an ungated import here is an `unused_imports` failure under the crate's
    // `-D warnings` feature-matrix builds.
    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "replication"
    ))]
    use std::net::TcpListener;
    use std::num::NonZeroUsize;
    use std::pin::Pin;
    use std::sync::Arc;
    use std::task::{Context, Poll, Waker};
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    use std::time::Duration;
    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "replication"
    ))]
    use std::time::Instant;

    use super::*;

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    use crate::{
        ByteView, CollectionMeta, Config, NoHook, RapiraCodec, TtlPolicy, TypedMap, TypedTree,
        Upsert, VarTypedMap, VarTypedTree, ZeroMap, ZeroTree,
    };
    #[cfg(feature = "rapira-codec")]
    use armour_core::GetType;
    #[cfg(feature = "rapira-codec")]
    use rapira::Rapira;
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    use tempfile::tempdir;
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    use zerocopy::{FromBytes, Immutable, IntoBytes};

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct WatchRow(u64);

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    impl CollectionMeta for WatchRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_api_rows";
    }

    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct EncodeOnceRow(u64);

    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    impl CollectionMeta for EncodeOnceRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_encode_once_rows";
    }

    // Test-only codec state deliberately allows the storage encode and rejects
    // an accidental second encode during watch publication.
    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    thread_local! {
        static ENCODE_ONCE_REMAINING: std::cell::Cell<Option<usize>> = const { std::cell::Cell::new(None) };
    }

    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    #[derive(Clone, Copy)]
    struct EncodeOnceCodec;

    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    impl crate::Codec<EncodeOnceRow> for EncodeOnceCodec {
        fn encode_to(&self, value: &EncodeOnceRow, buf: &mut Vec<u8>) -> crate::DbResult<()> {
            let allowed = ENCODE_ONCE_REMAINING.with(|remaining| match remaining.get() {
                Some(0) => false,
                Some(left) => {
                    remaining.set(Some(left - 1));
                    true
                }
                None => true,
            });
            if !allowed {
                return Err(crate::DbError::Internal("test: second codec encode"));
            }
            buf.clear();
            buf.extend_from_slice(&value.0.to_be_bytes());
            Ok(())
        }

        fn decode_from(&self, bytes: &[u8]) -> crate::DbResult<EncodeOnceRow> {
            Ok(EncodeOnceRow(u64::from_be_bytes(
                bytes
                    .try_into()
                    .map_err(|_| crate::DbError::CorruptedEntry { offset: 0 })?,
            )))
        }
    }

    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    struct EncodeOnceScope(Option<usize>);

    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    impl EncodeOnceScope {
        fn allow_one() -> Self {
            Self::allow(1)
        }

        fn allow(budget: usize) -> Self {
            let previous = ENCODE_ONCE_REMAINING.with(|remaining| {
                let previous = remaining.get();
                remaining.set(Some(budget));
                previous
            });
            Self(previous)
        }
    }

    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    impl Drop for EncodeOnceScope {
        fn drop(&mut self) {
            ENCODE_ONCE_REMAINING.with(|remaining| remaining.set(self.0));
        }
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct WatchRightRow(u64);

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    impl CollectionMeta for WatchRightRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_tx_right_rows";
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[derive(Clone, Debug, PartialEq, Rapira, GetType)]
    struct TtlWatchRow(u64);

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    impl CollectionMeta for TtlWatchRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_ttl_rows";
        const TTL: Option<TtlPolicy<Self>> = Some(TtlPolicy {
            revision: 1,
            deadline: |row, _| (row.0 > 0).then_some(row.0),
        });
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[derive(Clone, Copy, Debug, PartialEq, FromBytes, IntoBytes, Immutable, GetType)]
    #[repr(transparent)]
    struct WatchZeroRow(u64);

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    impl CollectionMeta for WatchZeroRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_api_zero_rows";
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[derive(Clone, Copy, FromBytes, IntoBytes, Immutable, GetType)]
    #[repr(transparent)]
    struct WatchZeroMapRow(u64);

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    impl CollectionMeta for WatchZeroMapRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_api_zero_map_rows";
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[derive(Clone, Copy, Debug, PartialEq, FromBytes, IntoBytes, Immutable, GetType)]
    #[repr(transparent)]
    struct TtlWatchZeroRow(u64);

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    impl CollectionMeta for TtlWatchZeroRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_ttl_zero_tree_rows";
        const TTL: Option<TtlPolicy<Self>> = Some(TtlPolicy {
            revision: 1,
            deadline: |row, _| (row.0 > 0).then_some(row.0),
        });
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[derive(Clone, Copy, Debug, PartialEq, FromBytes, IntoBytes, Immutable, GetType)]
    #[repr(transparent)]
    struct TtlWatchZeroMapRow(u64);

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    impl CollectionMeta for TtlWatchZeroMapRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "watch_ttl_zero_map_rows";
        const TTL: Option<TtlPolicy<Self>> = Some(TtlPolicy {
            revision: 1,
            deadline: |row, _| (row.0 > 0).then_some(row.0),
        });
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn all_six_families_return_a_typed_receiver() {
        let dir = tempdir().expect("temporary directory");
        let tree = TypedTree::open_hooked(
            dir.path().join("tree"),
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open typed tree");
        let mut rx: WatchReceiver<_> = tree.watch(Default::default());
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));

        let map = TypedMap::open_hooked(
            dir.path().join("map"),
            Config::balanced().iterable(true).build(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open typed map");
        let _: WatchReceiver<_> = map.watch(Default::default());

        let var_tree = VarTypedTree::open_hooked(
            dir.path().join("var-tree"),
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open var typed tree");
        let _: WatchReceiver<_> = var_tree.watch(Default::default());

        let var_map = VarTypedMap::open_hooked(
            dir.path().join("var-map"),
            Config::balanced().iterable(true).build(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open var typed map");
        let _: WatchReceiver<_> = var_map.watch(Default::default());

        let zero_tree: ZeroTree<
            [u8; 8],
            { std::mem::size_of::<WatchZeroRow>() },
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, true>,
        > = ZeroTree::open_hooked(
            dir.path().join("zero-tree"),
            Config::test(),
            WatchHook::<WatchZeroRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open zero tree");
        let _: WatchReceiver<_> = zero_tree.watch(Default::default());

        let zero_map: ZeroMap<
            [u8; 8],
            { std::mem::size_of::<WatchZeroRow>() },
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, true>,
        > = ZeroMap::open_hooked(
            dir.path().join("zero-map"),
            Config::balanced().iterable(true).build(),
            WatchHook::<WatchZeroRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open zero map");
        let _: WatchReceiver<_> = zero_map.watch(Default::default());
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[test]
    fn watch_map_without_iterable_is_refused_at_open() {
        let dir = tempdir().expect("temporary directory");
        let direct = TypedMap::open_hooked(
            dir.path(),
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        );
        assert!(
            matches!(direct, Err(crate::DbError::WatchRequiresIterable)),
            "watched maps need iterable indexes"
        );

        let managed_dir = tempdir().expect("temporary directory");
        let db = crate::armour::Db::open_test(managed_dir.path()).expect("open db");
        let managed = db.open_typed_map::<WatchRow, RapiraCodec, _>(
            Config::test(),
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
            &[],
        );
        assert!(matches!(
            managed,
            Err(crate::DbError::WatchRequiresIterable)
        ));
    }

    /// `Db::drop` routes through `shutdown`, so a database that is simply
    /// dropped must close its subscriptions too — otherwise a receiver held
    /// past the `Db` would block forever on a channel nothing can publish to.
    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[test]
    fn dropping_the_database_closes_its_subscriptions() {
        let dir = tempdir().expect("temporary directory");
        let db = crate::armour::Db::open_test(dir.path()).expect("open db");
        let tree = db
            .open_typed_tree::<WatchRow, RapiraCodec, _>(
                Config::test(),
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
                &[],
            )
            .expect("open watched tree");
        let mut receiver = tree.watch(WatchOptions::default());
        tree.put(&key(1), WatchRow(1)).expect("write");

        drop(tree);
        drop(db);

        // Drained first, then terminal.
        assert_eq!(
            receiver
                .try_recv()
                .expect("event queued before drop")
                .new()
                .expect("decode"),
            Some(WatchRow(1))
        );
        assert_eq!(receiver.try_recv(), Err(WatchError::Closed));
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[test]
    fn watch_close_keeps_receiver_live_but_shutdown_closes_it() {
        let dir = tempdir().expect("temporary directory");
        let mut db = crate::armour::Db::open_test(dir.path()).expect("open db");
        let tree = db
            .open_typed_tree::<WatchRow, RapiraCodec, _>(
                Config::test(),
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
                &[],
            )
            .expect("open watched tree");
        let mut receiver = tree.watch(WatchOptions::default());

        db.close().expect("non-terminal close");
        tree.put(&key(1), WatchRow(1)).expect("write after close");
        assert_eq!(
            receiver.recv_blocking().unwrap().new().unwrap(),
            Some(WatchRow(1))
        );

        db.shutdown().expect("shutdown");
        assert_eq!(receiver.try_recv(), Err(WatchError::Closed));
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[test]
    fn watch_shutdown_after_sweeper_drains_ttl_delete_before_closed() {
        struct ReleaseOnDrop(Option<Arc<std::sync::Barrier>>);
        impl Drop for ReleaseOnDrop {
            fn drop(&mut self) {
                if let Some(release) = self.0.take() {
                    release.wait();
                }
            }
        }
        struct JoinOnDrop<T>(Option<std::thread::JoinHandle<T>>);
        impl<T> Drop for JoinOnDrop<T> {
            fn drop(&mut self) {
                if let Some(join) = self.0.take() {
                    let _ = join.join();
                }
            }
        }

        let dir = tempdir().expect("temporary directory");
        let mut db = crate::armour::Db::open_test(dir.path()).expect("open db");
        let hook = WatchHook::<TtlWatchRow, NoHook, true>::with_old(NoHook);
        let channel = Arc::clone(&hook.channel);
        let tree = db
            .open_typed_tree::<TtlWatchRow, RapiraCodec, _>(Config::test(), hook, &[])
            .expect("open ttl tree");
        let mut receiver = tree.watch(WatchOptions::default());
        let ttl_key = key(13);
        tree.put(&ttl_key, TtlWatchRow(1))
            .expect("insert expiring row");
        let _ = receiver.try_recv().expect("insert watch event");

        let (published, release_publish) = channel.arm_before_publish();
        let mut sweep = JoinOnDrop(Some({
            let tree = Arc::clone(&tree);
            std::thread::spawn(move || tree.sweep_expired(u64::MAX, 1))
        }));
        published.wait();
        let (disable_waiting, release_disable_wait) = tree.arm_ttl_disable_wait();
        let mut shutdown = JoinOnDrop(Some(std::thread::spawn(move || db.shutdown())));
        disable_waiting.wait();
        let mut publish_release = ReleaseOnDrop(Some(release_publish));
        let mut disable_release = ReleaseOnDrop(Some(release_disable_wait));
        assert!(
            !shutdown.0.as_ref().expect("shutdown handle").is_finished(),
            "shutdown must still be blocked on the active TTL pass"
        );
        publish_release.0.take().expect("publish release").wait();
        disable_release.0.take().expect("disable release").wait();
        assert_eq!(
            sweep
                .0
                .take()
                .expect("sweeper handle")
                .join()
                .expect("sweeper thread")
                .unwrap()
                .deleted,
            1
        );
        shutdown
            .0
            .take()
            .expect("shutdown handle")
            .join()
            .expect("shutdown thread")
            .expect("shutdown");

        let deleted = receiver.try_recv().expect("ttl delete remains queued");
        assert_eq!(*deleted.key(), ttl_key);
        assert_eq!(deleted.kind(), WatchKind::Delete);
        assert_eq!(receiver.try_recv(), Err(WatchError::Closed));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn watched_typed_tree() -> (
        TypedTree<[u8; 8], WatchRow, RapiraCodec, WatchHook<WatchRow, NoHook, true>>,
        WatchReceiver<TypedTree<[u8; 8], WatchRow, RapiraCodec, WatchHook<WatchRow, NoHook, true>>>,
    ) {
        let dir = tempdir().expect("temporary directory");
        // Keep the directory alive through the collection's engine handle.
        let path = dir.keep();
        let mut config = Config::test();
        config.shard_count = 1;
        let tree = TypedTree::open_hooked(
            path,
            config,
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open watched typed tree");
        let rx = tree.watch(Default::default());
        (tree, rx)
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    type WatchedTypedTree =
        TypedTree<[u8; 8], WatchRow, RapiraCodec, WatchHook<WatchRow, NoHook, true>>;

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_seed(tree: &WatchedTypedTree, key: [u8; 8], value: u64) -> DbResult<()> {
        tree.put(&key, WatchRow(value))?;
        Ok(())
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_put(tree: &WatchedTypedTree) -> DbResult<()> {
        matrix_seed(tree, key(1), 1)
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_insert(tree: &WatchedTypedTree) -> DbResult<()> {
        tree.insert(&key(2), WatchRow(2))
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_delete(tree: &WatchedTypedTree) -> DbResult<()> {
        matrix_seed(tree, key(3), 3)?;
        tree.delete(&key(3))?;
        Ok(())
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_cas(tree: &WatchedTypedTree) -> DbResult<()> {
        matrix_seed(tree, key(4), 4)?;
        tree.cas(&key(4), &WatchRow(4), WatchRow(5))
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_compare_delete(tree: &WatchedTypedTree) -> DbResult<()> {
        matrix_seed(tree, key(5), 5)?;
        tree.compare_delete(&key(5), &WatchRow(5))
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_update(tree: &WatchedTypedTree) -> DbResult<()> {
        matrix_seed(tree, key(6), 6)?;
        tree.update(&key(6), |_| WatchRow(7))?;
        Ok(())
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_fetch_update(tree: &WatchedTypedTree) -> DbResult<()> {
        matrix_seed(tree, key(7), 7)?;
        tree.fetch_update(&key(7), |_| WatchRow(8))?;
        Ok(())
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_upsert_put_and_keep(tree: &WatchedTypedTree) -> DbResult<()> {
        tree.upsert(&key(8), |_| Upsert::Put(WatchRow(8)))?;
        tree.upsert(&key(8), |_| Upsert::Keep)?;
        Ok(())
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_atomic_shard_handle(tree: &WatchedTypedTree) -> DbResult<()> {
        tree.atomic(&key(9), |shard| {
            shard.put(&key(9), WatchRow(9))?;
            Ok(())
        })
    }
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn matrix_update_many(tree: &WatchedTypedTree) -> DbResult<()> {
        tree.update_many(vec![(key(10), 10), (key(11), 11)], |_, _, value| {
            if *value == 10 {
                Upsert::Put(WatchRow(*value))
            } else {
                Upsert::Keep
            }
        })?;
        Ok(())
    }

    /// Table-driven public mutation inventory. The production mutation that must
    /// make each row fail is its missing/incorrect `publish_watch` branch; a
    /// `Keep` is deliberately represented by the absence of a second event.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_mutation_matrix_reports_exact_with_old_kinds() {
        struct Case {
            name: &'static str,
            run: fn(&WatchedTypedTree) -> DbResult<()>,
            expected: &'static [WatchKind],
        }
        const INSERT: &[WatchKind] = &[WatchKind::Insert];
        const UPDATE: &[WatchKind] = &[WatchKind::Insert, WatchKind::Update];
        const DELETE: &[WatchKind] = &[WatchKind::Insert, WatchKind::Delete];
        const PUT_AND_KEEP: &[WatchKind] = &[WatchKind::Insert];
        const UPDATE_MANY: &[WatchKind] = &[WatchKind::Insert];
        let cases = [
            Case {
                name: "put",
                run: matrix_put,
                expected: INSERT,
            },
            Case {
                name: "insert",
                run: matrix_insert,
                expected: INSERT,
            },
            Case {
                name: "delete",
                run: matrix_delete,
                expected: DELETE,
            },
            Case {
                name: "cas",
                run: matrix_cas,
                expected: UPDATE,
            },
            Case {
                name: "compare_delete",
                run: matrix_compare_delete,
                expected: DELETE,
            },
            Case {
                name: "update",
                run: matrix_update,
                expected: UPDATE,
            },
            Case {
                name: "fetch_update",
                run: matrix_fetch_update,
                expected: UPDATE,
            },
            Case {
                name: "upsert put and Keep",
                run: matrix_upsert_put_and_keep,
                expected: PUT_AND_KEEP,
            },
            Case {
                name: "atomic shard handle",
                run: matrix_atomic_shard_handle,
                expected: INSERT,
            },
            Case {
                name: "update_many",
                run: matrix_update_many,
                expected: UPDATE_MANY,
            },
        ];

        for case in cases {
            let (tree, mut rx) = watched_typed_tree();
            (case.run)(&tree).expect(case.name);
            let mut kinds = Vec::new();
            while let Ok(event) = rx.try_recv() {
                kinds.push(event.kind());
            }
            assert_eq!(kinds, case.expected, "{}", case.name);
        }
    }

    /// Both delivery modes must agree on row visibility while differing only in
    /// retained old data: with-old distinguishes an overwrite as `Update`, and
    /// new-only deliberately collapses it to `Upsert`.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_all_six_family_modes_preserve_kind_contract() {
        macro_rules! assert_mode {
            ($receiver:expr, $expected:expr, $name:expr) => {{
                let event = $receiver.try_recv().expect($name);
                assert_eq!(event.kind(), $expected, $name);
                if $expected == WatchKind::Update {
                    assert!(event.old().expect("decode old").is_some(), $name);
                } else {
                    assert_eq!(event.old().expect("decode old"), None, $name);
                }
            }};
        }
        let dir = tempdir().unwrap();
        let old_typed = TypedTree::open_hooked(
            dir.path().join("old-typed"),
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .unwrap();
        let new_typed = TypedTree::open_hooked(
            dir.path().join("new-typed"),
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
        )
        .unwrap();
        let old_var = VarTypedTree::open_hooked(
            dir.path().join("old-var"),
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .unwrap();
        let new_var = VarTypedTree::open_hooked(
            dir.path().join("new-var"),
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
        )
        .unwrap();
        let old_map = TypedMap::open_hooked(
            dir.path().join("old-map"),
            watched_map_config(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .unwrap();
        let new_map = TypedMap::open_hooked(
            dir.path().join("new-map"),
            watched_map_config(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
        )
        .unwrap();
        let old_var_map = VarTypedMap::open_hooked(
            dir.path().join("old-var-map"),
            watched_map_config(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .unwrap();
        let new_var_map = VarTypedMap::open_hooked(
            dir.path().join("new-var-map"),
            watched_map_config(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
        )
        .unwrap();
        type ZOld = WatchHook<WatchZeroRow, NoHook, true>;
        type ZNew = WatchHook<WatchZeroRow, NoHook, false>;
        let old_zero = ZeroTree::<[u8; 8], 8, WatchZeroRow, ZOld>::open_hooked(
            dir.path().join("old-zero"),
            Config::test(),
            WatchHook::with_old(NoHook),
        )
        .unwrap();
        let new_zero = ZeroTree::<[u8; 8], 8, WatchZeroRow, ZNew>::open_hooked(
            dir.path().join("new-zero"),
            Config::test(),
            WatchHook::new_only(NoHook),
        )
        .unwrap();
        let old_zero_map = ZeroMap::<[u8; 8], 8, WatchZeroRow, ZOld>::open_hooked(
            dir.path().join("old-zero-map"),
            watched_map_config(),
            WatchHook::with_old(NoHook),
        )
        .unwrap();
        let new_zero_map = ZeroMap::<[u8; 8], 8, WatchZeroRow, ZNew>::open_hooked(
            dir.path().join("new-zero-map"),
            watched_map_config(),
            WatchHook::new_only(NoHook),
        )
        .unwrap();
        let mut r1 = old_typed.watch(Default::default());
        let mut r2 = new_typed.watch(Default::default());
        let mut r3 = old_var.watch(Default::default());
        let mut r4 = new_var.watch(Default::default());
        let mut r5 = old_map.watch(Default::default());
        let mut r6 = new_map.watch(Default::default());
        let mut r7 = old_var_map.watch(Default::default());
        let mut r8 = new_var_map.watch(Default::default());
        let mut r9 = old_zero.watch(Default::default());
        let mut r10 = new_zero.watch(Default::default());
        let mut r11 = old_zero_map.watch(Default::default());
        let mut r12 = new_zero_map.watch(Default::default());
        let k = key(91);
        old_typed.put(&k, WatchRow(1)).unwrap();
        new_typed.put(&k, WatchRow(1)).unwrap();
        old_var.put(&k, &WatchRow(1)).unwrap();
        new_var.put(&k, &WatchRow(1)).unwrap();
        old_map.put(&k, WatchRow(1)).unwrap();
        new_map.put(&k, WatchRow(1)).unwrap();
        old_var_map.put(&k, &WatchRow(1)).unwrap();
        new_var_map.put(&k, &WatchRow(1)).unwrap();
        old_zero.put(&k, &WatchZeroRow(1)).unwrap();
        new_zero.put(&k, &WatchZeroRow(1)).unwrap();
        old_zero_map.put(&k, &WatchZeroRow(1)).unwrap();
        new_zero_map.put(&k, &WatchZeroRow(1)).unwrap();
        assert_mode!(r1, WatchKind::Insert, "initial put");
        assert_mode!(r2, WatchKind::Upsert, "initial put");
        assert_mode!(r3, WatchKind::Insert, "initial put");
        assert_mode!(r4, WatchKind::Upsert, "initial put");
        assert_mode!(r5, WatchKind::Insert, "initial put");
        assert_mode!(r6, WatchKind::Upsert, "initial put");
        assert_mode!(r7, WatchKind::Insert, "initial put");
        assert_mode!(r8, WatchKind::Upsert, "initial put");
        assert_mode!(r9, WatchKind::Insert, "initial put");
        assert_mode!(r10, WatchKind::Upsert, "initial put");
        assert_mode!(r11, WatchKind::Insert, "initial put");
        assert_mode!(r12, WatchKind::Upsert, "initial put");
        old_typed.put(&k, WatchRow(2)).unwrap();
        new_typed.put(&k, WatchRow(2)).unwrap();
        old_var.put(&k, &WatchRow(2)).unwrap();
        new_var.put(&k, &WatchRow(2)).unwrap();
        old_map.put(&k, WatchRow(2)).unwrap();
        new_map.put(&k, WatchRow(2)).unwrap();
        old_var_map.put(&k, &WatchRow(2)).unwrap();
        new_var_map.put(&k, &WatchRow(2)).unwrap();
        old_zero.put(&k, &WatchZeroRow(2)).unwrap();
        new_zero.put(&k, &WatchZeroRow(2)).unwrap();
        old_zero_map.put(&k, &WatchZeroRow(2)).unwrap();
        new_zero_map.put(&k, &WatchZeroRow(2)).unwrap();
        assert_mode!(r1, WatchKind::Update, "with-old overwrite");
        assert_mode!(r3, WatchKind::Update, "with-old overwrite");
        assert_mode!(r5, WatchKind::Update, "with-old overwrite");
        assert_mode!(r7, WatchKind::Update, "with-old overwrite");
        assert_mode!(r9, WatchKind::Update, "with-old overwrite");
        assert_mode!(r11, WatchKind::Update, "with-old overwrite");
        assert_mode!(r2, WatchKind::Upsert, "new-only overwrite");
        assert_mode!(r4, WatchKind::Upsert, "new-only overwrite");
        assert_mode!(r6, WatchKind::Upsert, "new-only overwrite");
        assert_mode!(r8, WatchKind::Upsert, "new-only overwrite");
        assert_mode!(r10, WatchKind::Upsert, "new-only overwrite");
        assert_mode!(r12, WatchKind::Upsert, "new-only overwrite");
        old_typed.delete(&k).unwrap();
        new_typed.delete(&k).unwrap();
        old_var.delete(&k).unwrap();
        new_var.delete(&k).unwrap();
        old_map.delete(&k).unwrap();
        new_map.delete(&k).unwrap();
        old_var_map.delete(&k).unwrap();
        new_var_map.delete(&k).unwrap();
        old_zero.delete(&k).unwrap();
        new_zero.delete(&k).unwrap();
        old_zero_map.delete(&k).unwrap();
        new_zero_map.delete(&k).unwrap();
        macro_rules! assert_delete {
            ($receiver:expr, $old:expr) => {{
                let event = $receiver.try_recv().expect("delete event");
                assert_eq!(event.kind(), WatchKind::Delete);
                assert_eq!(event.new().expect("decode new"), None);
                assert_eq!(event.old().expect("decode old").is_some(), $old);
            }};
        }
        assert_delete!(r1, true);
        assert_delete!(r2, false);
        assert_delete!(r3, true);
        assert_delete!(r4, false);
        assert_delete!(r5, true);
        assert_delete!(r6, false);
        assert_delete!(r7, true);
        assert_delete!(r8, false);
        assert_delete!(r9, true);
        assert_delete!(r10, false);
        assert_delete!(r11, true);
        assert_delete!(r12, false);
    }

    // The production mutations these rows protect are the public facade calls
    // below.  In particular, a missing `publish_watch` in any success branch,
    // or publishing before a `KeyExists`/`CasMismatch`, changes an exact
    // receiver outcome.  `update_many` is intentionally NotApplicable to Map:
    // TypedMap, VarTypedMap, and ZeroMap expose no public `update_many`; Map
    // instead exposes the public `retain` operation covered by matrix_map.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_all_family_public_mutation_matrix() {
        macro_rules! payload {
            ($event:expr, $field:ident, $style:ident, none) => {
                assert_eq!($event.$field().expect("decode event"), None)
            };
            ($event:expr, $field:ident, $style:ident, $value:expr) => {
                assert_eq!(
                    $event.$field().expect("decode event"),
                    Some(row!($style, $value))
                )
            };
        }
        macro_rules! assert_event {
            ($rx:expr, $style:ident, old, $kind:expr, $key:expr, $old:tt, $new:tt, $name:expr) => {{
                let event = $rx.try_recv().expect($name);
                assert_eq!(event.kind(), $kind, $name);
                assert_eq!(event.key(), &$key, $name);
                payload!(event, old, $style, $old);
                payload!(event, new, $style, $new);
            }};
            ($rx:expr, $style:ident, new, $kind:expr, $key:expr, $old:tt, $new:tt, $name:expr) => {{
                let event = $rx.try_recv().expect($name);
                assert_eq!(
                    event.kind(),
                    if $kind == WatchKind::Delete {
                        WatchKind::Delete
                    } else {
                        WatchKind::Upsert
                    },
                    $name
                );
                assert_eq!(event.key(), &$key, $name);
                payload!(event, old, $style, none);
                payload!(event, new, $style, $new);
            }};
        }
        macro_rules! assert_one {
            ($rx:expr, $style:ident, $mode:ident, $kind:expr, $key:expr, $old:tt, $new:tt, $name:expr) => {{
                assert_event!($rx, $style, $mode, $kind, $key, $old, $new, $name);
                assert_eq!($rx.try_recv(), Err(WatchError::Empty), $name);
            }};
        }
        macro_rules! initial_kind {
            (old) => {
                WatchKind::Insert
            };
            (new) => {
                WatchKind::Upsert
            };
        }
        macro_rules! write {
            (owned, $target:expr, $key:expr, $value:expr) => {
                $target.put(&$key, WatchRow($value))
            };
            (borrowed, $target:expr, $key:expr, $value:expr) => {
                $target.put(&$key, &WatchRow($value))
            };
            (zero, $target:expr, $key:expr, $value:expr) => {
                $target.put(&$key, &WatchZeroRow($value))
            };
        }
        macro_rules! cas {
            (owned, $target:expr, $key:expr, $expected:expr, $value:expr) => {
                $target.cas(&$key, &WatchRow($expected), WatchRow($value))
            };
            (borrowed, $target:expr, $key:expr, $expected:expr, $value:expr) => {
                $target.cas(&$key, &WatchRow($expected), &WatchRow($value))
            };
            (zero, $target:expr, $key:expr, $expected:expr, $value:expr) => {
                $target.cas(&$key, &WatchZeroRow($expected), &WatchZeroRow($value))
            };
        }
        macro_rules! compare_delete {
            (owned, $target:expr, $key:expr, $expected:expr) => {
                $target.compare_delete(&$key, &WatchRow($expected))
            };
            (borrowed, $target:expr, $key:expr, $expected:expr) => {
                $target.compare_delete(&$key, &WatchRow($expected))
            };
            (zero, $target:expr, $key:expr, $expected:expr) => {
                $target.compare_delete(&$key, &WatchZeroRow($expected))
            };
        }
        macro_rules! row {
            (zero, $value:expr) => {
                WatchZeroRow($value)
            };
            ($style:ident, $value:expr) => {
                WatchRow($value)
            };
        }
        macro_rules! insert {
            (owned, $target:expr, $key:expr, $value:expr) => {
                $target.insert(&$key, WatchRow($value))
            };
            (borrowed, $target:expr, $key:expr, $value:expr) => {
                $target.insert(&$key, &WatchRow($value))
            };
            (zero, $target:expr, $key:expr, $value:expr) => {
                $target.insert(&$key, &WatchZeroRow($value))
            };
        }
        macro_rules! seed {
            ($style:ident, $target:expr, $rx:expr, $mode:ident, $key:expr, $value:expr) => {{
                write!($style, $target, $key, $value).expect("seed");
                assert_one!(
                    $rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    $key,
                    none,
                    $value,
                    "seed"
                );
            }};
        }
        macro_rules! matrix_tree {
            ($name:expr, $style:ident, $mode:ident, $open:expr) => {{
                let dir = tempdir().expect("temporary directory");
                let tree = $open(dir.path());
                let mut rx = tree.watch(Default::default());
                let insert_key = key(1);
                insert!($style, tree, insert_key, 1).expect("insert");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    insert_key,
                    none,
                    1,
                    "insert"
                );
                let absent_upsert_key = key(101);
                tree.upsert(&absent_upsert_key, |_| Upsert::Put(row!($style, 101)))
                    .expect("absent upsert put");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    absent_upsert_key,
                    none,
                    101,
                    "absent upsert put"
                );
                let occupied = key(2);
                seed!($style, tree, rx, $mode, occupied, 2);
                assert!(matches!(
                    insert!($style, tree, occupied, 3),
                    Err(crate::DbError::KeyExists)
                ));
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                let cas_key = key(3);
                seed!($style, tree, rx, $mode, cas_key, 3);
                cas!($style, tree, cas_key, 3, 4).expect("cas");
                assert_one!(rx, $style, $mode, WatchKind::Update, cas_key, 3, 4, "cas");
                let cas_mismatch = key(4);
                seed!($style, tree, rx, $mode, cas_mismatch, 4);
                assert!(matches!(
                    cas!($style, tree, cas_mismatch, 99, 5),
                    Err(crate::DbError::CasMismatch)
                ));
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                let delete_key = key(5);
                seed!($style, tree, rx, $mode, delete_key, 5);
                compare_delete!($style, tree, delete_key, 5).expect("compare delete");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Delete,
                    delete_key,
                    5,
                    none,
                    "compare delete"
                );
                let direct_delete_key = key(51);
                seed!($style, tree, rx, $mode, direct_delete_key, 51);
                tree.delete(&direct_delete_key).expect("direct delete");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Delete,
                    direct_delete_key,
                    51,
                    none,
                    "direct delete"
                );
                let delete_mismatch = key(6);
                seed!($style, tree, rx, $mode, delete_mismatch, 6);
                assert!(matches!(
                    compare_delete!($style, tree, delete_mismatch, 99),
                    Err(crate::DbError::CasMismatch)
                ));
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                let update_key = key(7);
                seed!($style, tree, rx, $mode, update_key, 7);
                tree.update(&update_key, |_| row!($style, 8))
                    .expect("update");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Update,
                    update_key,
                    7,
                    8,
                    "update"
                );
                let fetch_key = key(8);
                seed!($style, tree, rx, $mode, fetch_key, 8);
                tree.fetch_update(&fetch_key, |_| row!($style, 9))
                    .expect("fetch update");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Update,
                    fetch_key,
                    8,
                    9,
                    "fetch update"
                );
                let upsert_key = key(9);
                seed!($style, tree, rx, $mode, upsert_key, 9);
                tree.upsert(&upsert_key, |_| Upsert::Put(row!($style, 10)))
                    .expect("upsert put");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Update,
                    upsert_key,
                    9,
                    10,
                    "upsert put"
                );
                tree.upsert(&upsert_key, |_| Upsert::<_>::Keep)
                    .expect("upsert keep");
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                tree.atomic(&key(52), |_| Ok(())).expect("atomic closure");
                assert_eq!(
                    rx.try_recv(),
                    Err(WatchError::Empty),
                    "empty atomic is silent"
                );
                let atomic_key = key(10);
                tree.atomic(&atomic_key, |shard| {
                    write!($style, shard, atomic_key, 10).map(|_| ())
                })
                .expect("atomic shard put");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    atomic_key,
                    none,
                    10,
                    "atomic shard put"
                );
                tree.update_many(vec![(key(11), 11u64), (key(12), 12u64)], |_, _, value| {
                    Upsert::Put(row!($style, *value))
                })
                .expect("update many");
                assert_event!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    key(11),
                    none,
                    11,
                    "update many first"
                );
                assert_event!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    key(12),
                    none,
                    12,
                    "update many second"
                );
                assert_eq!(
                    rx.try_recv(),
                    Err(WatchError::Empty),
                    "update many drains exactly"
                );
                let _ = $name;
            }};
        }
        macro_rules! matrix_map {
            ($name:expr, $style:ident, $mode:ident, $open:expr) => {{
                let dir = tempdir().expect("temporary directory");
                let map = $open(dir.path());
                let mut rx = map.watch(Default::default());
                let insert_key = key(21);
                insert!($style, map, insert_key, 1).expect("insert");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    insert_key,
                    none,
                    1,
                    "insert"
                );
                let absent_upsert_key = key(121);
                map.upsert(&absent_upsert_key, |_| Upsert::Put(row!($style, 121)))
                    .expect("absent upsert put");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    absent_upsert_key,
                    none,
                    121,
                    "absent upsert put"
                );
                let occupied = key(22);
                seed!($style, map, rx, $mode, occupied, 2);
                assert!(matches!(
                    insert!($style, map, occupied, 3),
                    Err(crate::DbError::KeyExists)
                ));
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                let cas_key = key(23);
                seed!($style, map, rx, $mode, cas_key, 3);
                cas!($style, map, cas_key, 3, 4).expect("cas");
                assert_one!(rx, $style, $mode, WatchKind::Update, cas_key, 3, 4, "cas");
                let cas_mismatch = key(24);
                seed!($style, map, rx, $mode, cas_mismatch, 4);
                assert!(matches!(
                    cas!($style, map, cas_mismatch, 99, 5),
                    Err(crate::DbError::CasMismatch)
                ));
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                let delete_key = key(25);
                seed!($style, map, rx, $mode, delete_key, 5);
                compare_delete!($style, map, delete_key, 5).expect("compare delete");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Delete,
                    delete_key,
                    5,
                    none,
                    "compare delete"
                );
                let direct_delete_key = key(61);
                seed!($style, map, rx, $mode, direct_delete_key, 61);
                map.delete(&direct_delete_key).expect("direct delete");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Delete,
                    direct_delete_key,
                    61,
                    none,
                    "direct delete"
                );
                let delete_mismatch = key(26);
                seed!($style, map, rx, $mode, delete_mismatch, 6);
                assert!(matches!(
                    compare_delete!($style, map, delete_mismatch, 99),
                    Err(crate::DbError::CasMismatch)
                ));
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                let update_key = key(27);
                seed!($style, map, rx, $mode, update_key, 7);
                map.update(&update_key, |_| row!($style, 8))
                    .expect("update");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Update,
                    update_key,
                    7,
                    8,
                    "update"
                );
                let fetch_key = key(28);
                seed!($style, map, rx, $mode, fetch_key, 8);
                map.fetch_update(&fetch_key, |_| row!($style, 9))
                    .expect("fetch update");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Update,
                    fetch_key,
                    8,
                    9,
                    "fetch update"
                );
                let upsert_key = key(29);
                seed!($style, map, rx, $mode, upsert_key, 9);
                map.upsert(&upsert_key, |_| Upsert::Put(row!($style, 10)))
                    .expect("upsert put");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Update,
                    upsert_key,
                    9,
                    10,
                    "upsert put"
                );
                map.upsert(&upsert_key, |_| Upsert::<_>::Keep)
                    .expect("upsert keep");
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
                map.atomic(&key(62), |_| Ok(())).expect("atomic closure");
                assert_eq!(
                    rx.try_recv(),
                    Err(WatchError::Empty),
                    "empty atomic is silent"
                );
                let atomic_key = key(30);
                map.atomic(&atomic_key, |shard| {
                    write!($style, shard, atomic_key, 10).map(|_| ())
                })
                .expect("atomic shard put");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    initial_kind!($mode),
                    atomic_key,
                    none,
                    10,
                    "atomic shard put"
                );
                let remove = key(31);
                let keep = key(32);
                seed!($style, map, rx, $mode, remove, 31);
                seed!($style, map, rx, $mode, keep, 32);
                map.iter_view()
                    .expect("watched map is iterable")
                    .retain(|key, _| *key != remove)
                    .expect("retain");
                assert_one!(
                    rx,
                    $style,
                    $mode,
                    WatchKind::Delete,
                    remove,
                    31,
                    none,
                    "retain remove"
                );
                assert_eq!(
                    rx.try_recv(),
                    Err(WatchError::Empty),
                    "retain keep is silent"
                );
                let _ = $name;
            }};
        }

        matrix_tree!("typed tree old", owned, old, |path| TypedTree::open_hooked(
            path,
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook)
        )
        .unwrap());
        matrix_tree!("typed tree new", owned, new, |path| TypedTree::open_hooked(
            path,
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook)
        )
        .unwrap());
        matrix_tree!("var typed tree old", borrowed, old, |path| {
            VarTypedTree::open_hooked(
                path,
                Config::test(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
            )
            .unwrap()
        });
        matrix_tree!("var typed tree new", borrowed, new, |path| {
            VarTypedTree::open_hooked(
                path,
                Config::test(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
            )
            .unwrap()
        });
        matrix_tree!("zero tree old", zero, old, |path| ZeroTree::<
            [u8; 8],
            8,
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, true>,
        >::open_hooked(
            path,
            Config::test(),
            WatchHook::with_old(NoHook)
        )
        .unwrap());
        matrix_tree!("zero tree new", zero, new, |path| ZeroTree::<
            [u8; 8],
            8,
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, false>,
        >::open_hooked(
            path,
            Config::test(),
            WatchHook::new_only(NoHook)
        )
        .unwrap());
        matrix_map!("typed map old", owned, old, |path| TypedMap::open_hooked(
            path,
            watched_map_config(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook)
        )
        .unwrap());
        matrix_map!("typed map new", owned, new, |path| TypedMap::open_hooked(
            path,
            watched_map_config(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook)
        )
        .unwrap());
        matrix_map!("var typed map old", borrowed, old, |path| {
            VarTypedMap::open_hooked(
                path,
                watched_map_config(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
            )
            .unwrap()
        });
        matrix_map!("var typed map new", borrowed, new, |path| {
            VarTypedMap::open_hooked(
                path,
                watched_map_config(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
            )
            .unwrap()
        });
        matrix_map!("zero map old", zero, old, |path| ZeroMap::<
            [u8; 8],
            8,
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, true>,
        >::open_hooked(
            path,
            watched_map_config(),
            WatchHook::with_old(NoHook)
        )
        .unwrap());
        matrix_map!("zero map new", zero, new, |path| ZeroMap::<
            [u8; 8],
            8,
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, false>,
        >::open_hooked(
            path,
            watched_map_config(),
            WatchHook::new_only(NoHook)
        )
        .unwrap());
    }

    /// Each public Tree facade must surface a real later `update_many` error
    /// after publishing exactly its applied prefix. The scoped fault guards live
    /// below the public facade and restore their thread-local state on drop.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_public_update_many_error_keeps_exact_applied_prefix() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        struct CountWake(Arc<AtomicUsize>);
        impl std::task::Wake for CountWake {
            fn wake(self: Arc<Self>) {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }

        macro_rules! row {
            (zero, $value:expr) => {
                WatchZeroRow($value)
            };
            ($style:ident, $value:expr) => {
                WatchRow($value)
            };
        }
        macro_rules! assert_prefix {
            ($event:expr, $style:ident, old, $key:expr, $value:expr) => {{
                let event = $event;
                assert_eq!(event.key(), &$key);
                assert_eq!(event.kind(), WatchKind::Insert);
                assert_eq!(event.old().expect("decode old"), None);
                assert_eq!(event.new().expect("decode new"), Some(row!($style, $value)));
            }};
            ($event:expr, $style:ident, new, $key:expr, $value:expr) => {{
                let event = $event;
                assert_eq!(event.key(), &$key);
                assert_eq!(event.kind(), WatchKind::Upsert);
                assert_eq!(event.old().expect("decode old"), None);
                assert_eq!(event.new().expect("decode new"), Some(row!($style, $value)));
            }};
        }
        macro_rules! run {
            ($style:ident, $mode:ident, $open:expr, $fault:expr) => {{
                let dir = tempdir().expect("temporary directory");
                let tree = $open(dir.path());
                let mut rx = tree.watch(Default::default());
                let first = key(71);
                let second = key(72);
                let wakes = Arc::new(AtomicUsize::new(0));
                let waker = Waker::from(Arc::new(CountWake(Arc::clone(&wakes))));
                let mut pending = Box::pin(rx.recv());
                assert!(matches!(
                    pending.as_mut().poll(&mut Context::from_waker(&waker)),
                    Poll::Pending
                ));
                let _fault = $fault;
                let err = tree
                    .update_many(vec![(first, 1u64), (second, 2u64)], |_, _, value| {
                        Upsert::Put(row!($style, *value))
                    })
                    .expect_err("second public update_many mutation fails");
                assert!(matches!(
                    err,
                    crate::DbError::Internal("test: update_many injected failure")
                ));
                assert_eq!(
                    wakes.load(Ordering::SeqCst),
                    1,
                    "applied prefix wakes receiver exactly once"
                );
                let event = match pending.as_mut().poll(&mut Context::from_waker(&waker)) {
                    Poll::Ready(Ok(event)) => event,
                    other => panic!("woken receiver must yield applied prefix, got {other:?}"),
                };
                assert_prefix!(event, $style, $mode, first, 1);
                drop(pending);
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
            }};
        }
        run!(
            owned,
            old,
            |path| TypedTree::open_hooked(
                path,
                Config::test(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook)
            )
            .unwrap(),
            crate::typed_tree::UpdateManyFailAfter::after(1)
        );
        run!(
            owned,
            new,
            |path| TypedTree::open_hooked(
                path,
                Config::test(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, false>::new_only(NoHook)
            )
            .unwrap(),
            crate::typed_tree::UpdateManyFailAfter::after(1)
        );
        run!(
            borrowed,
            old,
            |path| VarTypedTree::open_hooked(
                path,
                Config::test(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook)
            )
            .unwrap(),
            crate::var_tree::UpdateManyFailAfter::after(1)
        );
        run!(
            borrowed,
            new,
            |path| VarTypedTree::open_hooked(
                path,
                Config::test(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, false>::new_only(NoHook)
            )
            .unwrap(),
            crate::var_tree::UpdateManyFailAfter::after(1)
        );
        run!(
            zero,
            old,
            |path| {
                ZeroTree::<[u8; 8], 8, WatchZeroRow, WatchHook<WatchZeroRow, NoHook, true>>::open_hooked(path, Config::test(), WatchHook::with_old(NoHook)).unwrap()
            },
            crate::const_tree::UpdateManyFailAfter::after(1)
        );
        run!(
            zero,
            new,
            |path| {
                ZeroTree::<[u8; 8], 8, WatchZeroRow, WatchHook<WatchZeroRow, NoHook, false>>::open_hooked(path, Config::test(), WatchHook::new_only(NoHook)).unwrap()
            },
            crate::const_tree::UpdateManyFailAfter::after(1)
        );
    }

    /// Recovery and migration rebuild internal state, not application writes.
    /// Enabling hook delivery in either path would make this test observe an
    /// event before a caller performs a post-open mutation.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_migration_and_recovery_emit_nothing() {
        let dir = tempdir().expect("temporary directory");
        let path = dir.path().join("recovered");
        let key = key(42);
        {
            let tree = TypedTree::<[u8; 8], WatchRow, RapiraCodec>::open(
                &path,
                Config::test(),
                RapiraCodec,
            )
            .expect("seed collection");
            tree.put(&key, WatchRow(1)).expect("seed row");
            tree.close().expect("close seeded collection");
        }

        let tree = TypedTree::open_hooked(
            &path,
            Config::test(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("recover watched collection");
        let mut rx = tree.watch(Default::default());
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));

        tree.migrate(|_, _| crate::MigrateAction::Update(WatchRow(2)))
            .expect("hook-free migration");
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));
        assert_eq!(tree.get(&key).map(|row| row.0), Some(2));
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "replication"
    ))]
    #[test]
    fn watch_follower_apply_emits_nothing_but_facade_writes_do() {
        let leader_dir = tempdir().unwrap();
        let follower_dir = tempdir().unwrap();
        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        drop(listener);
        let leader =
            Arc::new(VarTypedTree::open(leader_dir.path(), Config::test(), RapiraCodec).unwrap());
        let follower = Arc::new(
            VarTypedTree::open_hooked(
                follower_dir.path(),
                Config::test(),
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
            )
            .unwrap(),
        );
        let mut rx = follower.watch(Default::default());
        let local_key = key(70);
        follower.put(&local_key, &WatchRow(70)).unwrap();
        assert_eq!(
            rx.try_recv().unwrap().key(),
            &local_key,
            "watch hook is live"
        );
        follower.delete(&local_key).unwrap();
        let _ = rx.try_recv().unwrap();

        let leader_signal = crate::ShutdownSignal::new();
        let follower_signal = crate::ShutdownSignal::new();
        let _server = leader
            .test_inner_arc()
            .start_replication_server(addr, leader_signal.clone())
            .unwrap();
        let follower_raw = follower.test_inner_arc();
        let registry = Arc::new(crate::replication::ReplicationRegistry::new(
            follower_raw.as_replication_target(),
        ));
        let _client = follower_raw
            .start_replication_client(addr, registry, follower_signal.clone())
            .unwrap();
        let replicated_key = key(71);
        leader.put(&replicated_key, &WatchRow(71)).unwrap();
        let start = Instant::now();
        while follower.get(&replicated_key).is_none() && start.elapsed() < Duration::from_secs(10) {
            std::thread::sleep(Duration::from_millis(10));
        }
        assert_eq!(follower.get(&replicated_key).map(|row| row.0), Some(71));
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));
        follower_signal.shutdown();
        leader_signal.shutdown();
    }

    /// The safe consumer recipe is subscribe, scan, then rescan on `Lagged`.
    /// `iter_view` is deliberately used here: watched Maps require it and its
    /// weak iterator is never treated as a snapshot.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_then_scan_and_post_lagged_rebuild_converge() {
        let dir = tempdir().expect("temporary directory");
        let map = TypedMap::open_hooked(
            dir.path(),
            watched_map_config(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open watched iterable map");
        let mut rx = map.watch(WatchOptions {
            capacity: NonZeroUsize::new(1).unwrap(),
        });
        let first = key(1);
        let second = key(2);
        map.put(&first, WatchRow(10)).unwrap();
        map.put(&second, WatchRow(20)).unwrap();

        let scan = || {
            map.iter_view()
                .expect("watched map is iterable")
                .iter()
                .map(|(key, row)| (key, row.0))
                .collect::<std::collections::BTreeMap<_, _>>()
        };
        let initial_mirror = scan();
        assert_eq!(
            initial_mirror.len(),
            2,
            "watch-first scan sees both seeded rows"
        );
        map.put(&first, WatchRow(11)).unwrap();
        map.delete(&second).unwrap();
        assert!(matches!(rx.try_recv(), Err(WatchError::Lagged(_))));

        // The operational rule: an event is a pointer, not a snapshot. Only its
        // key is used — the handler re-reads the row and applies what it finds.
        // Applying `event.new()` instead would converge here by accident (the
        // writes have stopped) while being wrong against a live writer.
        let mut mirror = scan();
        let mut applied = 0usize;
        while let Ok(event) = rx.try_recv() {
            match map.get(event.key()) {
                Some(row) => {
                    mirror.insert(*event.key(), row.0);
                }
                None => {
                    mirror.remove(event.key());
                }
            }
            applied += 1;
        }
        assert!(applied > 0, "the rebuild must consume the post-gap events");
        assert_eq!(mirror, scan());
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_then_scan_observes_a_writer_that_waited_for_the_shard_lock() {
        let dir = tempdir().expect("temporary directory");
        let mut config = Config::test();
        config.shard_count = 1;
        let hook = WatchHook::<WatchRow, NoHook, false>::new_only(NoHook);
        let (writer_at_gate, release_writer) = hook.channel.arm_before_ctx_lock();
        let tree = Arc::new(
            TypedTree::open_hooked(dir.path(), config, RapiraCodec, hook)
                .expect("open watched tree"),
        );
        let key = key(93);
        let (shard_locked, release_shard) = (
            Arc::new(std::sync::Barrier::new(2)),
            Arc::new(std::sync::Barrier::new(2)),
        );
        let blocker = {
            let tree = Arc::clone(&tree);
            let shard_locked = Arc::clone(&shard_locked);
            let release_shard = Arc::clone(&release_shard);
            std::thread::spawn(move || {
                tree.atomic(&key, |_| {
                    shard_locked.wait();
                    release_shard.wait();
                    Ok(())
                })
                .expect("release blocking atomic")
            })
        };
        shard_locked.wait();

        let writer = {
            let tree = Arc::clone(&tree);
            std::thread::spawn(move || tree.put(&key, WatchRow(1)).map(|_| ()))
        };
        writer_at_gate.wait();

        let mut receiver = tree.watch(Default::default());
        assert!(
            tree.get(&key).is_none(),
            "the watch-first scan runs before the waiting writer commits"
        );

        release_writer.wait();
        release_shard.wait();
        blocker.join().expect("blocking atomic thread");
        writer
            .join()
            .expect("writer thread")
            .expect("writer result");

        let event = receiver
            .try_recv()
            .expect("watch-first scan must not lose the waiting write");
        assert_eq!(event.kind(), WatchKind::Upsert);
        assert_eq!(event.key(), &key);
        assert_eq!(event.new().expect("decode new"), Some(WatchRow(1)));
    }

    /// The window the participation guard exists for: a producer samples the
    /// gate, finds no receiver, and only *then* makes its row visible. A
    /// receiver registering in between is in neither half of the recipe — the
    /// producer will publish nothing to it, and its catch-up scan runs before
    /// the row exists. `watch()` must therefore not return until that producer
    /// has finished.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_waits_out_a_producer_that_already_sampled_an_empty_gate() {
        use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};

        let dir = tempdir().expect("temporary directory");
        let mut config = Config::test();
        config.shard_count = 1;
        let hook = WatchHook::<WatchRow, NoHook, false>::new_only(NoHook);
        let (writer_past_gate, release_writer) = hook.channel.arm_after_watch_gate();
        let (receiver_linked, release_receiver) = hook.channel.arm_after_subscribe_link();
        let tree = Arc::new(
            TypedTree::open_hooked(dir.path(), config, RapiraCodec, hook)
                .expect("open watched tree"),
        );
        let key = key(94);

        let writer = {
            let tree = Arc::clone(&tree);
            std::thread::spawn(move || tree.put(&key, WatchRow(7)).map(|_| ()))
        };
        // Parked past its gate sample (no receiver linked), before its row exists.
        writer_past_gate.wait();

        let scanned = Arc::new(AtomicBool::new(false));
        let subscriber = {
            let tree = Arc::clone(&tree);
            let scanned = Arc::clone(&scanned);
            std::thread::spawn(move || {
                let receiver = tree.watch(Default::default());
                let seen = tree.get(&key).is_some();
                scanned.store(true, AtomicOrdering::Release);
                (receiver, seen)
            })
        };
        // Linked and counted, but not yet draining.
        receiver_linked.wait();
        release_receiver.wait();

        // Asserting a *non*-event: the only way to observe that `watch()` has
        // not returned is to give it time to. The wait is one-sided — a correct
        // build spends it, a regressed one fails immediately.
        std::thread::sleep(Duration::from_millis(200));
        assert!(
            !scanned.load(AtomicOrdering::Acquire),
            "watch() returned while a producer that sampled an empty gate was still in flight"
        );

        release_writer.wait();
        writer
            .join()
            .expect("writer thread")
            .expect("writer result");
        let (_receiver, seen) = subscriber.join().expect("subscriber thread");
        assert!(
            seen,
            "the catch-up scan must cover the write the producer decided not to publish"
        );
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn watched_map_config() -> Config {
        let mut config = Config::test();
        config.iterable = true;
        config
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    fn watched_typed_map() -> (
        TypedMap<[u8; 8], WatchRow, RapiraCodec, WatchHook<WatchRow, NoHook, true>>,
        WatchReceiver<TypedMap<[u8; 8], WatchRow, RapiraCodec, WatchHook<WatchRow, NoHook, true>>>,
    ) {
        let dir = tempdir().expect("temporary directory");
        let path = dir.keep();
        let map = TypedMap::open_hooked(
            path,
            Config::balanced().iterable(true).build(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open watched typed map");
        let rx = map.watch(Default::default());
        (map, rx)
    }

    /// Two halves of one contract.
    ///
    /// The **new** payload is the bytes the successful write stored, so
    /// publication never calls the codec for it again — a write gets exactly
    /// one encode, and the event still carries the right value.
    ///
    /// The **old** payload is encoded from the value the mutation swapped out,
    /// which is still a live `T` in RAM; reading it back from its `DiskLoc`
    /// would put a `pread` inside the shard lock. That encode can fail, and
    /// when it does the event degrades — it must keep reporting `Update`, since
    /// the kind is a fact about the mutation, not about what could be
    /// materialized. Reporting `Insert` over a live row is the failure this
    /// pins.
    #[cfg(all(feature = "rapira-codec", feature = "typed-tree"))]
    #[test]
    fn typed_watch_new_bytes_are_stored_bytes_and_a_failed_old_encode_degrades() {
        let dir = tempfile::tempdir().expect("temporary directory");
        let key = 7u64.to_be_bytes();

        let tree = crate::TypedTree::open_hooked(
            dir.path().join("tree"),
            crate::Config::test(),
            EncodeOnceCodec,
            WatchHook::<EncodeOnceRow, crate::NoHook, true>::with_old(crate::NoHook),
        )
        .expect("open typed tree");
        let mut tree_rx = tree.watch(Default::default());

        {
            let _encode = EncodeOnceScope::allow_one();
            tree.put(&key, EncodeOnceRow(1)).expect("tree insert");
        }
        let tree_insert = tree_rx.try_recv().expect("tree insert event");
        assert_eq!(tree_insert.kind(), WatchKind::Insert);
        assert_eq!(tree_insert.key(), &key);
        assert_eq!(tree_insert.old().expect("tree insert old"), None);
        assert_eq!(
            tree_insert.new().expect("tree insert new"),
            Some(EncodeOnceRow(1))
        );

        // One allowance: the write spends it, so the old-value encode fails.
        {
            let _encode = EncodeOnceScope::allow_one();
            tree.put(&key, EncodeOnceRow(2)).expect("tree update");
        }
        let tree_degraded = tree_rx.try_recv().expect("tree degraded update event");
        assert_eq!(
            tree_degraded.kind(),
            WatchKind::Update,
            "a failed old-value encode must not turn an update into an insert"
        );
        assert_eq!(tree_degraded.key(), &key);
        assert_eq!(tree_degraded.old().expect("tree degraded old"), None);
        assert_eq!(
            tree_degraded.new().expect("tree degraded new"),
            Some(EncodeOnceRow(2))
        );

        // Two allowances: the write and the old-value encode both succeed.
        {
            let _encode = EncodeOnceScope::allow(2);
            tree.put(&key, EncodeOnceRow(3))
                .expect("tree second update");
        }
        let tree_update = tree_rx.try_recv().expect("tree update event");
        assert_eq!(tree_update.kind(), WatchKind::Update);
        assert_eq!(tree_update.key(), &key);
        assert_eq!(
            tree_update.old().expect("tree update old"),
            Some(EncodeOnceRow(2))
        );
        assert_eq!(
            tree_update.new().expect("tree update new"),
            Some(EncodeOnceRow(3))
        );

        let map = crate::TypedMap::open_hooked(
            dir.path().join("map"),
            crate::Config::balanced().iterable(true).build(),
            EncodeOnceCodec,
            WatchHook::<EncodeOnceRow, crate::NoHook, true>::with_old(crate::NoHook),
        )
        .expect("open typed map");
        let mut map_rx = map.watch(Default::default());

        {
            let _encode = EncodeOnceScope::allow_one();
            map.put(&key, EncodeOnceRow(3)).expect("map insert");
        }
        let map_insert = map_rx.try_recv().expect("map insert event");
        assert_eq!(map_insert.kind(), WatchKind::Insert);
        assert_eq!(map_insert.key(), &key);
        assert_eq!(map_insert.old().expect("map insert old"), None);
        assert_eq!(
            map_insert.new().expect("map insert new"),
            Some(EncodeOnceRow(3))
        );

        {
            let _encode = EncodeOnceScope::allow_one();
            map.put(&key, EncodeOnceRow(4))
                .expect("map degraded update");
        }
        let map_degraded = map_rx.try_recv().expect("map degraded update event");
        assert_eq!(
            map_degraded.kind(),
            WatchKind::Update,
            "a failed old-value encode must not turn an update into an insert"
        );
        assert_eq!(map_degraded.key(), &key);
        assert_eq!(map_degraded.old().expect("map degraded old"), None);
        assert_eq!(
            map_degraded.new().expect("map degraded new"),
            Some(EncodeOnceRow(4))
        );

        {
            let _encode = EncodeOnceScope::allow(2);
            map.put(&key, EncodeOnceRow(5)).expect("map update");
        }
        let map_update = map_rx.try_recv().expect("map update event");
        assert_eq!(map_update.kind(), WatchKind::Update);
        assert_eq!(map_update.key(), &key);
        assert_eq!(
            map_update.old().expect("map update old"),
            Some(EncodeOnceRow(4))
        );
        assert_eq!(
            map_update.new().expect("map update new"),
            Some(EncodeOnceRow(5))
        );
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_zero_tree_and_map_deliver_put_update_and_delete() {
        let dir = tempdir().expect("temporary directory");
        let tree: ZeroTree<
            [u8; 8],
            { std::mem::size_of::<WatchZeroRow>() },
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, true>,
        > = ZeroTree::open_hooked(
            dir.path().join("tree"),
            Config::test(),
            WatchHook::with_old(NoHook),
        )
        .expect("open watched zero tree");
        let map: ZeroMap<
            [u8; 8],
            { std::mem::size_of::<WatchZeroRow>() },
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, true>,
        > = ZeroMap::open_hooked(
            dir.path().join("map"),
            watched_map_config(),
            WatchHook::with_old(NoHook),
        )
        .expect("open watched zero map");
        let key = key(1);
        let mut tree_rx = tree.watch(Default::default());
        let mut map_rx = map.watch(Default::default());

        for write in [WatchZeroRow(1), WatchZeroRow(2)] {
            tree.put(&key, &write).expect("tree write");
            map.put(&key, &write).expect("map write");
        }
        tree.delete(&key).expect("tree delete");
        map.delete(&key).expect("map delete");

        macro_rules! assert_events {
            ($rx:expr) => {{
                let rx = &mut $rx;
                let inserted = rx.try_recv().expect("insert event");
                assert_eq!(inserted.kind(), WatchKind::Insert);
                assert_eq!(inserted.old().expect("decode old"), None);
                assert_eq!(inserted.new().expect("decode new"), Some(WatchZeroRow(1)));
                let updated = rx.try_recv().expect("update event");
                assert_eq!(updated.kind(), WatchKind::Update);
                assert_eq!(updated.old().expect("decode old"), Some(WatchZeroRow(1)));
                assert_eq!(updated.new().expect("decode new"), Some(WatchZeroRow(2)));
                let deleted = rx.try_recv().expect("delete event");
                assert_eq!(deleted.kind(), WatchKind::Delete);
                assert_eq!(deleted.old().expect("decode old"), Some(WatchZeroRow(2)));
                assert_eq!(deleted.new().expect("decode new"), None);
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
            }};
        }
        assert_events!(tree_rx);
        assert_events!(map_rx);
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_zero_new_only_reports_upsert_and_delete_without_old_payload() {
        let dir = tempdir().expect("temporary directory");
        let tree: ZeroTree<
            [u8; 8],
            { std::mem::size_of::<WatchZeroRow>() },
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, false>,
        > = ZeroTree::open_hooked(
            dir.path().join("tree"),
            Config::test(),
            WatchHook::new_only(NoHook),
        )
        .expect("open new-only zero tree");
        let map: ZeroMap<
            [u8; 8],
            { std::mem::size_of::<WatchZeroRow>() },
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, false>,
        > = ZeroMap::open_hooked(
            dir.path().join("map"),
            watched_map_config(),
            WatchHook::new_only(NoHook),
        )
        .expect("open new-only zero map");
        let key = key(1);
        let mut tree_rx = tree.watch(Default::default());
        let mut map_rx = map.watch(Default::default());
        for write in [WatchZeroRow(1), WatchZeroRow(2)] {
            tree.put(&key, &write).expect("tree write");
            map.put(&key, &write).expect("map write");
        }
        tree.delete(&key).expect("tree delete");
        map.delete(&key).expect("map delete");

        macro_rules! assert_events {
            ($rx:expr) => {{
                let rx = &mut $rx;
                let inserted = rx.try_recv().expect("insert event");
                assert_eq!(inserted.kind(), WatchKind::Upsert);
                assert_eq!(inserted.old().expect("decode old"), None);
                let updated = rx.try_recv().expect("update event");
                assert_eq!(updated.kind(), WatchKind::Upsert);
                assert_eq!(updated.old().expect("decode old"), None);
                let deleted = rx.try_recv().expect("delete event");
                assert_eq!(deleted.kind(), WatchKind::Delete);
                assert_eq!(deleted.old().expect("decode old"), None);
                assert_eq!(rx.try_recv(), Err(WatchError::Empty));
            }};
        }
        assert_events!(tree_rx);
        assert_events!(map_rx);
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_zero_tree_update_many_emits_only_applied_put_and_delete() {
        let dir = tempdir().expect("temporary directory");
        let mut config = Config::test();
        config.shard_count = 1;
        let tree: ZeroTree<
            [u8; 8],
            { std::mem::size_of::<WatchZeroRow>() },
            WatchZeroRow,
            WatchHook<WatchZeroRow, NoHook, true>,
        > = ZeroTree::open_hooked(dir.path(), config, WatchHook::with_old(NoHook))
            .expect("open watched zero tree");
        let put_key = key(1);
        let delete_key = key(2);
        let keep_key = key(3);
        let mut rx = tree.watch(Default::default());
        tree.put(&delete_key, &WatchZeroRow(2))
            .expect("seed delete");
        rx.try_recv().expect("seed event");

        tree.update_many(
            vec![(put_key, 1u8), (delete_key, 2), (keep_key, 3)],
            |_, _, action| match action {
                1 => Upsert::Put(WatchZeroRow(1)),
                2 => Upsert::Delete,
                _ => Upsert::Keep,
            },
        )
        .expect("batch applies");

        assert_eq!(rx.try_recv().expect("put event").key(), &put_key);
        let deleted = rx.try_recv().expect("delete event");
        assert_eq!(deleted.key(), &delete_key);
        assert_eq!(deleted.kind(), WatchKind::Delete);
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_zero_ttl_sweep_deletes_tree_and_map_rows() {
        type TreeHook = crate::TtlHook<TtlWatchZeroRow, WatchHook<TtlWatchZeroRow, NoHook, true>>;
        type MapHook =
            crate::TtlHook<TtlWatchZeroMapRow, WatchHook<TtlWatchZeroMapRow, NoHook, true>>;
        let dir = tempdir().expect("temporary directory");
        let tree = ZeroTree::<
            [u8; 8],
            { std::mem::size_of::<TtlWatchZeroRow>() },
            TtlWatchZeroRow,
            TreeHook,
        >::open_hooked(
            dir.path().join("tree"),
            Config::test(),
            crate::TtlHook::new(WatchHook::with_old(NoHook)),
        )
        .expect("open ttl zero tree");
        let map = ZeroMap::<
            [u8; 8],
            { std::mem::size_of::<TtlWatchZeroMapRow>() },
            TtlWatchZeroMapRow,
            MapHook,
        >::open_hooked(
            dir.path().join("map"),
            watched_map_config(),
            crate::TtlHook::new(WatchHook::with_old(NoHook)),
        )
        .expect("open ttl zero map");
        tree.as_inner()
            .build_ttl_index()
            .expect("build tree ttl index");
        map.as_inner()
            .build_ttl_index()
            .expect("build map ttl index");
        let expired = key(1);
        let mut tree_rx = tree.watch(Default::default());
        let mut map_rx = map.watch(Default::default());
        tree.put(&expired, &TtlWatchZeroRow(1)).expect("tree put");
        map.put(&expired, &TtlWatchZeroMapRow(1)).expect("map put");
        tree_rx.try_recv().expect("tree seed event");
        map_rx.try_recv().expect("map seed event");

        assert_eq!(tree.sweep_expired(2, 1).expect("tree sweep").deleted, 1);
        assert_eq!(map.sweep_expired(2, 1).expect("map sweep").deleted, 1);
        assert_eq!(
            tree_rx.try_recv().expect("tree expiry").kind(),
            WatchKind::Delete
        );
        assert_eq!(
            map_rx.try_recv().expect("map expiry").kind(),
            WatchKind::Delete
        );
        assert!(tree.get(&expired).is_none());
        assert!(map.get(&expired).is_none());
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[test]
    fn watch_zero_transactions_deliver_only_after_commit() {
        let dir = tempdir().expect("temporary directory");
        let db = Arc::new(crate::armour::Db::open_test(dir.path()).expect("open db"));
        let tree = db
            .open_zero_tree::<WatchZeroRow, { std::mem::size_of::<WatchZeroRow>() }, _>(
                Config::test(),
                WatchHook::with_old(NoHook),
                &[],
            )
            .expect("open managed zero tree");
        let map = db
            .open_zero_map::<WatchZeroMapRow, { std::mem::size_of::<WatchZeroMapRow>() }, _>(
                watched_map_config(),
                WatchHook::with_old(NoHook),
                &[],
            )
            .expect("open managed zero map");
        let tree_key = key(1);
        let map_key = key(2);
        let mut tree_rx = tree.watch(Default::default());
        let mut map_rx = map.watch(Default::default());

        db.tx()
            .put(&tree, &tree_key, &WatchZeroRow(1))
            .expect("plan zero tree write")
            .put(&map, &map_key, &WatchZeroMapRow(2))
            .expect("plan zero map write")
            .commit()
            .expect("commit zero transaction");

        assert_eq!(tree_rx.try_recv().expect("tree event").key(), &tree_key);
        assert_eq!(map_rx.try_recv().expect("map event").key(), &map_key);
        assert_eq!(tree_rx.try_recv(), Err(WatchError::Empty));
        assert_eq!(map_rx.try_recv(), Err(WatchError::Empty));

        let rejected_key = key(3);
        let err = db
            .tx()
            .insert(&tree, &rejected_key, &WatchZeroRow(3))
            .expect("plan first zero insert")
            .insert(&tree, &rejected_key, &WatchZeroRow(4))
            .expect("plan duplicate zero insert")
            .commit()
            .expect_err("duplicate zero insert fails before the log");
        assert!(matches!(err, crate::DbError::KeyExists));
        assert_eq!(tree_rx.try_recv(), Err(WatchError::Empty));
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[test]
    fn watch_transaction_delivers_each_applied_row_only_after_all_unlocks() {
        use std::sync::atomic::{AtomicBool, Ordering};

        struct TouchRightOnWake {
            right: Arc<
                TypedTree<
                    [u8; 8],
                    WatchRightRow,
                    RapiraCodec,
                    crate::TtlHook<WatchRightRow, WatchHook<WatchRightRow, NoHook, true>>,
                >,
            >,
            key: [u8; 8],
            saw_right_unlocked: Arc<AtomicBool>,
        }

        impl std::task::Wake for TouchRightOnWake {
            fn wake(self: Arc<Self>) {
                self.saw_right_unlocked.store(
                    self.right.test_try_lock_writer_shard(&self.key),
                    Ordering::Release,
                );
            }
        }

        let dir = tempdir().expect("temporary directory");
        let db = Arc::new(crate::armour::Db::open_test(dir.path()).expect("open db"));
        let config = Config::test();
        let left = db
            .open_typed_tree::<WatchRow, RapiraCodec, _>(
                config.clone(),
                WatchHook::with_old(NoHook),
                &[],
            )
            .expect("open left participant");
        let right = db
            .open_typed_tree::<WatchRightRow, RapiraCodec, _>(
                config,
                WatchHook::with_old(NoHook),
                &[],
            )
            .expect("open right participant");
        let left_key = key(1);
        let right_key = key(2);
        let mut left_rx = left.watch(Default::default());
        let mut right_rx = right.watch(Default::default());
        let saw_right_unlocked = Arc::new(AtomicBool::new(false));
        let waker = Waker::from(Arc::new(TouchRightOnWake {
            right: Arc::clone(&right),
            key: right_key,
            saw_right_unlocked: Arc::clone(&saw_right_unlocked),
        }));
        let mut recv = Box::pin(left_rx.recv());
        assert!(matches!(
            Pin::as_mut(&mut recv).poll(&mut Context::from_waker(&waker)),
            Poll::Pending
        ));

        db.tx()
            .put(&left, &left_key, &WatchRow(1))
            .expect("plan left write")
            .put(&right, &right_key, &WatchRightRow(2))
            .expect("plan right write")
            .commit()
            .expect("commit watched transaction");

        assert!(
            saw_right_unlocked.load(Ordering::Acquire),
            "the left subscriber waker must run after the right participant unlocks"
        );
        drop(recv);
        assert_eq!(left_rx.try_recv().expect("left event").key(), &left_key);
        assert_eq!(right_rx.try_recv().expect("right event").key(), &right_key);
    }

    #[cfg(all(
        feature = "rapira-codec",
        feature = "var-collections",
        feature = "armour"
    ))]
    #[test]
    fn watch_pre_log_transaction_failure_emits_nothing() {
        let dir = tempdir().expect("temporary directory");
        let db = Arc::new(crate::armour::Db::open_test(dir.path()).expect("open db"));
        let tree = db
            .open_typed_tree::<WatchRow, RapiraCodec, _>(
                Config::test(),
                WatchHook::with_old(NoHook),
                &[],
            )
            .expect("open watched participant");
        let mut rx = tree.watch(Default::default());
        let watched_key = key(1);

        let err = db
            .tx()
            .insert(&tree, &watched_key, &WatchRow(1))
            .expect("plan first insert")
            .insert(&tree, &watched_key, &WatchRow(2))
            .expect("plan duplicate insert")
            .commit()
            .expect_err("duplicate insert fails before the log");

        assert!(matches!(err, crate::DbError::KeyExists));
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_update_many_emits_each_applied_mutation_but_not_keep() {
        let (tree, mut rx) = watched_typed_tree();
        let first = key(1);
        let second = key(2);
        let third = key(3);

        let applied = tree
            .update_many(
                vec![(first, 10), (second, 20), (third, 30)],
                |key, _old, value| {
                    if *key == third {
                        Upsert::Keep
                    } else {
                        Upsert::Put(WatchRow(*value))
                    }
                },
            )
            .expect("batch applies");

        assert!(matches!(applied[2].1, crate::Applied::Kept));
        assert_eq!(rx.try_recv().expect("first event").key(), &first);
        assert_eq!(rx.try_recv().expect("second event").key(), &second);
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_retain_emits_delete_for_each_removed_map_row() {
        let (map, mut rx) = watched_typed_map();
        let first = key(1);
        let second = key(2);
        map.put(&first, WatchRow(10)).expect("first insert");
        map.put(&second, WatchRow(20)).expect("second insert");
        let _ = rx.try_recv();
        let _ = rx.try_recv();

        map.iter_view()
            .expect("iterable map")
            .retain(|key, _| *key == second)
            .expect("retain");

        let event = rx.try_recv().expect("delete event");
        assert_eq!(event.key(), &first);
        assert_eq!(event.kind(), WatchKind::Delete);
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_ttl_sweep_emits_delete_after_the_row_is_gone() {
        let dir = tempdir().expect("temporary directory");
        let tree = TypedTree::open_hooked(
            dir.path(),
            Config::test(),
            RapiraCodec,
            crate::TtlHook::new(WatchHook::<TtlWatchRow, NoHook, true>::with_old(NoHook)),
        )
        .expect("open watched ttl tree");
        tree.build_ttl_index().expect("build ttl index");
        let mut rx = tree.watch(Default::default());
        let expired = key(1);
        tree.put(&expired, TtlWatchRow(1))
            .expect("insert expired row");
        let _ = rx.try_recv().expect("insert event");

        let outcome = tree.sweep_expired(2, 1).expect("sweep expired row");
        assert_eq!(outcome.deleted, 1);

        let event = rx.try_recv().expect("delete event");
        assert_eq!(event.kind(), WatchKind::Delete);
        assert!(tree.get(&expired).is_none());
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_ttl_sweep_var_typed_facades_emit_raw_deletes() {
        type Hook = crate::TtlHook<TtlWatchRow, WatchHook<TtlWatchRow, NoHook, true>>;
        let dir = tempdir().expect("temporary directory");
        let tree = VarTypedTree::<[u8; 8], TtlWatchRow, RapiraCodec, Hook>::open_hooked(
            dir.path().join("tree"),
            Config::test(),
            RapiraCodec,
            crate::TtlHook::new(WatchHook::with_old(NoHook)),
        )
        .expect("open watched ttl var tree");
        let map = VarTypedMap::<[u8; 8], TtlWatchRow, RapiraCodec, Hook>::open_hooked(
            dir.path().join("map"),
            Config::balanced().iterable(true).build(),
            RapiraCodec,
            crate::TtlHook::new(WatchHook::with_old(NoHook)),
        )
        .expect("open watched ttl var map");
        tree.as_inner()
            .build_ttl_index()
            .expect("build tree ttl index");
        map.as_inner()
            .build_ttl_index()
            .expect("build map ttl index");
        let mut tree_rx = tree.watch(Default::default());
        let mut map_rx = map.watch(Default::default());
        let expired = key(1);
        tree.put(&expired, &TtlWatchRow(1)).expect("tree insert");
        map.put(&expired, &TtlWatchRow(1)).expect("map insert");
        let _ = tree_rx.try_recv().expect("tree insert event");
        let _ = map_rx.try_recv().expect("map insert event");

        assert_eq!(tree.sweep_expired(2, 1).expect("tree sweep").deleted, 1);
        assert_eq!(map.sweep_expired(2, 1).expect("map sweep").deleted, 1);

        assert_eq!(
            tree_rx.try_recv().expect("tree delete").kind(),
            WatchKind::Delete
        );
        assert_eq!(
            map_rx.try_recv().expect("map delete").kind(),
            WatchKind::Delete
        );
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn new_only_var_typed_nohook_does_not_read_old_but_projected_still_does() {
        let dir = tempdir().expect("temporary directory");
        let mut config = Config::test();
        config.block_cache.max_size = 0;
        let plain = VarTypedTree::open_hooked(
            dir.path().join("plain"),
            config.clone(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
        )
        .expect("open new-only tree");
        let projected_path = dir.path().join("projected");
        let projected = VarTypedTree::open_hooked(
            &projected_path,
            config,
            RapiraCodec,
            WatchHook::<WatchRow, _, false>::new_only(
                crate::Projected::new(crate::IndexTargets::new(), |_, _, _| Ok(())).with(NoHook),
            ),
        )
        .expect("open projected tree");
        let key = key(1);
        plain.put(&key, &WatchRow(1)).expect("plain seed");
        projected.put(&key, &WatchRow(1)).expect("projected seed");
        plain.flush_buffers().expect("plain flush");
        projected.clean_shutdown().expect("projected shutdown");

        let before = crate::var_tree::test_locked_value_reads();
        plain.put(&key, &WatchRow(2)).expect("plain overwrite");
        assert_eq!(crate::var_tree::test_locked_value_reads() - before, 0);

        drop(projected);
        let mut reopened_config = Config::test();
        reopened_config.block_cache.max_size = 0;
        let projected = VarTypedTree::open_hooked(
            &projected_path,
            reopened_config,
            RapiraCodec,
            WatchHook::<WatchRow, _, false>::new_only(
                crate::Projected::new(crate::IndexTargets::new(), |_, _, _| Ok(())).with(NoHook),
            ),
        )
        .expect("reopen projected tree");
        let before = crate::var_tree::test_locked_value_reads();
        projected
            .put(&key, &WatchRow(2))
            .expect("projected overwrite");
        assert_eq!(crate::var_tree::test_locked_value_reads() - before, 1);
    }

    /// The companion the dormant-gate test above cannot give: with a receiver
    /// actually linked, `NEEDS_OLD_VALUE = OLD || H::NEEDS_OLD_VALUE` must make
    /// `with_old` pay one locked value read and `new_only` pay none. Without a
    /// live subscriber both modes read nothing, so the mode itself is untested.
    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn a_live_subscriber_reads_the_old_value_only_in_with_old_mode() {
        let dir = tempdir().expect("temporary directory");
        let mut config = Config::test();
        config.block_cache.max_size = 0;

        let retaining = VarTypedTree::open_hooked(
            dir.path().join("with_old"),
            config.clone(),
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
        )
        .expect("open with_old tree");
        let new_only = VarTypedTree::open_hooked(
            dir.path().join("new_only"),
            config,
            RapiraCodec,
            WatchHook::<WatchRow, NoHook, false>::new_only(NoHook),
        )
        .expect("open new_only tree");

        let key = key(1);
        retaining.put(&key, &WatchRow(1)).expect("with_old seed");
        new_only.put(&key, &WatchRow(1)).expect("new_only seed");
        retaining.flush_buffers().expect("with_old flush");
        new_only.flush_buffers().expect("new_only flush");

        let mut retaining_rx = retaining.watch(Default::default());
        let mut new_only_rx = new_only.watch(Default::default());

        let before = crate::var_tree::test_locked_value_reads();
        retaining
            .put(&key, &WatchRow(2))
            .expect("with_old overwrite");
        assert_eq!(
            crate::var_tree::test_locked_value_reads() - before,
            1,
            "a watched with_old write must read the old value"
        );

        let before = crate::var_tree::test_locked_value_reads();
        new_only
            .put(&key, &WatchRow(2))
            .expect("new_only overwrite");
        assert_eq!(
            crate::var_tree::test_locked_value_reads() - before,
            0,
            "a watched new_only write must not read the old value"
        );

        let retained = retaining_rx.try_recv().expect("with_old event");
        assert_eq!(retained.kind(), WatchKind::Update);
        assert_eq!(retained.old().expect("with_old old"), Some(WatchRow(1)));
        let plain = new_only_rx.try_recv().expect("new_only event");
        assert_eq!(plain.kind(), WatchKind::Upsert);
        assert_eq!(plain.old().expect("new_only old"), None);
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_update_many_raw_var_prefix_wakes_after_replay_on_later_error() {
        use std::sync::atomic::{AtomicBool, Ordering};

        use crate::Codec as _;

        struct MarkerHook {
            watch: WatchHook<WatchRow, NoHook, true>,
            replayed: Arc<AtomicBool>,
        }

        impl crate::TypedWriteHook<[u8; 8], WatchRow> for MarkerHook {
            const WATCHES: bool = true;
            const NEEDS_WRITE: bool = true;

            fn watch_channel(&self) -> Option<&Arc<WatchChannel<[u8; 8]>>> {
                self.watch.watch_channel()
            }

            fn watch_old(&self) -> bool {
                self.watch.watch_old()
            }

            fn on_write(&self, _key: &[u8; 8], _old: Option<&WatchRow>, _new: Option<&WatchRow>) {
                self.replayed.store(true, Ordering::Release);
            }
        }

        impl sealed::Enabled for MarkerHook {}

        struct WakeAfterReplay(Arc<AtomicBool>, Arc<AtomicBool>);

        impl std::task::Wake for WakeAfterReplay {
            fn wake(self: Arc<Self>) {
                self.0
                    .store(self.1.load(Ordering::Acquire), Ordering::Release);
            }
        }

        let dir = tempdir().expect("temporary directory");
        let mut config = Config::test();
        config.shard_count = 1;
        let replayed = Arc::new(AtomicBool::new(false));
        let tree = VarTypedTree::open_hooked(
            dir.path(),
            config,
            RapiraCodec,
            MarkerHook {
                watch: WatchHook::with_old(NoHook),
                replayed: Arc::clone(&replayed),
            },
        )
        .expect("open watched var tree");
        let mut rx = tree.watch(Default::default());
        let first = key(1);
        let second = key(2);
        let woke_after_replay = Arc::new(AtomicBool::new(false));
        let waker = Waker::from(Arc::new(WakeAfterReplay(
            Arc::clone(&woke_after_replay),
            Arc::clone(&replayed),
        )));
        let mut recv = Box::pin(rx.recv());
        assert!(matches!(
            Pin::as_mut(&mut recv).poll(&mut Context::from_waker(&waker)),
            Poll::Pending
        ));

        let _fail = crate::var_tree::UpdateManyFailAfter::after(1);
        let err = tree
            .as_inner()
            .update_many(
                vec![(first, WatchRow(1)), (second, WatchRow(2))],
                |_, _, value| {
                    Upsert::Put(ByteView::from(
                        RapiraCodec.encode(value).expect("encode test value"),
                    ))
                },
            )
            .expect_err("second raw mutation fails");
        assert!(matches!(
            err,
            crate::DbError::Internal("test: update_many injected failure")
        ));

        assert!(woke_after_replay.load(Ordering::Acquire));
        drop(recv);
        assert_eq!(rx.try_recv().expect("applied prefix event").key(), &first);
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_facade_event_is_readable_after_the_row_becomes_visible() {
        let (tree, mut rx) = watched_typed_tree();
        let key = 7u64.to_be_bytes();

        tree.put(&key, WatchRow(70)).expect("put watched row");

        let event = rx.try_recv().expect("watch event");
        assert_eq!(tree.get(&key).map(|value| value.0), Some(70));
        assert_eq!(event.new().expect("decode event"), Some(WatchRow(70)));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_atomic_replays_events_in_application_order_after_unlock() {
        let (tree, mut rx) = watched_typed_tree();
        let first = 1u64.to_be_bytes();
        let second = 2u64.to_be_bytes();

        tree.atomic(&first, |shard| {
            shard.put(&first, WatchRow(1))?;
            shard.put(&second, WatchRow(2))?;
            Ok(())
        })
        .expect("atomic watched writes");

        assert_eq!(
            [
                rx.try_recv().expect("first event").key().to_owned(),
                rx.try_recv().expect("second event").key().to_owned(),
            ],
            [first, second]
        );
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_wake_is_not_under_lock() {
        use std::sync::atomic::{AtomicBool, Ordering};

        struct ReadOnWake {
            tree: Arc<TypedTree<[u8; 8], WatchRow, RapiraCodec, WatchHook<WatchRow, NoHook, true>>>,
            key: [u8; 8],
            writer_released: Arc<AtomicBool>,
        }

        impl std::task::Wake for ReadOnWake {
            fn wake(self: Arc<Self>) {
                self.writer_released.store(
                    self.tree.test_try_lock_writer_shard(&self.key),
                    Ordering::Release,
                );
            }
        }

        let dir = tempdir().expect("temporary directory");
        let mut config = Config::test();
        config.shard_count = 2;
        let tree = Arc::new(
            TypedTree::open_hooked(
                dir.path(),
                config,
                RapiraCodec,
                WatchHook::<WatchRow, NoHook, true>::with_old(NoHook),
            )
            .expect("open watched typed tree"),
        );
        let write_key = 1u64.to_be_bytes();

        let mut rx = tree.watch(Default::default());
        let writer_released = Arc::new(AtomicBool::new(false));
        let waker = Waker::from(Arc::new(ReadOnWake {
            tree: Arc::clone(&tree),
            key: write_key,
            writer_released: Arc::clone(&writer_released),
        }));
        let mut recv = Box::pin(rx.recv());
        assert!(matches!(
            Pin::as_mut(&mut recv).poll(&mut Context::from_waker(&waker)),
            Poll::Pending
        ));

        tree.put(&write_key, WatchRow(2))
            .expect("publish watched row");
        assert!(writer_released.load(Ordering::Acquire));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_var_typed_facades_publish_raw_mutations() {
        let dir = tempdir().expect("temporary directory");
        let hook = || WatchHook::<WatchRow, NoHook, true>::with_old(NoHook);
        let tree =
            VarTypedTree::open_hooked(dir.path().join("tree"), Config::test(), RapiraCodec, hook())
                .expect("open watched var tree");
        let map = VarTypedMap::open_hooked(
            dir.path().join("map"),
            watched_map_config(),
            RapiraCodec,
            hook(),
        )
        .expect("open watched var map");
        let mut tree_rx = tree.watch(Default::default());
        let mut map_rx = map.watch(Default::default());
        let key = 9u64.to_be_bytes();

        tree.put(&key, &WatchRow(9)).expect("put var tree row");
        map.put(&key, &WatchRow(9)).expect("put var map row");

        assert_eq!(
            tree_rx
                .try_recv()
                .expect("tree event")
                .new()
                .expect("decode"),
            Some(WatchRow(9))
        );
        assert_eq!(
            map_rx.try_recv().expect("map event").new().expect("decode"),
            Some(WatchRow(9))
        );
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_ttl_wrapper_preserves_old_update_and_delete() {
        let dir = tempdir().expect("temporary directory");
        let tree = TypedTree::open_hooked(
            dir.path(),
            Config::test(),
            RapiraCodec,
            crate::TtlHook::new(WatchHook::<WatchRow, NoHook, true>::with_old(NoHook)),
        )
        .expect("open watched ttl tree");
        let mut rx = tree.watch(Default::default());
        let key = 11u64.to_be_bytes();

        tree.put(&key, WatchRow(1)).expect("insert");
        let _ = rx.try_recv().expect("insert event");
        tree.put(&key, WatchRow(2)).expect("update");
        let update = rx.try_recv().expect("update event");
        tree.delete(&key).expect("delete");
        let delete = rx.try_recv().expect("delete event");

        assert_eq!(update.kind(), WatchKind::Update);
        assert_eq!(update.old().expect("decode old"), Some(WatchRow(1)));
        assert_eq!(delete.kind(), WatchKind::Delete);
        assert_eq!(delete.old().expect("decode old"), Some(WatchRow(2)));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_projected_wrapper_preserves_old_update_and_delete() {
        let dir = tempdir().expect("temporary directory");
        let hook = crate::Projected::new(crate::IndexTargets::new(), |_, _, _| Ok(()))
            .with(WatchHook::<WatchRow, NoHook, true>::with_old(NoHook));
        let tree = TypedTree::open_hooked(dir.path(), Config::test(), RapiraCodec, hook)
            .expect("open watched projected tree");
        let mut rx = tree.watch(Default::default());
        let key = 12u64.to_be_bytes();

        tree.put(&key, WatchRow(1)).expect("insert");
        let _ = rx.try_recv().expect("insert event");
        tree.put(&key, WatchRow(2)).expect("update");
        let update = rx.try_recv().expect("update event");
        tree.delete(&key).expect("delete");
        let delete = rx.try_recv().expect("delete event");

        assert_eq!(update.kind(), WatchKind::Update);
        assert_eq!(update.old().expect("decode old"), Some(WatchRow(1)));
        assert_eq!(delete.kind(), WatchKind::Delete);
        assert_eq!(delete.old().expect("decode old"), Some(WatchRow(2)));
    }

    #[cfg(all(feature = "rapira-codec", feature = "var-collections"))]
    #[test]
    fn watch_hook_preserves_inner_no_write_and_enabled_wrappers() {
        type Hook = WatchHook<WatchRow, NoHook, true>;
        assert!(<Hook as crate::TypedWriteHook<[u8; 8], WatchRow>>::WATCHES);
        assert!(!<Hook as crate::TypedWriteHook<[u8; 8], WatchRow>>::NEEDS_WRITE);

        fn assert_enabled<H: sealed::Enabled>() {}
        assert_enabled::<crate::ttl::TtlHook<WatchRow, Hook>>();
        assert_enabled::<crate::Projected<[u8; 8], WatchRow, Hook>>();
        assert_enabled::<crate::ZeroHookAdapter<[u8; 8], WatchZeroRow, Hook>>();
        assert_enabled::<crate::VarTypedHookAdapter<[u8; 8], WatchRow, RapiraCodec, Hook>>();
    }

    struct U64Spec;
    #[derive(Clone)]
    struct U64Decoder;

    impl Default for U64Decoder {
        fn default() -> Self {
            Self
        }
    }

    impl sealed::Spec for U64Spec {}

    impl WatchDecoder<u64> for U64Decoder {
        fn decode(&self, bytes: &[u8]) -> DbResult<u64> {
            Ok(u64::from_be_bytes(
                bytes.try_into().expect("test payload length"),
            ))
        }
    }

    impl WatchSpec for U64Spec {
        type Key = [u8; 8];
        type Value = u64;
        type Decoder = U64Decoder;
    }

    fn key(value: u64) -> [u8; 8] {
        value.to_be_bytes()
    }
    fn bytes(value: u64) -> Box<[u8]> {
        value.to_be_bytes().into()
    }

    #[test]
    fn recv_timeout_reports_timeout_on_an_empty_ring_and_actually_waits() {
        let (_channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        let timeout = std::time::Duration::from_millis(60);
        let started = std::time::Instant::now();
        assert_eq!(rx.recv_timeout(timeout), Err(WatchError::Timeout));
        assert!(
            started.elapsed() >= timeout,
            "recv_timeout returned before its deadline"
        );
    }

    /// Precedence: an event already in the ring outranks an expired deadline.
    /// A zero timeout must therefore behave like `try_recv` on a non-empty
    /// ring, not like an immediate `Timeout`.
    #[test]
    fn recv_timeout_returns_a_queued_event_even_with_an_expired_deadline() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(2).unwrap());
        channel.publish_for_test(key(1), None, Some(bytes(10)));

        let event = rx
            .recv_timeout(std::time::Duration::ZERO)
            .expect("queued event outranks the deadline");
        assert_eq!(*event.key(), key(1));
        assert_eq!(event.new().unwrap(), Some(10));
        assert_eq!(
            rx.recv_timeout(std::time::Duration::ZERO),
            Err(WatchError::Timeout)
        );
    }

    /// `Closed` outranks `Timeout`: a terminal channel must not make a caller
    /// spend its deadline waiting for events that can never arrive.
    #[test]
    fn recv_timeout_reports_closed_without_waiting_out_the_deadline() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        channel.publish_for_test(key(1), None, Some(bytes(10)));
        channel.close_terminal();

        assert_eq!(
            *rx.recv_timeout(std::time::Duration::ZERO).unwrap().key(),
            key(1)
        );
        let started = std::time::Instant::now();
        assert_eq!(
            rx.recv_timeout(std::time::Duration::from_secs(30)),
            Err(WatchError::Closed)
        );
        assert!(
            started.elapsed() < std::time::Duration::from_secs(5),
            "a closed channel must not wait out the deadline"
        );
    }

    /// `Lagged(n)` is a count, not a flag: every existing overflow test loses
    /// exactly one event, which cannot tell an exact count from a saturating
    /// "at least one".
    #[test]
    fn lagged_counts_every_dropped_event_not_just_the_first() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        for value in 1..=4u64 {
            channel.publish_for_test(key(value), None, Some(bytes(value * 10)));
        }

        assert_eq!(rx.try_recv(), Err(WatchError::Lagged(3)));
        let survivor = rx.try_recv().expect("the newest event survives");
        assert_eq!(*survivor.key(), key(4));
        assert_eq!(survivor.new().unwrap(), Some(40));
        assert_eq!(rx.try_recv(), Err(WatchError::Empty));

        // The gap is reported once, and the counter restarts from it.
        channel.publish_for_test(key(5), None, Some(bytes(50)));
        channel.publish_for_test(key(6), None, Some(bytes(60)));
        assert_eq!(rx.try_recv(), Err(WatchError::Lagged(1)));
        assert_eq!(*rx.try_recv().expect("post-gap event").key(), key(6));
    }

    #[test]
    fn lagged_is_reported_before_the_post_gap_event() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        channel.publish_for_test(key(1), None, Some(bytes(10)));
        channel.publish_for_test(key(2), None, Some(bytes(20)));

        assert_eq!(rx.try_recv(), Err(WatchError::Lagged(1)));
        let event = rx.try_recv().unwrap();
        assert_eq!(*event.key(), key(2));
        assert_eq!(event.kind(), WatchKind::Upsert);
        assert_eq!(event.new().unwrap(), Some(20));
    }

    #[test]
    fn terminal_close_drains_before_closed() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(2).unwrap());
        channel.publish_for_test(key(1), None, Some(bytes(10)));
        channel.close_terminal();
        assert!(rx.try_recv().is_ok());
        assert_eq!(rx.try_recv(), Err(WatchError::Closed));
    }

    #[test]
    fn recv_many_keeps_a_front_gap_and_zero_max_is_a_noop() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        channel.publish_for_test(key(1), None, Some(bytes(10)));
        channel.publish_for_test(key(2), None, Some(bytes(20)));
        let mut events = Vec::new();
        assert_eq!(rx.recv_many(&mut events, 0), Ok(0));
        assert!(events.is_empty());
        assert_eq!(rx.recv_many(&mut events, 2), Err(WatchError::Lagged(1)));
        assert!(events.is_empty());
        assert_eq!(rx.recv_many(&mut events, 2), Ok(1));
        assert_eq!(*events.pop().unwrap().key(), key(2));
    }

    #[test]
    fn capacity_two_reports_the_gap_before_surviving_events() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(2).unwrap());
        channel.publish_for_test(key(1), None, Some(bytes(10)));
        channel.publish_for_test(key(2), None, Some(bytes(20)));
        channel.publish_for_test(key(3), None, Some(bytes(30)));

        assert_eq!(rx.try_recv(), Err(WatchError::Lagged(1)));
        assert_eq!(*rx.try_recv().unwrap().key(), key(2));
        assert_eq!(*rx.try_recv().unwrap().key(), key(3));
    }

    #[test]
    fn capacity_three_reports_the_gap_before_all_surviving_events() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(3).unwrap());
        for value in 1..=4 {
            channel.publish_for_test(key(value), None, Some(bytes(value * 10)));
        }

        assert_eq!(rx.try_recv(), Err(WatchError::Lagged(1)));
        assert_eq!(*rx.try_recv().unwrap().key(), key(2));
        assert_eq!(*rx.try_recv().unwrap().key(), key(3));
        assert_eq!(*rx.try_recv().unwrap().key(), key(4));
    }

    #[test]
    fn recv_many_reports_a_front_gap_before_surviving_events() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(2).unwrap());
        channel.publish_for_test(key(1), None, Some(bytes(10)));
        channel.publish_for_test(key(2), None, Some(bytes(20)));
        channel.publish_for_test(key(3), None, Some(bytes(30)));

        let mut events = Vec::new();
        assert_eq!(rx.recv_many(&mut events, 2), Err(WatchError::Lagged(1)));
        assert!(events.is_empty());
        assert_eq!(rx.recv_many(&mut events, 2), Ok(2));
        assert_eq!(*events.remove(0).key(), key(2));
        assert_eq!(*events.remove(0).key(), key(3));
    }

    #[test]
    fn terminal_close_waits_for_an_entered_producer() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        let (producer_entered, release_producer) = channel.arm_before_publish();
        let (close_marked, release_close) = channel.arm_after_terminal_mark();
        let producer_channel = Arc::clone(&channel);
        let producer = std::thread::spawn(move || {
            producer_channel.publish_for_test(key(5), None, Some(bytes(50)));
        });
        producer_entered.wait();

        let close_channel = Arc::clone(&channel);
        let closer = std::thread::spawn(move || close_channel.close_terminal());
        close_marked.wait();
        release_close.wait();
        release_producer.wait();
        producer.join().unwrap();
        closer.join().unwrap();

        assert_eq!(*rx.try_recv().unwrap().key(), key(5));
        assert_eq!(rx.try_recv(), Err(WatchError::Closed));
    }

    #[test]
    fn final_deregister_during_started_producer_traversal_is_safe() {
        struct ReleaseAndJoin {
            release: Option<Arc<std::sync::Barrier>>,
            publisher: Option<std::thread::JoinHandle<()>>,
        }
        impl Drop for ReleaseAndJoin {
            fn drop(&mut self) {
                if let Some(release) = self.release.take() {
                    release.wait();
                }
                if let Some(publisher) = self.publisher.take() {
                    let _ = publisher.join();
                }
            }
        }

        let (channel, rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        let (started, release) = channel.arm_after_traversal_starts();
        let publisher = {
            let channel = Arc::clone(&channel);
            std::thread::spawn(move || channel.publish_for_test(key(8), None, Some(bytes(80))))
        };
        let mut cleanup = ReleaseAndJoin {
            release: Some(release),
            publisher: Some(publisher),
        };
        started.wait();
        drop(rx);
        assert_eq!(channel.debug_subscriber_count(), 0);
        cleanup.release.take().expect("traversal release").wait();
        cleanup.publisher.take().expect("publisher").join().unwrap();
        assert_eq!(channel.debug_subscriber_count(), 0);
    }

    #[test]
    fn terminal_close_wakes_a_polled_receiver_and_blocks_later_pushes() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        struct CountWake(Arc<AtomicUsize>);
        impl std::task::Wake for CountWake {
            fn wake(self: Arc<Self>) {
                self.0.fetch_add(1, Ordering::SeqCst);
            }
        }

        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        let wakes = Arc::new(AtomicUsize::new(0));
        let waker = Waker::from(Arc::new(CountWake(Arc::clone(&wakes))));
        let mut pending = Box::pin(rx.recv());
        assert!(matches!(
            pending.as_mut().poll(&mut Context::from_waker(&waker)),
            Poll::Pending
        ));
        channel.close_terminal();
        assert_eq!(wakes.load(Ordering::SeqCst), 1);
        drop(pending);
        channel.publish_for_test(key(9), None, Some(bytes(90)));
        assert_eq!(rx.try_recv(), Err(WatchError::Closed));
    }

    #[test]
    fn dropped_recv_future_unregisters_its_waker_and_event_remains() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        let mut future = Box::pin(rx.recv());
        assert!(matches!(
            Pin::as_mut(&mut future).poll(&mut Context::from_waker(Waker::noop())),
            Poll::Pending
        ));
        drop(future);
        channel.publish_for_test(key(3), None, Some(bytes(30)));
        assert_eq!(rx.try_recv().unwrap().new().unwrap(), Some(30));
        assert_eq!(channel.debug_registered_wakers(), 0);
    }

    #[test]
    fn blocking_receiver_observes_publish_between_empty_check_and_park() {
        let (channel, mut rx) = test_channel::<[u8; 8], U64Spec>(NonZeroUsize::new(1).unwrap());
        let (checked_empty, release) = test_support::arm_before_condvar_park();
        let join = std::thread::spawn(move || rx.recv_blocking());
        checked_empty.wait();
        channel.publish_for_test(key(4), None, Some(bytes(40)));
        release.wait();
        assert_eq!(join.join().unwrap().unwrap().new().unwrap(), Some(40));
    }
}