ruststream 0.7.2

//! Native capability-trait implementations for the in-memory broker.
//!
//! These run on the broker's own in-process semantics: buffered transactions, greedy batching,
//! correlated request / reply over an in-process inbox. The conformance harness offers a suite
//! per capability (see
//! `conformance::capabilities`), and these implementations are the executable reference for
//! them.

use std::{
    fmt,
    future::{Future, ready},
    marker::PhantomData,
    num::{NonZeroU64, NonZeroUsize},
    sync::{
        Arc, Mutex,
        atomic::{AtomicUsize, Ordering},
    },
    task::{Poll, ready},
    time::Duration,
};

use futures::{Stream, task::AtomicWaker};
use thiserror::Error;
use tokio::{sync::mpsc, time::timeout};
use tracing::warn;

use super::{
    Bus, Discarding, LogMode, MemoryDelivery, MemoryError, MemoryMessage, MemoryOutbound,
    MemoryPublisher, MemoryState, MemorySubscriber, Retaining, SubscriptionShared, log,
};
#[cfg(feature = "testing")]
use crate::testing::coordinator::Coordinator;
use crate::{
    BatchSubscriber, BytesMut, IncomingMessage, OutgoingMessage, OwnedTransactions, Partitioned,
    Positioned, Publisher, RequestReply, Seekable, Seeker, Str, Subscriber, Take, Transaction,
    TransactionalPublisher,
};

/// The well-known header the [`Partitioned`] implementation reads the partition key from.
pub const PARTITION_KEY_HEADER: &str = "partition-key";

/// Errors returned by [`MemoryRequester`] operations.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum RequestError {
    /// No reply arrived in the request inbox before the timeout elapsed.
    #[error("no reply to \"{subject}\" within {timeout:?}")]
    Timeout {
        /// The subject the request was published to.
        subject: String,
        /// How long the requester waited for a reply.
        timeout: Duration,
    },
    /// The operation ran through a handle aliasing a bus that was shut down
    /// ([`ConnectedBroker::shutdown`](crate::ConnectedBroker::shutdown)) and not revived by a
    /// sibling clone's [`Broker::connect`](crate::Broker::connect).
    #[error("the memory broker is shut down")]
    ShutDown,
}

/// Publisher with native request / reply support, returned by
/// [`MemoryBroker::requester`](super::MemoryBroker::requester).
///
/// [`request`](RequestReply::request) publishes the message with a unique in-process inbox in
/// the `reply-to` header and resolves on the first message delivered to that inbox. A responder
/// reads `reply-to` from the request headers and publishes its reply there. Plain
/// [`publish`](Publisher::publish) is the usual fire-and-forget fanout.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream::OutgoingMessage;
/// use ruststream::memory::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Serialize)]
/// struct Ask {
///     id: u64,
/// }
///
/// #[derive(Deserialize)]
/// struct Quote {
///     price: u64,
/// }
///
/// #[derive(OutSlot)]
/// struct Pricing;
///
/// #[subscriber("orders")]
/// async fn price(order: &Order, Out(pricing): Out<impl RequestReply, Pricing>) -> HandlerOutcome {
///     let Ok(ask) = serde_json::to_vec(&Ask { id: order.id }) else {
///         return HandlerOutcome::drop();
///     };
///     // The reply comes back on an inbox the request names in its `reply-to` header.
///     let request = OutgoingMessage::new("quotes.ask", ask.as_slice());
///     let Ok(reply) = pricing.request(request, Duration::from_secs(2)).await else {
///         return HandlerOutcome::retry();
///     };
///     match serde_json::from_slice::<Quote>(reply.payload()) {
///         Ok(quote) => {
///             tracing::info!(order.id, quote.price, "priced");
///             HandlerOutcome::ack()
///         }
///         Err(_) => HandlerOutcome::drop(),
///     }
/// }
///
/// fn app() -> RustStream {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
///         b.include(price).out(Pricing, Request).build();
///     })
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Clone)]
pub struct MemoryRequester {
    state: Arc<MemoryState>,
}

impl MemoryRequester {
    pub(super) fn new(state: Arc<MemoryState>) -> Self {
        Self { state }
    }
}

impl fmt::Debug for MemoryRequester {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("MemoryRequester").finish_non_exhaustive()
    }
}

impl Publisher for MemoryRequester {
    // The same bus, kept the same way: see `MemoryPublisher`.
    type Payload = Take;
    type Error = RequestError;
    type Options = ();

