otel-arrow-dfe-engine 0.61.0

// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

//! Core value types shared across the topic crate.

use otel_arrow_dfe_config::SubscriptionGroupName;
use otel_arrow_dfe_config::topic::{TopicBroadcastAckMode, TopicBroadcastOnLagPolicy};
use otel_arrow_dfe_telemetry::otel_warn;
use parking_lot::Mutex;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::{Notify, OwnedSemaphorePermit};
use tokio::time::{Instant, sleep_until};

/// A delivered message envelope carrying a broker-assigned id and the payload.
#[derive(Debug)]
pub struct Envelope<T> {
    /// Broker-assigned message identifier.
    pub id: u64,
    /// Whether the message was published through the tracked-outcome path.
    pub tracked: bool,
    /// Shared payload reference delivered to subscribers.
    pub payload: Arc<T>,
}

impl<T> Clone for Envelope<T> {
    fn clone(&self) -> Self {
        Self {
            id: self.id,
            tracked: self.tracked,
            payload: Arc::clone(&self.payload),
        }
    }
}

/// Items yielded by `Subscription::recv`.
#[derive(Debug)]
pub enum RecvItem<T> {
    /// A normal message.
    Message(Envelope<T>),
    /// Notification that this broadcast subscriber lagged and missed messages.
    Lagged {
        /// Number of dropped messages for this subscriber.
        missed: u64,
    },
}

/// Result of a non-blocking publish attempt.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PublishOutcome {
    /// Message was accepted for delivery.
    Published,
    /// Message was dropped because at least one balanced queue was full.
    ///
    /// In mixed topics, `try_publish` is all-or-nothing: returning this
    /// outcome means the message was not published to either the balanced or
    /// broadcast side.
    DroppedOnFull,
}

/// Terminal outcome for a tracked publish.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TrackedPublishOutcome {
    /// Message processing succeeded.
    Ack,
    /// Message processing failed downstream.
    Nack {
        /// Reason attached by the downstream nack or generated by the runtime.
        reason: Arc<str>,
    },
    /// Message did not receive a terminal Ack/Nack before the configured timeout.
    TimedOut,
    /// The topic closed before the tracked message received a terminal outcome.
    TopicClosed,
}

/// Result of a non-blocking tracked publish attempt.
#[derive(Debug)]
pub enum TrackedTryPublishOutcome {
    /// Message was accepted for tracked delivery.
    Published(TrackedPublishReceipt),
    /// Message was dropped because a balanced queue was full.
    DroppedOnFull,
    /// No tracked-outcome slot was available for this publisher.
    MaxInFlightReached,
}

/// Opaque capacity token for one tracked publish.
///
/// The topic handle acquires this token before calling tracked publish APIs.
/// Backend implementations only need to hold it until the tracked publish
/// reaches a terminal outcome, at which point dropping the token releases the
/// in-flight capacity back to the publisher.
pub struct TrackedPublishPermit(Option<OwnedSemaphorePermit>);

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

impl TrackedPublishPermit {
    pub(crate) fn from_tokio_owned(permit: OwnedSemaphorePermit) -> Self {
        Self(Some(permit))
    }
}

impl Drop for TrackedPublishPermit {
    fn drop(&mut self) {
        _ = self.0.take();
    }
}

/// Shared tracker for tracked publish receipts and timeout handling.
///
/// Backends can register tracked publishes in this tracker, then resolve or
/// discard them later by message id. The tracker runs one shared timeout worker
/// for all pending tracked publishes.
#[derive(Clone)]
pub struct TrackedPublishTracker {
    inner: Arc<TrackedPublishTrackerInner>,
}

struct TrackedPublishTrackerInner {
    entries: Mutex<HashMap<u64, Arc<TrackedPublishEntry>>>,
    closed: AtomicBool,
    wakeups: Notify,
    timeout_worker_started: AtomicBool,
}

impl std::fmt::Debug for TrackedPublishTracker {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let pending = self.inner.entries.lock().len();
        f.debug_struct("TrackedPublishTracker")
            .field("pending", &pending)
            .finish()
    }
}

impl Default for TrackedPublishTracker {
    fn default() -> Self {
        Self::new()
    }
}

