ruststream-pulsar 0.7.1

//! [`PulsarSubscription`]: the subscription descriptor.
//!
//! The subscription type is an enum, not a set of sibling flags, so combinations that do not
//! exist are unrepresentable; the ack timeout and the dead-letter policy the registration
//! declares are consumer-side settings the Pulsar client owns.

use std::collections::HashMap;
use std::sync::{Mutex, MutexGuard};
use std::time::Duration;

#[cfg(feature = "asyncapi")]
use ruststream::asyncapi::Bindings;
use ruststream::runtime::IntoSource;
use ruststream::{BrokerMoves, DeclareRetryError, RetryDeclaration, SubscriptionSource};
#[cfg(feature = "asyncapi")]
use serde::Serialize;

#[cfg(feature = "asyncapi")]
use crate::bindings::{extension, topic_channel};
use crate::broker::ConnectedPulsarBroker;
use crate::error::PulsarError;
use crate::pattern::TopicPattern;
use crate::subscriber::PulsarSubscriber;
use crate::topic::PulsarTopic;

/// How competing consumers on one subscription share its messages.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SubscriptionType {
    /// One consumer holds the subscription.
    ///
    /// A server refuses a second attach as busy, and the client retries it until
    /// [`OperationRetries`](crate::OperationRetries) runs out, which by default is never: a second
    /// consumer of this subscription is a service that does not start. A bound on the retries
    /// turns the wait into [`PulsarError::Subscribe`](crate::PulsarError::Subscribe). In process,
    /// under the test harness, the attach is refused with that error at once, so the test fails
    /// rather than hangs.
    Exclusive,
    /// Competing consumers, round-robin. The default.
    #[default]
    Shared,
    /// One active consumer with hot standbys.
    Failover,
    /// Competing consumers with per-key ordering (the `Partitioned` capability's transport).
    KeyShared,
}

impl SubscriptionType {
    /// Whether the broker counts a message's redeliveries on a subscription of this type.
    ///
    /// It counts them where the subscription dispatches to several consumers, and nowhere else:
    /// a subscription with one active consumer keeps no redelivery tracker, so every delivery
    /// arrives with a count of zero. A dead-letter policy is that count against a limit, so on
    /// [`Exclusive`](Self::Exclusive) and [`Failover`](Self::Failover) the limit is never
    /// reached and a spent message circles the subscription instead of moving on.
    pub(crate) const fn counts_redeliveries(self) -> bool {
        matches!(self, Self::Shared | Self::KeyShared)
    }
}

#[cfg(feature = "asyncapi")]
impl SubscriptionType {
    /// Pulsar's own spelling, which is what the generated document reports.
    pub(crate) const fn as_str(self) -> &'static str {
        match self {
            Self::Exclusive => "Exclusive",
            Self::Shared => "Shared",
            Self::Failover => "Failover",
            Self::KeyShared => "KeyShared",
        }
    }
}

/// What a Pulsar consumer is, in the crate's own vocabulary.
///
/// The specification's `pulsar` operation object is empty, so the subscription's name, its type
/// and its acknowledgement timeout travel under an extension key instead of being squeezed into
/// a field that means something else.
#[cfg(feature = "asyncapi")]
#[derive(Serialize)]
struct PulsarOperation<'a> {
    subscription: &'a str,
    #[serde(rename = "subscriptionType")]
    subscription_type: &'static str,
    #[serde(rename = "ackTimeoutMillis", skip_serializing_if = "Option::is_none")]
    ack_timeout_millis: Option<u64>,
    /// The topics a multi-topic subscription reads. Absent for a single topic, which the
    /// channel address already names, and for a pattern, which names none.
    #[serde(skip_serializing_if = "Option::is_none")]
    topics: Option<&'a [String]>,
    /// The regular expression a pattern subscription selects its topics with.
    #[serde(rename = "topicPattern", skip_serializing_if = "Option::is_none")]
    topic_pattern: Option<&'a str>,
}

/// The consumer's dead-letter policy, as the registration declared it: a delivery limit and the
/// topic a spent delivery moves to.
///
/// Both halves together, because that is what the Pulsar client applies. The live consumer and
/// the in-process transport read this one value, so a cap driven in a test is the cap production
/// configures.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct DeadLetterRoute {
    pub(crate) topic: String,
    /// How many times the message is handed to a handler before it moves on.
    pub(crate) max_deliveries: u32,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum Topics {
    List(Vec<String>),
    Pattern(String),
}