    fn publish(
        &self,
        msg: OutgoingMessage<'_, BytesMut>,
        _options: Option<&Self::Options>,
    ) -> impl Future<Output = Result<(), Self::Error>> {
        let (name, payload, headers) = msg.into_parts();
        ready(
            self.state
                .fanout(name, payload.freeze(), headers)
                .map_err(|_| RequestError::ShutDown),
        )
    }
}

impl RequestReply for MemoryRequester {
    /// A reply arrives on a single-use inbox rather than a subscription, so it carries no
    /// position and no seeker whichever broker the requester came from.
    type Reply = MemoryMessage<Discarding>;

    async fn request(
        &self,
        msg: OutgoingMessage<'_, BytesMut>,
        wait: Duration,
    ) -> Result<Self::Reply, Self::Error> {
        let id = self.state.inbox_seq.fetch_add(1, Ordering::Relaxed);
        let inbox = format!("_inbox.{id}");
        let (tx, mut rx) = mpsc::unbounded_channel();
        self.state
            .register(&inbox, tx.clone())
            .map_err(|_| RequestError::ShutDown)?;

        // The name is the caller's buffer, so it outlives the message consumed here; the inbox
        // is written into the request's own map rather than into a copy of it.
        let (subject, payload, mut headers) = msg.into_parts();
        headers.insert(Str::from_static("reply-to"), inbox.clone());
        if self
            .state
            .fanout(subject, payload.freeze(), headers)
            .is_err()
        {
            self.state.unregister(&inbox);
            return Err(RequestError::ShutDown);
        }

        let outcome = timeout(wait, rx.recv()).await;
        self.state.unregister(&inbox);
        match outcome {
            Ok(Some(reply)) => Ok(MemoryMessage {
                delivery: Some(reply),
                subscription: None,
                // The inbox reply is consumed here, not by a dispatch loop, and its enqueue was not
                // counted (see `fanout`), so it carries no coordinator.
                #[cfg(feature = "testing")]
                coordinator: None,
                mode: PhantomData,
            }),
            // `tx` is held on this stack frame, so the channel cannot report closed.
            Ok(None) => unreachable!("request inbox closed while its sender is held"),
            Err(_) => Err(RequestError::Timeout {
                subject: subject.to_owned(),
                timeout: wait,
            }),
        }
    }
}

/// Greedy batching: a batch is the first awaited delivery plus everything already buffered, up to
/// the size the registration asked for. Partial batches ship immediately, so no deadline timer is
/// needed - the in-memory transport has nothing to wait for. Each batch is allocated at the
/// requested size up front, so a partial batch holds the room of a full one.
impl<Log: LogMode> BatchSubscriber for MemorySubscriber<Log> {
    type Batch = Vec<MemoryMessage<Log>>;

    fn batches(
        &mut self,
        size: NonZeroUsize,
    ) -> impl Stream<Item = Result<Self::Batch, <Self as Subscriber>::Error>> + Send + '_ {
        let limit = size.get();
        #[cfg(feature = "testing")]
        let coordinator = self.coordinator();
        // The drain happens inside a single poll, so no batch state is buffered between polls
        // and the stream stays cancel-safe, like `MemorySubscriber::stream`.
        futures::stream::poll_fn(move |cx| {
            // Same ordering as `MemorySubscriber::stream`: register, then apply a pending seek.
            // Only a seeker wakes this waker, and a discarding subscription mints none.
            if log::retains::<Log>() {
                self.subscription.seek.waker.register(cx.waker());
            }
            self.apply_pending_seek();
            let first = loop {
                match ready!(self.rx.poll_recv(cx)) {
                    // A stale pre-seek copy (a requeue that raced the seek): drop it, the
                    // replay already covers everything from the watermark on.
                    Some(delivery)
                        if log::retains::<Log>()
                            && delivery.seq < self.subscription.seek.watermark() =>
                    {
                        #[cfg(feature = "testing")]
                        if let Some(coordinator) = &coordinator {
                            coordinator.consumed();
                        }
                    }
                    Some(delivery) => break delivery,
                    None => return Poll::Ready(None),
                }
            };
            // Sized for the batch the registration asked for: a batch grown one push at a time
            // reallocates six times on its way to 64.
            let mut batch = Vec::with_capacity(limit);
            batch.push(MemoryMessage {
                delivery: Some(first),
                subscription: Some(Arc::clone(&self.subscription)),
                #[cfg(feature = "testing")]
                coordinator: coordinator.clone(),
                mode: PhantomData,
            });
            while batch.len() < limit {
                match self.rx.poll_recv(cx) {
                    // The same stale-copy filter as above, off the batch.
                    Poll::Ready(Some(delivery))
                        if log::retains::<Log>()
                            && delivery.seq < self.subscription.seek.watermark() =>
                    {
                        #[cfg(feature = "testing")]
                        if let Some(coordinator) = &coordinator {
                            coordinator.consumed();
                        }
                    }
                    Poll::Ready(Some(delivery)) => batch.push(MemoryMessage {
                        delivery: Some(delivery),
                        subscription: Some(Arc::clone(&self.subscription)),
                        #[cfg(feature = "testing")]
                        coordinator: coordinator.clone(),
                        mode: PhantomData,
                    }),
                    // Drained (pending) or closed: ship what we have; a closed channel ends
                    // the stream on the next poll.
                    Poll::Ready(None) | Poll::Pending => break,
                }
            }
            Poll::Ready(Some(Ok(batch)))
        })
    }
}