impl TrackedPublishTracker {
    /// Create a new empty tracked publish tracker.
    #[must_use]
    pub fn new() -> Self {
        Self {
            inner: Arc::new(TrackedPublishTrackerInner {
                entries: Mutex::new(HashMap::new()),
                closed: AtomicBool::new(false),
                wakeups: Notify::new(),
                timeout_worker_started: AtomicBool::new(false),
            }),
        }
    }

    /// Register one tracked publish and return its receipt.
    ///
    /// `timeout` starts once this method admits the tracked publish into the
    /// tracker. If no terminal outcome is resolved before that deadline, the
    /// receipt resolves as [`TrackedPublishOutcome::TimedOut`].
    ///
    /// If the tracker was already closed, the returned receipt resolves
    /// immediately as [`TrackedPublishOutcome::TopicClosed`].
    pub fn register(
        &self,
        message_id: u64,
        timeout: Duration,
        permit: TrackedPublishPermit,
    ) -> TrackedPublishReceipt {
        if self.inner.closed.load(Ordering::Acquire) {
            let entry = Arc::new(TrackedPublishEntry::new(Instant::now(), permit));
            let _resolved = entry.resolve(TrackedPublishOutcome::TopicClosed);
            return TrackedPublishReceipt::new(message_id, entry);
        }

        self.ensure_timeout_worker();
        let entry = Arc::new(TrackedPublishEntry::new(Instant::now() + timeout, permit));
        let replaced = self
            .inner
            .entries
            .lock()
            .insert(message_id, Arc::clone(&entry));
        if replaced.is_some() {
            otel_warn!(
                "topic.tracked_publish.duplicate_message_id",
                message_id = message_id,
                message = "Tracked publish tracker registered a duplicate message id and overwrote the previous entry"
            );
        }
        self.inner.wakeups.notify_one();
        TrackedPublishReceipt::new(message_id, entry)
    }

    /// Register one tracked publish that resolves via `all`-mode consensus.
    ///
    /// The entry resolves as Ack only once every subscriber in `pending` has
    /// acked (via [`TrackedPublishTracker::resolve_ack_from`]); a Nack or a
    /// required subscriber disappearing (via
    /// [`TrackedPublishTracker::nack_pending_for_subscriber`]) resolves it as
    /// Nack.
    ///
    /// An empty `pending` set resolves immediately as Nack without registering
    /// or arming a timeout, preventing a publish with no eligible subscribers
    /// from reporting successful delivery.
    ///
    /// As with [`TrackedPublishTracker::register`], a closed tracker resolves the
    /// returned receipt immediately as [`TrackedPublishOutcome::TopicClosed`].
    pub fn register_consensus(
        &self,
        message_id: u64,
        timeout: Duration,
        permit: TrackedPublishPermit,
        members: impl Into<Arc<HashSet<BroadcastSubscriberId>>>,
        seq: u64,
    ) -> TrackedPublishReceipt {
        let members = members.into();
        if self.inner.closed.load(Ordering::Acquire) {
            let entry = Arc::new(TrackedPublishEntry::new_consensus(
                Instant::now(),
                permit,
                members,
                seq,
            ));
            let _resolved = entry.resolve(TrackedPublishOutcome::TopicClosed);
            return TrackedPublishReceipt::new(message_id, entry);
        }

        if members.is_empty() {
            let entry = Arc::new(TrackedPublishEntry::new_consensus(
                Instant::now(),
                permit,
                members,
                seq,
            ));
            let _resolved = entry.resolve(TrackedPublishOutcome::Nack {
                reason: Arc::from("broadcast publish had no eligible subscribers"),
            });
            return TrackedPublishReceipt::new(message_id, entry);
        }

        self.ensure_timeout_worker();
        let entry = Arc::new(TrackedPublishEntry::new_consensus(
            Instant::now() + timeout,
            permit,
            members,
            seq,
        ));
        let replaced = self
            .inner
            .entries
            .lock()
            .insert(message_id, Arc::clone(&entry));
        if replaced.is_some() {
            otel_warn!(
                "topic.tracked_publish.duplicate_message_id",
                message_id = message_id,
                message = "Tracked publish tracker registered a duplicate message id and overwrote the previous entry"
            );
        }
        self.inner.wakeups.notify_one();
        TrackedPublishReceipt::new(message_id, entry)
    }