/// How long a partial batch waits for more deliveries before the handler sees it, unless the
/// descriptor names another value.
///
/// Short, because the cost of waiting is latency on a batch that is already useful; a service
/// trading latency for fuller batches raises it with [`PulsarSubscription::batch_wait`].
pub(crate) const DEFAULT_BATCH_WAIT: Duration = Duration::from_millis(10);

/// A subscription descriptor for one Pulsar subscription over one or more topics.
///
/// It names the subscription, the topics it reads and the settings its consumer opens with:
/// [`new`](Self::new) reads one topic, [`topics`](Self::topics) a fixed list and
/// [`pattern`](Self::pattern) every topic whose full name a regular expression matches. Where the
/// subscription starts reading is the framework's `start_at(..)` clause over
/// [`PulsarPosition`](crate::PulsarPosition).
///
/// The registration declares the retry cap and the dead-letter topic at the mount site with
/// `b.include(handler).max_attempts(nonzero!(5)).dead_letter("orders-dlq")`, and this descriptor
/// turns the declaration into the consumer's own `DeadLetterPolicy`. The Pulsar client moves a
/// spent delivery itself, so the registration takes no `out_retry(..)` step. The cap needs a
/// subscription that dispatches to competing consumers - Pulsar counts a message's redeliveries
/// on [`Shared`](SubscriptionType::Shared) and [`KeyShared`](SubscriptionType::KeyShared) - so a
/// declaration over the other two types refuses to start.
///
/// # Examples
///
/// Implements [`SubscriptionSource`] for the broker, so the declaration sits inline in the
/// `#[subscriber(..)]` decorator:
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_pulsar::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(
///     PulsarSubscription::new("orders", "workers")
///         .subscription_type(SubscriptionType::Shared)
///         .ack_timeout(Duration::from_secs(30))
/// )]
/// async fn handle(order: &Order) -> HandlerOutcome {
///     tracing::info!(order.id, "accepted");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         PulsarBroker::new("pulsar://localhost:6650"),
///         |b| {
///             b.include(handle).max_attempts(nonzero!(5)).dead_letter("orders-dlq");
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
///
/// The three forms side by side, one handler each:
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_pulsar::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(PulsarSubscription::new("persistent://acme/orders/created", "workers"))]
/// async fn fulfil(order: &Order) -> HandlerOutcome {
///     tracing::info!(order.id, "fulfilling");
///     HandlerOutcome::ack()
/// }
///
/// #[subscriber(PulsarSubscription::topics(["orders-eu", "orders-us"], "regional"))]
/// async fn tally(order: &Order) -> HandlerOutcome {
///     tracing::info!(order.id, "counted");
///     HandlerOutcome::ack()
/// }
///
/// #[subscriber(
///     PulsarSubscription::pattern("persistent://public/default/orders-.*", "audit")
///         .subscription_type(SubscriptionType::KeyShared)
///         .ack_timeout(Duration::from_secs(30))
/// )]
/// async fn audit(order: &Order) -> HandlerOutcome {
///     tracing::info!(order.id, "audited");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(
///         PulsarBroker::new("pulsar://localhost:6650"),
///         |b| {
///             b.include(fulfil);
///             b.include(tally);
///             b.include(audit);
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
#[must_use]
pub struct PulsarSubscription {
    pub(crate) topics: Topics,
    pub(crate) subscription: String,
    pub(crate) sub_type: SubscriptionType,
    pub(crate) ack_timeout: Option<Duration>,
    pub(crate) batch_wait: Duration,
    /// What the registration declared about its retries, taken in before the consumer opens.
    retry: RetryDeclaration,
}

impl PulsarSubscription {
    /// Subscribes `subscription` to `topic` (a bare name, a `tenant/namespace/topic` triple,
    /// or a fully qualified name).
    pub fn new(topic: impl Into<String>, subscription: impl Into<String>) -> Self {
        Self {
            topics: Topics::List(vec![topic.into()]),
            subscription: subscription.into(),
            sub_type: SubscriptionType::default(),
            ack_timeout: None,
            batch_wait: DEFAULT_BATCH_WAIT,
            retry: RetryDeclaration::new(),
        }
    }