/// In-process transactions: [`begin_transaction`](TransactionalPublisher::begin_transaction)
/// switches this handle to buffering, [`commit`](TransactionalPublisher::commit) fans out the
/// buffer in publish order, [`abort`](TransactionalPublisher::abort) discards it.
///
/// Misuse errors per the trait contract: a second `begin_transaction` returns
/// [`MemoryError::TransactionBusy`] (leaving the open transaction untouched), and `commit` /
/// `abort` without an open transaction return [`MemoryError::NoTransaction`]. Clones of the
/// handle do not share the transaction (see [`MemoryPublisher`]).
impl TransactionalPublisher for MemoryPublisher {
    fn begin_transaction(&self) -> impl Future<Output = Result<(), MemoryError>> {
        let mut txn = self.txn.lock().expect("memory broker mutex poisoned");
        if txn.is_some() {
            return ready(Err(MemoryError::TransactionBusy));
        }
        *txn = Some(Vec::new());
        drop(txn);
        ready(Ok(()))
    }

    fn commit(&self) -> impl Future<Output = Result<(), MemoryError>> {
        let buffered = self
            .txn
            .lock()
            .expect("memory broker mutex poisoned")
            .take();
        let Some(buffered) = buffered else {
            return ready(Err(MemoryError::NoTransaction));
        };
        for outbound in buffered {
            if let Err(err) = self
                .state
                .fanout(&outbound.name, outbound.payload, outbound.headers)
            {
                return ready(Err(err));
            }
        }
        ready(Ok(()))
    }

    fn abort(&self) -> impl Future<Output = Result<(), MemoryError>> {
        ready(
            self.txn
                .lock()
                .expect("memory broker mutex poisoned")
                .take()
                .map(|_| ())
                .ok_or(MemoryError::NoTransaction),
        )
    }
}

/// An owned in-process transaction, opened by
/// [`transaction`](crate::OwnedTransactions::transaction) on a [`MemoryPublisher`].
///
/// A private delivery buffer, fanned out to the bus in publish order on commit and discarded
/// on abort.
///
/// Unlike the handle-level [`TransactionalPublisher`] buffer, any number of these can be open
/// on one handle at a time, and the handle keeps publishing directly while they are.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
/// # mod demo {
/// use ruststream::memory::prelude::*;
/// use serde::{Deserialize, Serialize};
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[derive(Serialize, Outgoing)]
/// #[outgoing(name = "ledger.settled")]
/// struct Settled {
///     id: u64,
/// }
///
/// #[derive(OutSlot)]
/// #[publishes(Settled)]
/// struct Ledger;
///
/// #[subscriber("orders")]
/// async fn settle(
///     order: &Order,
///     Out(ledger): Out<impl OwnedTransactions, Ledger, Settled>,
/// ) -> HandlerOutcome {
///     // Nothing reaches `ledger.settled` until the commit, and an abort discards it all.
///     let Ok(mut txn) = ledger.transaction().await else {
///         return HandlerOutcome::retry();
///     };
///     if txn
///         .message(&Settled { id: order.id })
///         .publish()
///         .await
///         .is_err()
///         || txn.commit().await.is_err()
///     {
///         return HandlerOutcome::retry();
///     }
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
///         b.include(settle).out(Ledger, Publish).build();
///     })
/// }
/// # }
/// # fn main() {}
/// ```
#[must_use = "a transaction does nothing until settled with commit() or abort()"]
pub struct MemoryTransaction {
    state: Arc<MemoryState>,
    buffered: Vec<MemoryOutbound>,
    settled: bool,
}

impl fmt::Debug for MemoryTransaction {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("MemoryTransaction")
            .field("buffered", &self.buffered.len())
            .field("settled", &self.settled)
            .finish_non_exhaustive()
    }
}