    /// Apply a single broadcast subscriber's Ack to an `all`-mode tracked publish.
    ///
    /// Returns [`AckFromResult::Resolved`] if this Ack completed the consensus (the
    /// entry resolved as Ack and was removed), [`AckFromResult::StillPending`] if
    /// the Ack was recorded but other subscribers are still required, or
    /// [`AckFromResult::NotTracked`] if the message id is unknown, already
    /// resolved, or not a consensus (`all`-mode) entry. First-wins (`First`-kind)
    /// publishes must be resolved via [`TrackedPublishTracker::resolve`] instead.
    ///
    /// The tracker lock is held only for lookup and terminal removal. Ack
    /// bookkeeping runs under the entry lock so unrelated messages do not
    /// serialize on the tracker-wide map mutex.
    #[must_use]
    pub fn resolve_ack_from(
        &self,
        message_id: u64,
        subscriber_id: BroadcastSubscriberId,
    ) -> AckFromResult {
        let entry = {
            let entries = self.inner.entries.lock();
            let Some(entry) = entries.get(&message_id).cloned() else {
                return AckFromResult::NotTracked;
            };
            entry
        };
        let result = entry.resolve_ack_from(subscriber_id);
        if matches!(result, AckFromResult::Resolved) {
            let mut entries = self.inner.entries.lock();
            if entries
                .get(&message_id)
                .is_some_and(|current| Arc::ptr_eq(current, &entry))
            {
                let _ = entries.remove(&message_id);
                drop(entries);
                self.inner.wakeups.notify_one();
            }
        }
        result
    }

    /// Apply a single broadcast subscriber's Nack to a consensus (`all`-mode)
    /// entry, resolving the publish as Nack **only if** the subscriber is still
    /// required (present in the pending set).
    ///
    /// Returns [`NackFromResult::Resolved`] if the Nack resolved the entry (and
    /// it was removed), [`NackFromResult::NotRequired`] if the message is tracked
    /// but the subscriber no longer requires it (it already acked or was never
    /// eligible), or [`NackFromResult::NotTracked`] if the message id is unknown,
    /// already resolved, or not a consensus (`all`-mode) entry. First-wins
    /// (`First`-kind) publishes must be resolved via
    /// [`TrackedPublishTracker::resolve`] instead.
    ///
    /// This guard keeps per-subscriber signalling idempotent and order
    /// insensitive: a subscriber that has already acked cannot retract its
    /// contribution with a later (e.g. duplicated) Nack. It mirrors the
    /// disappearance handling in [`TrackedPublishTracker::nack_pending_for_subscriber`],
    /// which also only nacks entries that still require the subscriber.
    ///
    /// Lock order is tracker map -> entry (matching the timeout worker).
    #[must_use]
    pub fn resolve_nack_from(
        &self,
        message_id: u64,
        subscriber_id: BroadcastSubscriberId,
        reason: Arc<str>,
    ) -> NackFromResult {
        let mut entries = self.inner.entries.lock();
        let Some(entry) = entries.get(&message_id).cloned() else {
            return NackFromResult::NotTracked;
        };
        match entry.nack_if_requires(subscriber_id, &reason) {
            NackFromResult::Resolved => {
                let _ = entries.remove(&message_id);
                drop(entries);
                self.inner.wakeups.notify_one();
                NackFromResult::Resolved
            }
            NackFromResult::NotRequired => NackFromResult::NotRequired,
            NackFromResult::NotTracked => NackFromResult::NotTracked,
        }
    }

    /// Resolve every pending `all`-mode entry that still requires
    /// `subscriber_id` as Nack with `reason`.
    ///
    /// Called when a required subscriber disappears (lag-disconnect or
    /// drop/close) before acking outstanding messages. Idempotent and a no-op
    /// for entries that do not require the subscriber.
    pub fn nack_pending_for_subscriber(
        &self,
        subscriber_id: BroadcastSubscriberId,
        reason: Arc<str>,
    ) {
        let mut resolved_any = false;
        self.inner.entries.lock().retain(|_, entry| {
            let resolved = matches!(
                entry.nack_if_requires(subscriber_id, &reason),
                NackFromResult::Resolved
            );
            if resolved {
                resolved_any = true;
            }
            !resolved
        });
        if resolved_any {
            self.inner.wakeups.notify_one();
        }
    }