    /// Subscribes to every topic in the list.
    pub fn topics<I, S>(topics: I, subscription: impl Into<String>) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        Self {
            topics: Topics::List(topics.into_iter().map(Into::into).collect()),
            ..Self::new(String::new(), subscription)
        }
    }

    /// Subscribes to every topic of one namespace whose fully qualified name matches `pattern`,
    /// a regular expression.
    ///
    /// The namespace is the one the pattern's prefix names: `persistent://acme/orders/.*` and
    /// `acme/orders/.*` read `acme/orders`, and a pattern with no `tenant/namespace/` part, such
    /// as `orders-.*`, reads `public/default`. The scheme is part of the regular expression, so
    /// one pattern may select persistent and non-persistent topics alike. The pattern is checked
    /// on subscribe: one that is no regular expression, or whose tenant or namespace is itself a
    /// regular expression and so names no single namespace, refuses to start with
    /// [`PulsarError::Invalid`].
    pub fn pattern(pattern: impl Into<String>, subscription: impl Into<String>) -> Self {
        Self {
            topics: Topics::Pattern(pattern.into()),
            ..Self::new(String::new(), subscription)
        }
    }

    /// Sets the subscription type. Defaults to [`SubscriptionType::Shared`].
    ///
    /// Only the two types that dispatch to competing consumers carry a retry cap: the broker
    /// counts a message's redeliveries there and nowhere else, so a registration that declares
    /// `max_attempts(..)` over an [`Exclusive`](SubscriptionType::Exclusive) or a
    /// [`Failover`](SubscriptionType::Failover) subscription refuses to start.
    pub fn subscription_type(mut self, sub_type: SubscriptionType) -> Self {
        self.sub_type = sub_type;
        self
    }

    /// Redelivers messages that stay unacknowledged longer than `timeout`.
    pub fn ack_timeout(mut self, timeout: Duration) -> Self {
        self.ack_timeout = Some(timeout);
        self
    }

    /// Caps how long a partial batch waits for more deliveries after its first one.
    ///
    /// Only a batch handler observes this: the client hands over one delivery at a time, so a
    /// `batch(n)` registration on this subscription assembles its batches here, and a batch
    /// closes when it holds `n` deliveries or when this has elapsed, whichever comes first.
    /// Defaults to 10 ms; raising it trades latency for fuller batches on a sparse topic.
    ///
    /// # Examples
    ///
    /// ```
    /// # mod demo {
    /// use std::time::Duration;
    ///
    /// use ruststream_pulsar::prelude::*;
    /// use serde::Deserialize;
    ///
    /// #[derive(Deserialize)]
    /// struct Reading {
    ///     value: f64,
    /// }
    ///
    /// #[subscriber(
    ///     PulsarSubscription::new("readings", "aggregator")
    ///         .batch_wait(Duration::from_millis(250))
    /// )]
    /// async fn aggregate(readings: &[Reading]) -> HandlerOutcome {
    ///     let total: f64 = readings.iter().map(|reading| reading.value).sum();
    ///     tracing::info!(count = readings.len(), total, "readings settled");
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("readings", "0.1.0")).with_broker(
    ///         PulsarBroker::new("pulsar://localhost:6650"),
    ///         |b| {
    ///             b.include(aggregate.batch(nonzero!(64)));
    ///         },
    ///     )
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    pub fn batch_wait(mut self, wait: Duration) -> Self {
        self.batch_wait = wait;
        self
    }

    /// The subscription name.
    #[must_use]
    pub fn subscription(&self) -> &str {
        &self.subscription
    }

    /// Takes in what a registration declared about its retries. What the [`SubscriptionSource`]
    /// impl does with the declaration, and what a bare topic name's descriptor is built with.
    pub(crate) fn declaring(mut self, declaration: &RetryDeclaration) -> Self {
        self.retry = declaration.clone();
        self
    }

    /// The name the framework reports for a subscription on this descriptor: the topic when
    /// there is exactly one, the subscription name otherwise, since a list and a pattern have no
    /// single topic to name. The test harness's routing answer reads it too, so a subscription
    /// is owed a publish under the name it is reported by.
    pub(crate) fn source_name(&self) -> &str {
        match &self.topics {
            Topics::List(topics) if topics.len() == 1 => &topics[0],
            _ => &self.subscription,
        }
    }

    /// The consumer's dead-letter policy, named in a refusal by the subscription this
    /// descriptor opens.
    pub(crate) fn dead_letter_policy(&self) -> Result<Option<DeadLetterRoute>, PulsarError> {
        let subject = format!("subscription '{}'", self.subscription);
        let route = resolve_dead_letter(&self.retry, &subject)?;
        if route.is_some() && !self.sub_type.counts_redeliveries() {
            let sharing = self.sub_type;
            return Err(PulsarError::Invalid(format!(
                "{subject} declares a retry cap on a {sharing:?} subscription: Pulsar counts a \
                 message's redeliveries only where a subscription dispatches to competing \
                 consumers, so on this one the cap is never reached and a spent delivery circles \
                 the subscription for ever. Declare the subscription \
                 SubscriptionType::Shared or SubscriptionType::KeyShared, or drop the cap and the \
                 dead-letter topic"
            )));
        }
        Ok(route)
    }

    /// What this subscription adds to its channel in the generated document.
    ///
    /// The namespace and the persistence come out of the topic name the descriptor was built
    /// with, which is the only place a consumer learns them. A subscription over several topics
    /// reports them only when every topic agrees, and a pattern subscription reports nothing:
    /// the specification's fields are single-valued, and a document that picks one topic's
    /// namespace to stand for all of them describes a deployment that does not exist.
    #[cfg(feature = "asyncapi")]
    fn describe_channel(&self) -> Bindings {
        let Topics::List(topics) = &self.topics else {
            return Bindings::new();
        };
        let mut parsed = topics.iter().map(|topic| PulsarTopic::parse(topic));
        let Some(Ok(first)) = parsed.next() else {
            return Bindings::new();
        };
        for rest in parsed {
            match rest {
                Ok(topic)
                    if topic.namespace() == first.namespace()
                        && topic.persistence() == first.persistence() => {}
                _ => return Bindings::new(),
            }
        }
        topic_channel(&first)
    }

    /// What this subscription adds to its `receive` operation.
    ///
    /// The specification's `pulsar` operation object is empty, so this is an extension: the
    /// subscription's name and type decide how competing consumers share the stream, and the
    /// acknowledgement timeout decides when an unsettled delivery comes back, all of which a
    /// reader of the document wants and no standard field carries.
    #[cfg(feature = "asyncapi")]
    fn describe_operation(&self) -> Bindings {
        let (topics, topic_pattern) = match &self.topics {
            Topics::List(topics) if topics.len() > 1 => (Some(topics.as_slice()), None),
            Topics::List(_) => (None, None),
            Topics::Pattern(pattern) => (None, Some(pattern.as_str())),
        };
        let body = PulsarOperation {
            subscription: &self.subscription,
            subscription_type: self.sub_type.as_str(),
            ack_timeout_millis: self
                .ack_timeout
                .map(|timeout| u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX)),
            topics,
            topic_pattern,
        };
        extension("x-ruststream-pulsar", &body)
    }

    pub(crate) fn display_topic(&self) -> String {
        match &self.topics {
            Topics::List(topics) => topics.join(","),
            Topics::Pattern(pattern) => pattern.clone(),
        }
    }

    /// Rejects descriptors that cannot form a subscription, before any I/O.
    pub(crate) fn validate(&self) -> Result<(), PulsarError> {
        if self.subscription.is_empty() {
            return Err(PulsarError::Invalid(
                "subscription name must be non-empty".into(),
            ));
        }
        match &self.topics {
            Topics::List(topics) => {
                if topics.is_empty() || topics.iter().any(String::is_empty) {
                    return Err(PulsarError::Invalid("topics must be non-empty".into()));
                }
                for topic in topics {
                    let _ = PulsarTopic::parse(topic)?;
                }
            }
            Topics::Pattern(pattern) => {
                TopicPattern::parse(pattern)?;
            }
        }
        self.dead_letter_policy()?;
        Ok(())
    }
}