impl Drop for MemoryTransaction {
    fn drop(&mut self) {
        // Destructors cannot run async work, so a drop can only discard the buffer; the warning
        // marks that as an abort the caller never wrote, mirroring the runtime's
        // TransactionScope drop.
        if !self.settled {
            warn!(
                target: "ruststream::memory",
                buffered = self.buffered.len(),
                "owned transaction dropped without commit or abort; its buffered messages are \
                 discarded"
            );
        }
    }
}

impl Transaction for MemoryTransaction {
    // The buffer holds the messages until the commit fans them out, so it keeps what it is
    // handed.
    type Payload = Take;
    type Error = MemoryError;
    type Options = ();

    fn publish(
        &mut self,
        msg: OutgoingMessage<'_, BytesMut>,
        _options: Option<&Self::Options>,
    ) -> impl Future<Output = Result<(), MemoryError>> {
        // Buffering is local to this value and never touches the bus; a commit against a
        // shut-down bus is what reports the error.
        self.buffered.push(MemoryOutbound::from(msg));
        ready(Ok(()))
    }

    fn commit(mut self) -> impl Future<Output = Result<(), MemoryError>> {
        // Settled before the flush: a failed commit has still consumed the transaction (the
        // buffer is lost per the Transaction contract), so the drop warning must not fire.
        self.settled = true;
        for outbound in std::mem::take(&mut self.buffered) {
            if let Err(err) = self
                .state
                .fanout(&outbound.name, outbound.payload, outbound.headers)
            {
                return ready(Err(err));
            }
        }
        ready(Ok(()))
    }

    fn abort(mut self) -> impl Future<Output = Result<(), MemoryError>> {
        self.settled = true;
        ready(Ok(()))
    }
}

/// Owned transactions: every [`transaction`](OwnedTransactions::transaction) call opens an
/// independent buffer-owning `MemoryTransaction`, so any number can be open concurrently on one
/// handle, next to (and unaffected by) the handle-level [`TransactionalPublisher`] transaction.
impl OwnedTransactions for MemoryPublisher {
    type Transaction = MemoryTransaction;

    fn transaction(&self) -> impl Future<Output = Result<MemoryTransaction, MemoryError>> {
        // Opening allocates a buffer and never touches the bus; a shut-down bus surfaces at
        // commit, the visibility point, like the handle-level begin.
        ready(Ok(MemoryTransaction {
            state: Arc::clone(&self.state),
            buffered: Vec::new(),
            settled: false,
        }))
    }
}

/// The partition key is carried in the [`PARTITION_KEY_HEADER`] header; producers that want
/// per-key ordering set it at publish time. Messages without the header return `None` (any
/// partition will do).
impl<Log: LogMode> Partitioned for MemoryMessage<Log> {
    fn partition_key(&self) -> Option<&[u8]> {
        self.headers().get(PARTITION_KEY_HEADER)
    }
}

/// Shared between a subscriber's polling side and the seekers minted off it.
///
/// A seek is a handoff, not an in-place mutation: the seeker records the target and wakes the
/// stream, and the subscriber applies it inside its own poll, where `&mut` access to the
/// receiver is available (see [`MemorySubscriber::apply_pending_seek`]).
#[derive(Default)]
pub(super) struct SeekControl {
    /// The requested log position, taken by the subscriber inside its next poll.
    pending: Mutex<Option<usize>>,
    /// Deliveries stamped below this log position are stale pre-seek copies (a requeue racing
    /// the seek) and are dropped by the polling side.
    watermark: AtomicUsize,
    /// Wakes the subscriber's stream task after `pending` is set.
    pub(super) waker: AtomicWaker,
}

impl SeekControl {
    /// The stale-delivery cutoff, read on the polling side for every delivery.
    ///
    /// Acquire pairs with the Release store in [`Seeker::seek`]: a poll that observes a
    /// delivery enqueued after a seek also observes that seek's watermark.
    pub(super) fn watermark(&self) -> usize {
        self.watermark.load(Ordering::Acquire)
    }
}