    /// Resolve every pending `all`-mode entry that still requires `subscriber_id`
    /// **and** was published before `seq_threshold` as Nack with `reason`.
    ///
    /// Called when a subscriber lags under the `DropOldest` policy: it stays
    /// connected but the messages it skipped (ring sequence `< seq_threshold`)
    /// were overwritten before it could read them, so it can never ack them.
    /// Entries at or after `seq_threshold` are left pending - the subscriber can
    /// still read and ack those. Idempotent and a no-op for unrelated entries.
    pub fn nack_owed_before(
        &self,
        subscriber_id: BroadcastSubscriberId,
        seq_threshold: u64,
        reason: Arc<str>,
    ) {
        let mut resolved_any = false;
        self.inner.entries.lock().retain(|_, entry| {
            let resolved = entry.nack_if_requires_before(subscriber_id, seq_threshold, &reason);
            if resolved {
                resolved_any = true;
            }
            !resolved
        });
        if resolved_any {
            self.inner.wakeups.notify_one();
        }
    }

    /// Resolve one tracked publish by message id.
    ///
    /// Returns `true` if this call resolved a pending publish. Returns `false`
    /// if the message id was not tracked or had already been resolved.
    #[must_use]
    pub fn resolve(&self, message_id: u64, outcome: TrackedPublishOutcome) -> bool {
        let Some(entry) = self.inner.entries.lock().remove(&message_id) else {
            return false;
        };
        let resolved = entry.resolve(outcome);
        self.inner.wakeups.notify_one();
        resolved
    }

    /// Discard one tracked publish without producing an outcome.
    ///
    /// This releases the in-flight slot associated with the publish.
    #[must_use]
    pub fn discard(&self, message_id: u64) -> bool {
        let removed = self.inner.entries.lock().remove(&message_id).is_some();
        if removed {
            self.inner.wakeups.notify_one();
        }
        removed
    }

    /// Resolve all pending tracked publishes as
    /// [`TrackedPublishOutcome::TopicClosed`].
    pub fn close_all(&self) {
        self.inner.closed.store(true, Ordering::Release);
        let drained = {
            self.inner
                .entries
                .lock()
                .drain()
                .map(|(_, entry)| entry)
                .collect::<Vec<_>>()
        };
        for entry in drained {
            let _resolved = entry.resolve(TrackedPublishOutcome::TopicClosed);
        }
        self.inner.wakeups.notify_waiters();
    }

    fn ensure_timeout_worker(&self) {
        if self
            .inner
            .timeout_worker_started
            .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
            .is_ok()
        {
            let tracker = self.clone();
            _ = tokio::spawn(async move {
                tracker.timeout_worker().await;
            });
        }
    }

    async fn timeout_worker(self) {
        loop {
            if self.inner.closed.load(Ordering::Acquire) && self.inner.entries.lock().is_empty() {
                break;
            }

            let next_deadline = self.next_deadline();
            match next_deadline {
                Some(deadline) => {
                    tokio::select! {
                        _ = sleep_until(deadline) => self.resolve_expired(deadline),
                        _ = self.inner.wakeups.notified() => {}
                    }
                }
                None => {
                    self.inner.wakeups.notified().await;
                }
            }
        }
    }

    fn next_deadline(&self) -> Option<Instant> {
        self.inner
            .entries
            .lock()
            .values()
            .filter(|entry| entry.is_pending())
            .map(|entry| entry.deadline())
            .min()
    }

    fn resolve_expired(&self, now: Instant) {
        let expired = {
            let mut entries = self.inner.entries.lock();
            let expired_ids: Vec<u64> = entries
                .iter()
                .filter(|(_, entry)| entry.is_pending() && entry.deadline() <= now)
                .map(|(id, _)| *id)
                .collect();
            expired_ids
                .into_iter()
                .filter_map(|id| entries.remove(&id))
                .collect::<Vec<_>>()
        };

        for entry in expired {
            let _resolved = entry.resolve(TrackedPublishOutcome::TimedOut);
        }
    }
}