/// The consumer's dead-letter policy, resolved from what a registration declared.
///
/// Pulsar's policy is a limit and a topic together - the client counts redeliveries and, past the
/// limit, produces the message to that topic and acknowledges the original - so half a
/// declaration is not a policy it can apply. A registration that declares one half refuses to
/// start rather than running with a cap nobody enforces.
///
/// `subject` is what the refusal calls the registration: a descriptor is known by the
/// subscription it names, a bare registration by the topic it opens.
fn resolve_dead_letter(
    retry: &RetryDeclaration,
    subject: &str,
) -> Result<Option<DeadLetterRoute>, PulsarError> {
    match (retry.max_attempts(), retry.dead_letter()) {
        (Some(attempts), Some(topic)) => {
            let _ = PulsarTopic::parse(topic)?;
            Ok(Some(DeadLetterRoute {
                topic: topic.to_owned(),
                max_deliveries: attempts.get(),
            }))
        }
        (None, None) => Ok(None),
        // The mount chain lets the two steps be written apart, so the pairing cannot be a
        // compile error here; the startup refusal is what keeps a half declaration from looking
        // applied.
        (Some(_), None) => Err(PulsarError::Invalid(format!(
            "{subject} declares max_attempts with no dead_letter: Pulsar applies a delivery \
             limit only together with the topic a spent delivery moves to, so declare \
             dead_letter(..) beside it or drop both"
        ))),
        (None, Some(topic)) => Err(PulsarError::Invalid(format!(
            "{subject} declares dead_letter('{topic}') with no max_attempts: Pulsar moves a \
             delivery to the dead-letter topic by counting redeliveries, so declare \
             max_attempts(..) beside it or drop both"
        ))),
    }
}