/// A position in one name's publish log on a retaining in-memory broker.
///
/// Construct it with [`start`](Self::start), [`sequence`](Self::sequence) or [`end`](Self::end),
/// or capture the position of a delivered [`MemoryMessage`] with [`Positioned::position`].
/// Sequence numbers are absolute and stable: the `n`-th message published to a name keeps
/// position `n` while the broker retains it, whatever the retention bound has evicted before it,
/// and seeking to a captured position redelivers exactly that message.
///
/// The two constructed positions follow the broker's own vocabulary: [`start`](Self::start) is
/// the oldest message still retained, [`end`](Self::end) is the tip, past everything published so
/// far. A sequence the retention bound has already evicted is refused with
/// [`MemoryError::PositionEvicted`]; one past the tip resumes with the next publish.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
/// # mod demo {
/// use ruststream::memory::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Entry {
///     id: u64,
/// }
///
/// /// Rebuilds its view from everything the broker still holds, on every start.
/// #[subscriber("audit", start_at(MemoryPosition::start()))]
/// async fn rebuild(entry: &Entry) -> HandlerOutcome {
///     tracing::info!(entry.id, "replayed");
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     // The last 1024 messages of every name stay replayable.
///     let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(1024)));
///     RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
///         b.include(rebuild);
///     })
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
#[must_use]
pub enum MemoryPosition {
    /// The oldest message the name still retains.
    Start,
    /// The zero-based `n`-th message published to the name.
    Sequence(usize),
    /// The tip of the log: everything published so far is behind it.
    End,
}

impl MemoryPosition {
    /// The oldest retained message: seeking here replays everything the broker still holds for
    /// the name.
    pub const fn start() -> Self {
        Self::Start
    }

    /// The position of the zero-based `n`-th message published to the name.
    pub const fn sequence(n: usize) -> Self {
        Self::Sequence(n)
    }

    /// The end of the log: seeking here skips everything published so far and resumes with
    /// the next publish (the memory analog of a "latest" position).
    pub const fn end() -> Self {
        Self::End
    }
}

/// A clonable handle repositioning one [`MemorySubscriber`], minted by
/// [`Seekable::seeker`] before the subscription's stream is opened.
///
/// The scope is that one subscription instance: other subscribers of the same name are
/// unaffected. After the bus shuts down, [`seek`](Seeker::seek) reports
/// [`MemoryError::ShutDown`] like any other aliased handle.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "macros", feature = "memory", feature = "json"))]
/// # mod demo {
/// use ruststream::Seeker;
/// use ruststream::memory::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Job {
///     id: u64,
///     poisoned_until: Option<usize>,
/// }
///
/// /// Skips forward past a region the producer marked poisoned.
/// #[subscriber("jobs")]
/// async fn work(job: &Job, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
///     if let Some(resume_at) = job.poisoned_until {
///         if seeker
///             .seek(MemoryPosition::sequence(resume_at))
///             .await
///             .is_err()
///         {
///             return HandlerOutcome::retry();
///         }
///     }
///     tracing::info!(job.id, "done");
///     HandlerOutcome::ack()
/// }
///
/// fn app() -> RustStream {
///     let broker = MemoryBroker::retaining(Retention::Messages(nonzero!(1024)));
///     RustStream::new(AppInfo::new("jobs", "0.1.0")).with_broker(broker, |b| {
///         b.include(work);
///     })
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Clone)]
pub struct MemorySeeker {
    // The subscription's shared block, which every delivery already holds: minting a seeker off
    // a delivery, as the per-delivery context does, is one reference count.
    subscription: Arc<SubscriptionShared>,
}

impl MemorySeeker {
    /// The seeker of the subscription `msg` came from: the reference count the delivery holds,
    /// taken once more, so a per-delivery context allocates nothing.
    ///
    /// # Panics
    ///
    /// Panics on a request-reply inbox message, which has no subscription behind it; no dispatch
    /// loop, and therefore no context, ever builds off one.
    fn of(msg: &MemoryMessage<Retaining>) -> Self {
        Self {
            subscription: Arc::clone(
                msg.subscription
                    .as_ref()
                    .expect("a seek context builds only off subscription deliveries"),
            ),
        }
    }
}

impl fmt::Debug for MemorySeeker {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("MemorySeeker")
            .field("name", &self.subscription.name)
            .finish_non_exhaustive()
    }
}

/// Seeking is native on a retaining broker: it keeps a bounded publish log per name, so a
/// subscription can re-enqueue any retained suffix of it. A discarding broker keeps no log, and
/// its subscriptions carry no seek surface at all.
impl Seekable for MemorySubscriber<Retaining> {
    type Seeker = MemorySeeker;

    fn seeker(&self) -> MemorySeeker {
        MemorySeeker {
            subscription: Arc::clone(&self.subscription),
        }
    }
}

impl Seeker for MemorySeeker {
    type Position = MemoryPosition;
    type Error = MemoryError;