/// Default configuration used when creating a tracked publisher from a topic handle.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TopicPublishOutcomeConfig {
    /// Maximum number of unresolved tracked publishes per tracked publisher.
    ///
    /// This limit is enforced before entering the topic runtime by requiring a
    /// caller-owned permit for each tracked publish.
    pub max_in_flight: usize,
    /// Maximum time to wait before a tracked publish resolves as timed out.
    ///
    /// The timer starts once the topic accepts the tracked publish.
    pub timeout: Duration,
}

impl Default for TopicPublishOutcomeConfig {
    fn default() -> Self {
        Self {
            max_in_flight: 1024,
            timeout: Duration::from_secs(30),
        }
    }
}

/// Receipt returned by tracked publish calls.
///
/// The receipt can be awaited for its terminal outcome via
/// [`TrackedPublishReceipt::wait_for_outcome`]. Dropping the receipt does not
/// cancel the underlying tracked publish or release its in-flight slot early;
/// the slot is released only when the publish reaches a terminal outcome or
/// times out.
pub struct TrackedPublishReceipt {
    message_id: u64,
    entry: Arc<TrackedPublishEntry>,
}

impl std::fmt::Debug for TrackedPublishReceipt {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TrackedPublishReceipt")
            .field("message_id", &self.message_id)
            .finish()
    }
}

impl TrackedPublishReceipt {
    pub(crate) fn new(message_id: u64, entry: Arc<TrackedPublishEntry>) -> Self {
        Self { message_id, entry }
    }

    /// Broker-assigned message identifier.
    #[must_use]
    pub const fn message_id(&self) -> u64 {
        self.message_id
    }

    /// Wait for the terminal tracked outcome.
    ///
    /// Consuming the receipt only waits for the already-tracked outcome. It
    /// does not own the tracked publish lifecycle, so dropping the receipt
    /// before awaiting does not cancel tracking.
    pub async fn wait_for_outcome(self) -> TrackedPublishOutcome {
        self.entry.wait_for_outcome().await
    }
}

/// Unique identifier for a broadcast subscriber within a single topic.
///
/// Allocated by the broadcast-only topic's subscriber registry and used by the
/// tracked publish tracker to drive `all`-mode consensus aggregation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct BroadcastSubscriberId(pub u64);

/// Outcome of applying a single broadcast subscriber's Ack to a tracked publish
/// in `all` (consensus) mode.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AckFromResult {
    /// This Ack completed the consensus; the entry resolved as Ack.
    Resolved,
    /// The Ack was recorded but other required subscribers are still pending.
    StillPending,
    /// The message was not tracked (already resolved, timed out, or unknown id).
    NotTracked,
}

/// Outcome of applying a single broadcast subscriber's Nack to a tracked publish
/// in `all` (consensus) mode.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum NackFromResult {
    /// The subscriber still required the message; the entry resolved as Nack.
    Resolved,
    /// The message is tracked but the subscriber no longer requires it (it has
    /// already acked, or was never in the eligible set), so the Nack is a no-op.
    NotRequired,
    /// The message was not tracked (already resolved, timed out, or unknown id).
    NotTracked,
}

#[derive(Debug, Clone)]
enum PendingKind {
    /// The first terminal resolution wins. Used by normal tracked publishes and
    /// by `first`-mode broadcast. The tracker does not track per-subscriber acks.
    First,
    /// Consensus aggregation: the entry resolves as Ack only once every subscriber
    /// in `members` has acked. Any Nack or required-subscriber disappearance
    /// resolves it as Nack.
    All {
        /// Immutable set of subscribers eligible at publish time.
        members: Arc<HashSet<BroadcastSubscriberId>>,
        /// Subscribers that have acked so far. The entry resolves as Ack once
        /// `acked` covers every member. Grows lazily as acks arrive.
        acked: HashSet<BroadcastSubscriberId>,
        /// Broadcast ring sequence used for lag processing.
        seq: u64,
    },
}

#[derive(Debug, Clone)]
enum TrackedPublishState {
    Pending(PendingKind),
    Resolved(TrackedPublishOutcome),
}

/// Internal shared state for one tracked publish receipt.
pub(crate) struct TrackedPublishEntry {
    state: Mutex<TrackedPublishState>,
    deadline: Instant,
    permit: Mutex<Option<TrackedPublishPermit>>,
    notify: Notify,
}