/// What registrations mounted by a bare topic name declared about their retries.
///
/// A descriptor carries its own declaration into the consumer it opens. A bare name has nothing
/// to carry it in: the framework hands the declaration to the broker in one call and asks it for
/// the subscription in another, so the connected form holds it in between, under the topic both
/// calls name. The two calls come in that order, one registration at a time, which is what makes
/// the topic enough to find it by.
#[derive(Debug, Default)]
pub(crate) struct DeclaredRetries {
    by_topic: Mutex<HashMap<String, RetryDeclaration>>,
}

impl DeclaredRetries {
    /// Takes the declaration a registration made over the bare name `topic`.
    ///
    /// # Errors
    ///
    /// Returns [`DeclareRetryError::Broker`] for half a policy: the Pulsar client counts
    /// redeliveries against a limit and produces the spent message to a topic, so neither half
    /// applies without the other.
    pub(crate) fn declare(
        &self,
        topic: &str,
        declaration: &RetryDeclaration,
    ) -> Result<(), DeclareRetryError> {
        // The same resolution the descriptor's own consumer is built from, so the two cannot
        // answer differently for one declaration. The refusal names the topic, because a bare
        // registration carries no subscription name of its own.
        resolve_dead_letter(declaration, &format!("topic '{topic}'"))
            .map_err(|err| DeclareRetryError::Broker(Box::new(err)))?;
        self.lock().insert(topic.to_owned(), declaration.clone());
        Ok(())
    }

    /// The descriptor a bare `topic` opens: the broker's default `subscription` over it,
    /// carrying what was declared for it.
    pub(crate) fn subscription(&self, topic: &str, subscription: &str) -> PulsarSubscription {
        let declared = self.lock().get(topic).cloned().unwrap_or_default();
        PulsarSubscription::new(topic, subscription).declaring(&declared)
    }

    /// Nothing here panics while the map is held, so the lock cannot have been poisoned.
    fn lock(&self) -> MutexGuard<'_, HashMap<String, RetryDeclaration>> {
        self.by_topic
            .lock()
            .expect("the declarations are read and written without panicking")
    }
}

// The manual path binds a `Handle` body to its source with `subscriber(source, body)`, which
// takes the descriptor as it is, the way the `#[subscriber(..)]` attribute does.
impl IntoSource for PulsarSubscription {
    type Source = Self;

    fn into_source(self) -> Self {
        self
    }
}

/// The Pulsar client moves a spent delivery itself: it counts the redeliveries of a message and,
/// past the consumer's limit, produces it to the dead-letter topic and acknowledges the original.
/// Nothing is republished from the service, on any of the three addressing forms, so
/// `out_retry(..)` over this descriptor is a compile error and the registration's declaration
/// reaches the consumer instead.
impl SubscriptionSource<ConnectedPulsarBroker> for PulsarSubscription {
    type Subscriber = PulsarSubscriber;
    type Copies = BrokerMoves;

    fn name(&self) -> &str {
        self.source_name()
    }

    async fn subscribe(
        self,
        connected: &ConnectedPulsarBroker,
    ) -> Result<PulsarSubscriber, PulsarError> {
        connected.subscribe_descriptor(self).await
    }