    /// Records the reposition and wakes the subscription; the stream applies it at the top of
    /// its next poll, so once this resolves the next delivery reflects the new position.
    ///
    /// Under a [`TestApp`](crate::testing::TestApp) the reposition counts as in-flight work
    /// until the subscription applies it, so driving the service to a standstill waits for the
    /// replay instead of racing the poll that produces it.
    ///
    /// # Errors
    ///
    /// Returns [`MemoryError::ShutDown`] through a handle aliasing a shut-down bus, and
    /// [`MemoryError::PositionEvicted`] for a sequence the retention bound has already dropped.
    fn seek(&self, to: MemoryPosition) -> impl Future<Output = Result<(), MemoryError>> {
        // The liveness check and the handoff happen under the bus lock (then the log lock,
        // fanout's order): a seek that returns Ok happened strictly before any shutdown, and
        // the resolved position is consistent with the log at that instant.
        let bus = self
            .subscription
            .state
            .subscribers
            .lock()
            .expect("memory broker mutex poisoned");
        if !matches!(&*bus, Bus::Live(_)) {
            return ready(Err(MemoryError::ShutDown));
        }
        let log = self
            .subscription
            .state
            .log
            .lock()
            .expect("memory broker mutex poisoned");
        let (oldest, tip) = log
            .name(&self.subscription.name)
            .map_or((0, 0), |name| (name.first_seq(), name.next_seq()));
        drop(log);
        let target = match to {
            MemoryPosition::Start => oldest,
            MemoryPosition::End => tip,
            // Past the tip is a position the name has not reached yet: the subscription waits
            // there rather than failing, and the next publish opens it.
            MemoryPosition::Sequence(n) if n >= oldest => n.min(tip),
            MemoryPosition::Sequence(n) => {
                return ready(Err(MemoryError::PositionEvicted {
                    requested: n,
                    oldest,
                }));
            }
        };
        // Watermark first (Release, paired with the Acquire load in the delivery filter), then
        // the pending target: a poll that takes the target must see its watermark.
        self.subscription
            .seek
            .watermark
            .store(target, Ordering::Release);
        let replaced = self
            .subscription
            .seek
            .pending
            .lock()
            .expect("memory broker mutex poisoned")
            .replace(target);
        drop(bus);
        // Under the harness, hold the reaction open from the moment the reposition is recorded
        // until the stream applies it: the replay it will enqueue is not counted yet, and a
        // quiescence wait landing in that window would return on an empty in-flight count.
        // One token per pending reposition, released when the subscription applies it: a second
        // seek before the first is applied replaces the target and owes nothing more.
        #[cfg(feature = "testing")]
        if replaced.is_none()
            && let Some(coordinator) = self.subscription.state.coordinator()
        {
            coordinator.enqueued();
        }
        #[cfg(not(feature = "testing"))]
        let _ = replaced;
        self.subscription.seek.waker.wake();
        ready(Ok(()))
    }
}

impl<Log: LogMode> MemorySubscriber<Log> {
    /// Applies a reposition requested through a [`MemorySeeker`], if one is pending.
    ///
    /// Runs at the top of the subscriber's own poll (the stream and batch paths both call it),
    /// which is what makes `&mut` access to the receiver possible: everything queued before
    /// the seek is drained, then the retained log from the resolved target on is re-enqueued.
    /// Holding both bus locks makes the swap atomic against `fanout`, which stamps and sends
    /// under the same locks in the same order.
    pub(super) fn apply_pending_seek(&mut self) {
        // A discarding subscription hands out no seeker, so nothing can ever be pending: the
        // constant drops the whole body, locks included, from its poll.
        if !log::retains::<Log>() {
            return;
        }
        // Take the target in its own statement so the pending guard is released before the
        // bus lock is acquired; `Seeker::seek` nests the locks in the other direction.
        let pending = self
            .subscription
            .seek
            .pending
            .lock()
            .expect("memory broker mutex poisoned")
            .take();
        let Some(target) = pending else { return };
        #[cfg(feature = "testing")]
        let coordinator = self.coordinator();
        // The seek counted itself in flight when it recorded this target; the token goes back
        // however this returns, so a quiescence wait cannot observe the gap between the request
        // and the replay, and a seek that raced shutdown still drains the in-flight count.
        #[cfg(feature = "testing")]
        let seek_token = SeekToken(coordinator.clone());

        let bus = self
            .subscription
            .state
            .subscribers
            .lock()
            .expect("memory broker mutex poisoned");
        // A seek that raced shutdown: the subscription is dead, replay nothing.
        if !matches!(&*bus, Bus::Live(_)) {
            return;
        }
        let log = self
            .subscription
            .state
            .log
            .lock()
            .expect("memory broker mutex poisoned");
        let retained = log.name(self.subscription.name.as_str());

        // The replay is counted in flight BEFORE the drain releases the queued deliveries:
        // decrementing first would let the in-flight count touch zero mid-swap, and a
        // concurrent quiescence wait (`TestApp::settle`) could observe that instant and return
        // before the replayed deliveries were processed.
        #[cfg(feature = "testing")]
        if let Some(coordinator) = &coordinator {
            for _ in retained.into_iter().flat_map(|log| log.replay_from(target)) {
                coordinator.enqueued();
            }
        }

        while self.rx.try_recv().is_ok() {
            // Every drained delivery was counted in flight when it was enqueued.
            #[cfg(feature = "testing")]
            if let Some(coordinator) = &coordinator {
                coordinator.consumed();
            }
        }

        if let Some(retained) = retained {
            for (seq, entry) in retained.replay_from(target) {
                let delivery = MemoryDelivery {
                    // The block the fanout built, name and payload and headers together: a
                    // replayed delivery is one reference count, not a rebuilt message.
                    shared: entry.shared(),
                    seq,
                    // A replay is a fresh delivery of what the log holds, not a redelivery of
                    // the copy a handler saw, so its count starts where a fanout's does.
                    deliveries: NonZeroU64::MIN,
                };
                // The send cannot fail: this subscriber holds both ends of its own channel.
                let _ = self.subscription.requeue.send(delivery);
            }
        }
        drop(log);
        drop(bus);
        #[cfg(feature = "testing")]
        drop(seek_token);
    }
}