impl TrackedPublishEntry {
    pub(crate) fn new(deadline: Instant, permit: TrackedPublishPermit) -> Self {
        Self {
            state: Mutex::new(TrackedPublishState::Pending(PendingKind::First)),
            deadline,
            permit: Mutex::new(Some(permit)),
            notify: Notify::new(),
        }
    }

    pub(crate) fn new_consensus(
        deadline: Instant,
        permit: TrackedPublishPermit,
        members: Arc<HashSet<BroadcastSubscriberId>>,
        seq: u64,
    ) -> Self {
        Self {
            state: Mutex::new(TrackedPublishState::Pending(PendingKind::All {
                members,
                acked: HashSet::new(),
                seq,
            })),
            deadline,
            permit: Mutex::new(Some(permit)),
            notify: Notify::new(),
        }
    }

    #[must_use]
    pub(crate) fn deadline(&self) -> Instant {
        self.deadline
    }

    #[must_use]
    pub(crate) fn is_pending(&self) -> bool {
        matches!(*self.state.lock(), TrackedPublishState::Pending(_))
    }

    pub(crate) fn resolve(&self, outcome: TrackedPublishOutcome) -> bool {
        let mut state = self.state.lock();
        if matches!(*state, TrackedPublishState::Resolved(_)) {
            return false;
        }
        *state = TrackedPublishState::Resolved(outcome);
        drop(state);
        _ = self.permit.lock().take();
        self.notify.notify_waiters();
        true
    }

    /// Apply a single subscriber's Ack to a consensus (`all`-mode) entry.
    ///
    /// Removes `subscriber_id` from the pending set; when the set empties the
    /// entry resolves as Ack. This is the consensus (`all`-mode) path only: a
    /// `First`-kind entry returns [`AckFromResult::NotTracked`] (first-wins
    /// callers must use the tracker's `resolve(message_id, Ack)` instead).
    /// Already-resolved entries are left untouched. Idempotent: a repeated Ack
    /// from the same subscriber never falsely completes the consensus.
    pub(crate) fn resolve_ack_from(&self, subscriber_id: BroadcastSubscriberId) -> AckFromResult {
        let mut state = self.state.lock();
        match &mut *state {
            TrackedPublishState::Resolved(_) => AckFromResult::NotTracked,
            TrackedPublishState::Pending(PendingKind::First) => AckFromResult::NotTracked,
            TrackedPublishState::Pending(PendingKind::All { members, acked, .. }) => {
                if !members.contains(&subscriber_id) {
                    return AckFromResult::StillPending;
                }
                let _ = acked.insert(subscriber_id);
                if acked.len() == members.len() {
                    *state = TrackedPublishState::Resolved(TrackedPublishOutcome::Ack);
                    drop(state);
                    _ = self.permit.lock().take();
                    self.notify.notify_waiters();
                    AckFromResult::Resolved
                } else {
                    AckFromResult::StillPending
                }
            }
        }
    }

    /// Resolve a consensus (`all`-mode) entry as Nack if it still requires
    /// `subscriber_id`.
    ///
    /// Used when a required subscriber disappears (lag-disconnect or drop/close)
    /// before acking. No-op for `First`-kind, already-resolved, or unrelated
    /// entries. `reason` is only cloned when the entry actually resolves, so
    /// scanning many entries that do not require the subscriber is allocation
    /// free.
    fn nack_if_requires(
        &self,
        subscriber_id: BroadcastSubscriberId,
        reason: &Arc<str>,
    ) -> NackFromResult {
        let mut state = self.state.lock();
        match &*state {
            TrackedPublishState::Resolved(_) | TrackedPublishState::Pending(PendingKind::First) => {
                return NackFromResult::NotTracked;
            }
            TrackedPublishState::Pending(PendingKind::All { members, acked, .. })
                if !members.contains(&subscriber_id) || acked.contains(&subscriber_id) =>
            {
                return NackFromResult::NotRequired;
            }
            TrackedPublishState::Pending(PendingKind::All { .. }) => {}
        }
        *state = TrackedPublishState::Resolved(TrackedPublishOutcome::Nack {
            reason: Arc::clone(reason),
        });
        drop(state);
        _ = self.permit.lock().take();
        self.notify.notify_waiters();
        NackFromResult::Resolved
    }