    /// Records the declaration; the consumer is built from it in `subscribe`, where the
    /// connection exists.
    fn declare_retry(self, declaration: &RetryDeclaration) -> Self {
        self.declaring(declaration)
    }

    #[cfg(feature = "asyncapi")]
    fn channel_bindings(&self) -> Bindings {
        self.describe_channel()
    }

    #[cfg(feature = "asyncapi")]
    fn operation_bindings(&self) -> Bindings {
        self.describe_operation()
    }
}

#[cfg(test)]
mod tests {
    use ruststream::nonzero;

    use super::*;

    /// Takes the declaration in the way the runtime does.
    fn declared(
        subscription: PulsarSubscription,
        declaration: &RetryDeclaration,
    ) -> PulsarSubscription {
        SubscriptionSource::declare_retry(subscription, declaration)
    }

    #[test]
    fn empty_subscription_is_rejected_before_io() {
        assert!(matches!(
            PulsarSubscription::new("orders", "").validate(),
            Err(PulsarError::Invalid(_))
        ));
    }

    #[test]
    fn malformed_topics_are_rejected_before_io() {
        assert!(matches!(
            PulsarSubscription::new("a/b", "workers").validate(),
            Err(PulsarError::Invalid(_))
        ));
    }

    /// Half a declaration is not a policy the client can apply, so it never reaches the broker.
    #[test]
    fn half_a_retry_declaration_is_rejected_before_io() {
        let capped = declared(
            PulsarSubscription::new("orders", "workers"),
            &RetryDeclaration::new().with_max_attempts(nonzero!(3u32)),
        );
        let message = match capped.validate() {
            Err(PulsarError::Invalid(message)) => message,
            other => panic!("a cap with no destination must be refused, got {other:?}"),
        };
        assert!(message.contains("dead_letter"), "{message}");

        let addressed = declared(
            PulsarSubscription::new("orders", "workers"),
            &RetryDeclaration::new().with_dead_letter("orders-dlq"),
        );
        let message = match addressed.validate() {
            Err(PulsarError::Invalid(message)) => message,
            other => panic!("a destination with no cap must be refused, got {other:?}"),
        };
        assert!(message.contains("max_attempts"), "{message}");
    }

    /// Both halves become the consumer's own policy, with the limit counted in deliveries.
    #[test]
    fn a_whole_declaration_becomes_the_consumers_policy() {
        let policy = declared(
            PulsarSubscription::new("orders", "workers"),
            &RetryDeclaration::new()
                .with_max_attempts(nonzero!(3u32))
                .with_dead_letter("orders-dlq"),
        );
        assert_eq!(
            policy.dead_letter_policy().expect("both halves declared"),
            Some(DeadLetterRoute {
                topic: "orders-dlq".to_owned(),
                max_deliveries: 3,
            })
        );
    }

    /// A cap the broker would never reach is refused at startup rather than left to circle the
    /// subscription: Pulsar keeps a redelivery count only where a subscription dispatches to
    /// competing consumers, which the live suite drives against a server.
    #[test]
    fn a_cap_on_a_single_active_subscription_is_rejected_before_io() {
        let whole = RetryDeclaration::new()
            .with_max_attempts(nonzero!(3u32))
            .with_dead_letter("orders-dlq");
        for sharing in [SubscriptionType::Exclusive, SubscriptionType::Failover] {
            let capped = declared(
                PulsarSubscription::new("orders", "workers").subscription_type(sharing),
                &whole,
            );
            let message = match capped.validate() {
                Err(PulsarError::Invalid(message)) => message,
                other => {
                    panic!("a cap on a {sharing:?} subscription must be refused, got {other:?}")
                }
            };
            assert!(message.contains(&format!("{sharing:?}")), "{message}");
            assert!(message.contains("KeyShared"), "{message}");
        }
    }

    /// The same subscription without a declaration is not the crate's business to refuse.
    #[test]
    fn a_single_active_subscription_without_a_cap_is_accepted() {
        for sharing in [SubscriptionType::Exclusive, SubscriptionType::Failover] {
            PulsarSubscription::new("orders", "workers")
                .subscription_type(sharing)
                .validate()
                .expect("a subscription that declares no cap opens on any type");
        }
    }

    #[test]
    fn malformed_patterns_are_rejected_before_io() {
        assert!(matches!(
            PulsarSubscription::pattern("orders-(", "workers").validate(),
            Err(PulsarError::Invalid(_))
        ));
    }
}