/// The in-flight token one pending reposition holds, released when the subscription has applied
/// it (whichever way [`apply_pending_seek`](MemorySubscriber::apply_pending_seek) returns).
#[cfg(feature = "testing")]
struct SeekToken(Option<Coordinator>);

#[cfg(feature = "testing")]
impl Drop for SeekToken {
    fn drop(&mut self) {
        if let Some(coordinator) = &self.0 {
            coordinator.consumed();
        }
    }
}

/// The position is the delivery's absolute sequence number in its name's publish log, assigned
/// at fanout and preserved across requeues, so a redelivered message reports the same position.
/// Only a retaining broker's deliveries have one: without a log there is nothing to be a
/// position in.
impl Positioned for MemoryMessage<Retaining> {
    type Position = MemoryPosition;

    fn position(&self) -> MemoryPosition {
        MemoryPosition::Sequence(self.delivery.as_ref().map_or(0, |d| d.seq))
    }
}

/// The in-memory broker's per-delivery context: this delivery's position in its name's publish
/// log, and the subscription's own seeker.
///
/// The runtime builds one per dispatched delivery (see [`BuildContext`](crate::BuildContext)),
/// and handlers read its fields by key - [`Position`] and [`SeekHandle`] - through the
/// [`Ctx`](crate::runtime::Ctx) extractor or `ctx.context(..)`. A body that manages a log
/// cursor names this type as its `C` axis and needs nothing else.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "memory", feature = "json"))]
/// # mod demo {
/// use ruststream::memory::{MemoryContext, Position, SeekHandle};
/// use ruststream::prelude::*;
/// use ruststream::Seeker;
/// # #[derive(serde::Deserialize, schemars::JsonSchema)]
/// # struct Job { id: u64 }
///
/// struct Replayer;
///
/// impl Handle<Job, (), (), MemoryContext> for Replayer {
///     async fn handle(
///         &self,
///         job: &Job,
///         _outs: &(),
///         ctx: &mut Context<'_, MemoryContext>,
///     ) -> Result<(), HandlerOutcome> {
///         let here = ctx.context(Position);
///         if job.id == u64::MAX && ctx.context(SeekHandle).seek(here).await.is_err() {
///             return Err(HandlerOutcome::retry());
///         }
///         Ok(())
///     }
/// }
/// # }
/// ```
#[derive(Debug)]
pub struct MemoryContext {
    position: MemoryPosition,
    seeker: MemorySeeker,
}

impl crate::BuildContext<MemoryMessage<Retaining>> for MemoryContext {
    /// # Panics
    ///
    /// Panics on a request-reply inbox message, which is not a subscription delivery; no
    /// dispatch loop (and therefore no per-delivery context) ever builds off one.
    fn build(msg: &MemoryMessage<Retaining>) -> Self {
        Self {
            position: Positioned::position(msg),
            seeker: MemorySeeker::of(msg),
        }
    }
}