    /// Resolve a consensus (`all`-mode) entry as Nack if it still requires
    /// `subscriber_id` **and** its publish sequence is `< seq_threshold`.
    /// Returns `true` if this call resolved the entry.
    ///
    /// Used when a subscriber lags under `DropOldest`: only the messages it
    /// skipped (published before the new read position) are nacked; messages it
    /// can still read are left pending. No-op for `First`-kind, already-resolved,
    /// or unrelated entries. `reason` is only cloned when the entry actually
    /// resolves.
    pub(crate) fn nack_if_requires_before(
        &self,
        subscriber_id: BroadcastSubscriberId,
        seq_threshold: u64,
        reason: &Arc<str>,
    ) -> bool {
        let mut state = self.state.lock();
        let requires = matches!(
            &*state,
            TrackedPublishState::Pending(PendingKind::All { members, acked, seq })
                if *seq < seq_threshold
                    && members.contains(&subscriber_id)
                    && !acked.contains(&subscriber_id)
        );
        if !requires {
            return false;
        }
        *state = TrackedPublishState::Resolved(TrackedPublishOutcome::Nack {
            reason: Arc::clone(reason),
        });
        drop(state);
        _ = self.permit.lock().take();
        self.notify.notify_waiters();
        true
    }

    async fn wait_for_outcome(&self) -> TrackedPublishOutcome {
        loop {
            let notified = self.notify.notified();
            if let Some(outcome) = self.current_outcome() {
                return outcome;
            }
            notified.await;
        }
    }

    fn current_outcome(&self) -> Option<TrackedPublishOutcome> {
        match &*self.state.lock() {
            TrackedPublishState::Pending(_) => None,
            TrackedPublishState::Resolved(outcome) => Some(outcome.clone()),
        }
    }
}

/// Subscription mode chosen by the subscriber.
#[derive(Debug, Clone)]
pub enum SubscriptionMode {
    /// Balanced (consumer-group) mode: each message goes to exactly one subscriber in the group.
    Balanced {
        /// Consumer-group identifier.
        group: SubscriptionGroupName,
    },
    /// Broadcast mode: each subscriber gets every message with configurable lag behavior.
    Broadcast,
}

/// Options provided at subscription time.
#[derive(Debug, Clone, Default)]
pub struct SubscriberOptions {}

/// Options for creating/opening a topic.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TopicOptions {
    /// Only balanced (consumer-group) subscriptions.
    BalancedOnly {
        /// Capacity for balanced consumer-group channels.
        capacity: usize,
    },
    /// Only broadcast subscriptions.
    BroadcastOnly {
        /// Ring-buffer capacity for broadcast subscribers.
        capacity: usize,
        /// Behavior when a broadcast subscriber falls behind the retained ring window.
        on_lag: TopicBroadcastOnLagPolicy,
        /// Aggregation mode for tracked broadcast Ack/Nack resolution.
        ack_mode: TopicBroadcastAckMode,
    },
    /// Both balanced and broadcast subscriptions.
    Mixed {
        /// Capacity for balanced consumer-group channels.
        balanced_capacity: usize,
        /// Ring-buffer capacity for broadcast subscribers.
        broadcast_capacity: usize,
        /// Behavior when a broadcast subscriber falls behind the retained ring window.
        on_lag: TopicBroadcastOnLagPolicy,
        /// Aggregation mode for tracked broadcast Ack/Nack resolution.
        ack_mode: TopicBroadcastAckMode,
    },
}

impl TopicOptions {
    /// Default balanced channel capacity.
    pub const DEFAULT_BALANCED_CAPACITY: usize = 1024;
    /// Default broadcast ring-buffer capacity.
    pub const DEFAULT_BROADCAST_CAPACITY: usize = 1024;
}

impl Default for TopicOptions {
    fn default() -> Self {
        Self::Mixed {
            balanced_capacity: TopicOptions::DEFAULT_BALANCED_CAPACITY,
            broadcast_capacity: TopicOptions::DEFAULT_BROADCAST_CAPACITY,
            on_lag: TopicBroadcastOnLagPolicy::DropOldest,
            ack_mode: TopicBroadcastAckMode::First,
        }
    }
}