/// The in-memory broker's subscription-scoped batch context: the subscription's own seeker,
/// shared by every delivery of the batch.
///
/// The runtime builds one per dispatched batch from the batch's first delivery (see
/// [`BuildBatchContext`](crate::BuildBatchContext)), and a batch body reads it by key -
/// [`SeekHandle`] - through `ctx.context(..)`. Per-delivery data (a [`Position`]) has no place
/// here: a batch spans many deliveries, so the position a body reacts to rides the elements
/// themselves (a `&[Message<H, T>]` batch reads it off each element's header contract), and
/// keeping this a separate type from [`MemoryContext`] is what rejects a batch body asking for
/// per-delivery fields at compile time.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "memory", feature = "json"))]
/// # mod demo {
/// use ruststream::Seeker;
/// use ruststream::memory::{MemoryBatchContext, MemoryPosition, SeekHandle};
/// use ruststream::prelude::*;
/// # #[derive(serde::Deserialize, schemars::JsonSchema)]
/// # struct Job { id: u64 }
///
/// struct Replayer;
///
/// impl Handle<[Job], (), (), MemoryBatchContext> for Replayer {
///     async fn handle(
///         &self,
///         batch: &[Job],
///         _outs: &(),
///         ctx: &mut Context<'_, MemoryBatchContext>,
///     ) -> Result<(), Vec<HandlerOutcome>> {
///         // A batch that saw the rewind marker repositions the whole subscription once it is
///         // settled; the next batch opens at the target.
///         if batch.iter().any(|job| job.id == u64::MAX)
///             && ctx
///                 .context(SeekHandle)
///                 .seek(MemoryPosition::start())
///                 .await
///                 .is_err()
///         {
///             return Err(batch.iter().map(|_| HandlerOutcome::retry()).collect());
///         }
///         Ok(())
///     }
/// }
/// # }
/// ```
#[derive(Debug)]
pub struct MemoryBatchContext {
    seeker: MemorySeeker,
}

impl crate::BuildBatchContext<MemoryMessage<Retaining>> for MemoryBatchContext {
    /// # Panics
    ///
    /// Panics on a request-reply inbox message, which is not a subscription delivery; no
    /// dispatch loop (and therefore no batch context) ever builds off one.
    fn build(first: &MemoryMessage<Retaining>) -> Self {
        Self {
            seeker: MemorySeeker::of(first),
        }
    }
}

impl crate::Field<MemoryBatchContext> for SeekHandle {
    type Value<'a> = &'a MemorySeeker;

    fn get(self, src: &MemoryBatchContext) -> &MemorySeeker {
        &src.seeker
    }
}

/// The key reading this delivery's [`MemoryPosition`] out of [`MemoryContext`].
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
/// # mod demo {
/// use ruststream::memory::Position;
/// use ruststream::prelude::*;
/// use ruststream::subscriber;
/// # #[derive(serde::Deserialize)]
/// # struct Job { id: u64 }
///
/// #[subscriber("jobs")]
/// async fn audit(job: &Job, Ctx(at): Ctx<Position>) -> HandlerOutcome {
///     println!("job {} sits at {at:?}", job.id);
///     HandlerOutcome::ack()
/// }
/// # }
/// ```
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct Position;

impl crate::ContextField for Position {
    type Context = MemoryContext;
    type Value = MemoryPosition;

    fn read(self, src: &MemoryContext) -> MemoryPosition {
        src.position
    }
}

impl crate::Field<MemoryContext> for Position {
    type Value<'a> = MemoryPosition;

    fn get(self, src: &MemoryContext) -> MemoryPosition {
        src.position
    }
}

/// The key reading the subscription's [`MemorySeeker`] out of [`MemoryContext`]: the reposition
/// handle, resolved once when the subscription opens and carried by every delivery's context.
///
/// # Examples
///
/// ```
/// # #[cfg(all(feature = "memory", feature = "macros", feature = "json"))]
/// # mod demo {
/// use ruststream::memory::{MemoryPosition, SeekHandle};
/// use ruststream::prelude::*;
/// use ruststream::{Seeker, subscriber};
/// # #[derive(serde::Deserialize)]
/// # struct Job { id: u64, poisoned_until: Option<usize> }
///
/// /// Skips forward past a region the producer marked poisoned.
/// #[subscriber("jobs")]
/// async fn work(job: &Job, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
///     if let Some(resume_at) = job.poisoned_until {
///         if seeker.seek(MemoryPosition::sequence(resume_at)).await.is_err() {
///             return HandlerOutcome::retry();
///         }
///     }
///     HandlerOutcome::ack()
/// }
/// # }
/// ```
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash)]
pub struct SeekHandle;

impl crate::ContextField for SeekHandle {
    type Context = MemoryContext;
    type Value = MemorySeeker;

    fn read(self, src: &MemoryContext) -> MemorySeeker {
        src.seeker.clone()
    }
}

impl crate::Field<MemoryContext> for SeekHandle {
    type Value<'a> = &'a MemorySeeker;

    fn get(self, src: &MemoryContext) -> &MemorySeeker {
        &src.seeker
    }
}

#[cfg(test)]
mod tests;