ruststream_gcp_pubsub/

subscription.rs

//! [`GooglePubSub`]: the subscription descriptor.
//!
//! Pub/Sub separates the topic from the subscription, and the descriptor keeps both explicit:
//! by default it names an existing subscription; `create_with_topic` opts into creating the
//! subscription (and its topic) on subscribe, which is what local development against the
//! emulator wants.

use std::borrow::Cow;
use std::collections::HashMap;
use std::num::NonZeroU32;
use std::ops::RangeInclusive;
use std::sync::{Mutex, PoisonError};
use std::time::Duration;

use ruststream::runtime::IntoSource;
use ruststream::{BrokerMoves, DeclareRetryError, FromName, RetryDeclaration, SubscriptionSource};

use crate::broker::ConnectedPubSubBroker;
use crate::error::PubSubError;
use crate::subscriber::PubSubSubscriber;

/// How long a partial batch waits for more deliveries before it goes out. One streaming-pull
/// burst crosses the network in tens of milliseconds, so a deadline much shorter than this
/// would cut most batches down to the first delivery that arrives.
const DEFAULT_BATCH_WAIT: Duration = Duration::from_millis(50);

/// The range Pub/Sub accepts for a dead-letter policy's `maxDeliveryAttempts`. A cap outside it
/// is refused before the subscription opens, rather than at the API call that would reject it.
const DELIVERY_ATTEMPTS: RangeInclusive<i32> = 5..=100;

/// How long the client keeps extending the ack deadline of a delivery nothing has settled, which
/// is the client's own default and the budget a delayed retry spends. Past it the client stops
/// extending, the lease expires, and the subscription redelivers on its own.
const DEFAULT_MAX_LEASE: Duration = Duration::from_hours(1);

/// What one delivery may do before it has to be settled, read off the subscription that opened
/// it: how many deliveries the dead-letter policy allows, and how long the client will keep a
/// delivery leased.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct DeliveryLimits {
    /// `None` where the subscription has no dead-letter policy, and Pub/Sub then counts nothing.
    pub(crate) max_delivery_attempts: Option<i32>,
    pub(crate) max_lease: Duration,
}

/// A subscription descriptor for one Pub/Sub subscription.
///
/// Implements [`SubscriptionSource`] for the connected broker, so it sits inline in the
/// `#[subscriber(..)]` decorator, and the declaration a service ships is the one its tests run:
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_gcp_pubsub::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Debug, Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(
///     GooglePubSub::new("orders-workers")
///         .max_outstanding(1_000)
///         .ack_extension(Duration::from_secs(60))
/// )]
/// 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(PubSubBroker::new("my-project"), |b| {
///             b.include(handle);
///         })
/// }
/// # }
/// # fn main() {}
/// ```
#[derive(Debug, Clone, PartialEq, Eq)]
#[must_use]
pub struct GooglePubSub {
    name: String,
    create_with_topic: Option<String>,
    max_outstanding: Option<i64>,
    ack_extension: Option<Duration>,
    batch_wait: Duration,
    /// What the registration declared with `max_attempts(..)`, taken in by `declare_retry`.
    max_attempts: Option<NonZeroU32>,
    /// What it declared with `dead_letter(..)`: the topic spent deliveries are published to.
    dead_letter: Option<String>,
    max_lease: Duration,
}

impl GooglePubSub {
    /// Names an existing subscription (short name or full
    /// `projects/{p}/subscriptions/{s}` resource name).
    ///
    /// The name is taken on trust: nothing is created, and the subscription is infrastructure
    /// this crate does not own. Where it answers to nothing, the streaming pull is what says so,
    /// and the stream yields [`PubSubError::Receive`] naming the subscription instead of
    /// delivering.
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            create_with_topic: None,
            max_outstanding: None,
            ack_extension: None,
            batch_wait: DEFAULT_BATCH_WAIT,
            max_attempts: None,
            dead_letter: None,
            max_lease: DEFAULT_MAX_LEASE,
        }
    }

    /// Creates the subscription bound to `topic` on subscribe when it does not exist yet (the
    /// topic is created too). Meant for local development and tests against the emulator;
    /// production subscriptions are usually managed as infrastructure.
    ///
    /// The subscription it creates enables message ordering, so a run of messages sharing an
    /// ordering key reaches the handler in publish order. A subscription that already exists
    /// keeps whatever its own configuration says: ordered delivery is a field of the
    /// subscription resource, and the crate does not change it on a subscription it did not
    /// create.
    pub fn create_with_topic(mut self, topic: impl Into<String>) -> Self {
        self.create_with_topic = Some(topic.into());
        self
    }

    /// Flow control: how many received messages may be outstanding (unacked) at once. Defaults
    /// to the client's 1000.
    ///
    /// The limit travels with the streaming pull and the service is what holds to it, pausing
    /// the stream while this many deliveries are unsettled and resuming below it.
    pub fn max_outstanding(mut self, messages: i64) -> Self {
        self.max_outstanding = Some(messages);
        self
    }

    /// How far each background ack-deadline extension reaches while a handler runs. The client
    /// clamps it to the protocol's 10s..=600s range; defaults to 60s.
    pub fn ack_extension(mut self, extension: Duration) -> Self {
        self.ack_extension = Some(extension);
        self
    }

    /// How long one delivery may stay unsettled before the client stops extending its ack
    /// deadline. Defaults to the client's own hour.
    ///
    /// This is the budget a delayed retry spends: `HandlerOutcome::retry_after(delay)` holds the
    /// delivery in the process for `delay`, and a delay longer than this is refused at the call,
    /// because the subscription would redeliver before it elapsed.
    ///
    /// ```
    /// # mod demo {
    /// use std::time::Duration;
    ///
    /// use ruststream_gcp_pubsub::prelude::*;
    /// use serde::Deserialize;
    ///
    /// #[derive(Debug, Deserialize)]
    /// struct Invoice {
    ///     id: u64,
    ///     ready: bool,
    /// }
    ///
    /// /// An invoice that is not ready yet comes back in ninety minutes, inside the lease.
    /// #[subscriber(
    ///     GooglePubSub::new("invoices-workers").max_lease(Duration::from_hours(2))
    /// )]
    /// async fn send(invoice: &Invoice) -> HandlerOutcome {
    ///     if !invoice.ready {
    ///         return HandlerOutcome::retry_after(Duration::from_mins(90));
    ///     }
    ///     tracing::info!(invoice.id, "sent");
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("invoices", "0.1.0"))
    ///         .with_broker(PubSubBroker::new("my-project"), |b| {
    ///             b.include(send);
    ///         })
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    pub fn max_lease(mut self, lease: Duration) -> Self {
        self.max_lease = lease;
        self
    }

    /// How long a partial batch waits for more deliveries after its first one, for a handler
    /// that takes a slice. Defaults to 50ms.
    ///
    /// How *large* a batch may be is not named here: that is the registration's
    /// `batch(n)`, which reaches the subscription on its own. This is the other half - how
    /// long the subscription is willing to wait for a batch that size, before handing over
    /// what it has.
    ///
    /// ```
    /// # mod demo {
    /// use std::time::Duration;
    ///
    /// use ruststream_gcp_pubsub::prelude::*;
    /// use serde::Deserialize;
    ///
    /// #[derive(Debug, Deserialize)]
    /// struct Order {
    ///     id: u64,
    /// }
    ///
    /// #[subscriber(GooglePubSub::new("orders-workers").batch_wait(Duration::from_millis(200)))]
    /// async fn index(orders: &[Order]) -> HandlerOutcome {
    ///     let ids: Vec<u64> = orders.iter().map(|order| order.id).collect();
    ///     tracing::info!(?ids, "indexed");
    ///     HandlerOutcome::ack()
    /// }
    ///
    /// #[ruststream::app]
    /// fn app() -> impl App {
    ///     RustStream::new(AppInfo::new("orders", "0.1.0"))
    ///         .with_broker(PubSubBroker::new("my-project"), |b| {
    ///             // Up to 100 orders per call, or whatever arrived within 200ms of the first.
    ///             b.include(index.batch(nonzero!(100)));
    ///         })
    /// }
    /// # }
    /// # fn main() {}
    /// ```
    pub fn batch_wait(mut self, wait: Duration) -> Self {
        self.batch_wait = wait;
        self
    }

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

    pub(crate) fn create_topic_ref(&self) -> Option<&str> {
        self.create_with_topic.as_deref()
    }

    pub(crate) fn max_outstanding_value(&self) -> Option<i64> {
        self.max_outstanding
    }

    pub(crate) fn ack_extension_value(&self) -> Option<Duration> {
        self.ack_extension
    }

    pub(crate) fn batch_wait_value(&self) -> Duration {
        self.batch_wait
    }

    /// What one delivery of this subscription may do before it has to be settled.
    pub(crate) fn limits(&self) -> DeliveryLimits {
        DeliveryLimits {
            max_delivery_attempts: self.dead_letter_policy().map(|(_, attempts)| attempts),
            max_lease: self.max_lease,
        }
    }

    /// The declared cap in the API's own type. A count past `i32::MAX` saturates onto a value
    /// [`validate`](Self::validate) refuses, so a cap too large to express is rejected rather
    /// than quietly reduced.
    fn declared_attempts(&self) -> Option<i32> {
        self.max_attempts
            .map(|attempts| i32::try_from(attempts.get()).unwrap_or(i32::MAX))
    }

    /// The dead-letter policy this subscription opens with: the topic spent deliveries go to and
    /// how many deliveries one message gets. `None` where the registration declared neither.
    pub(crate) fn dead_letter_policy(&self) -> Option<(&str, i32)> {
        Some((self.dead_letter.as_deref()?, self.declared_attempts()?))
    }

    /// Takes in a registration's cap and dead-letter destination, which become the
    /// subscription's dead-letter policy when it opens. The two `SubscriptionSource` impls and
    /// the by-name ledger below all go through here, so a declaration means the same thing
    /// whichever way the registration named its subscription.
    fn take_declaration(mut self, declaration: &RetryDeclaration) -> Self {
        self.max_attempts = declaration.max_attempts();
        self.dead_letter = declaration.dead_letter().map(ToOwned::to_owned);
        self
    }

    /// Rejects descriptors that cannot form a subscription, before any I/O.
    pub(crate) fn validate(&self) -> Result<(), PubSubError> {
        if self.name.is_empty() {
            return Err(PubSubError::InvalidDescriptor(
                "subscription name must be non-empty".into(),
            ));
        }
        if self.create_with_topic.as_deref() == Some("") {
            return Err(PubSubError::InvalidDescriptor(
                "topic name must be non-empty".into(),
            ));
        }
        self.validate_declaration()
    }

    /// Holds the registration's declaration to what a Pub/Sub dead-letter policy can express.
    ///
    /// A dead-letter policy is one resource field carrying both halves, so a subscription cannot
    /// honour half a declaration: a cap alone would leave a spent delivery circulating, and a
    /// destination alone has no attempt count to fire on. Saying so at startup is the earliest
    /// the crate can - the mount site's declaration reaches the descriptor as data, so nothing
    /// here is a type the compiler could reject.
    fn validate_declaration(&self) -> Result<(), PubSubError> {
        match (self.declared_attempts(), self.dead_letter.as_deref()) {
            (Some(attempts), Some(topic)) => {
                if topic.is_empty() {
                    return Err(PubSubError::InvalidDescriptor(
                        "dead-letter topic must be non-empty".into(),
                    ));
                }
                if !DELIVERY_ATTEMPTS.contains(&attempts) {
                    return Err(PubSubError::InvalidDescriptor(format!(
                        "max_attempts({attempts}) is outside the {}..={} a Pub/Sub dead-letter \
                         policy accepts",
                        DELIVERY_ATTEMPTS.start(),
                        DELIVERY_ATTEMPTS.end(),
                    )));
                }
                Ok(())
            }
            (Some(_), None) => Err(PubSubError::InvalidDescriptor(format!(
                "subscription '{}' declares max_attempts without dead_letter; a Pub/Sub \
                 dead-letter policy needs the topic too",
                self.name,
            ))),
            (None, Some(_)) => Err(PubSubError::InvalidDescriptor(format!(
                "subscription '{}' declares dead_letter without max_attempts; a Pub/Sub \
                 dead-letter policy needs the attempt count too",
                self.name,
            ))),
            (None, None) => Ok(()),
        }
    }
}

/// A subscription is identified by its name and nothing else, every other setting having a
/// default, so the mount site may supply the name instead of the declaration:
/// `#[subscriber(GooglePubSub)]` on the handler and `.name("orders-workers")` where it is mounted.
///
/// # Examples
///
/// ```
/// # mod demo {
/// use ruststream_gcp_pubsub::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Debug, Deserialize)]
/// struct Order {
///     id: u64,
/// }
///
/// #[subscriber(GooglePubSub)]
/// 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(PubSubBroker::new("my-project"), |b| {
///             // The deployment names the subscription; every other setting is the default.
///             b.include(handle.name("orders-workers"));
///         })
/// }
/// # }
/// # fn main() {}
/// ```
impl FromName for GooglePubSub {
    fn from_name(name: impl Into<Cow<'static, str>>) -> Self {
        Self::new(name.into().into_owned())
    }
}

// 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 GooglePubSub {
    type Source = Self;

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

/// A Pub/Sub subscription moves a spent delivery itself, so the copy path is
/// [`BrokerMoves`]: the subscription's dead-letter policy is the mechanism, and nothing is
/// published from the service to retry a message. `.out_retry(..)` over this descriptor is
/// therefore a compile error, and the error names it.
///
/// # Examples
///
/// `.max_attempts(n).dead_letter(topic)` at the mount site becomes the subscription's own
/// dead-letter policy, written when the service starts:
///
/// ```
/// # mod demo {
/// use std::time::Duration;
///
/// use ruststream_gcp_pubsub::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Debug, Deserialize)]
/// struct Payment {
///     id: u64,
///     settled: bool,
/// }
///
/// #[subscriber(GooglePubSub::new("payments-workers").batch_wait(Duration::from_millis(200)))]
/// async fn reconcile(payments: &[Payment]) -> HandlerOutcome {
///     if payments.iter().any(|payment| !payment.settled) {
///         return HandlerOutcome::retry_after(Duration::from_secs(5));
///     }
///     tracing::info!(count = payments.len(), "settled");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
///         PubSubBroker::new("my-project"),
///         |b| {
///             // Pub/Sub gives one payment five deliveries, then publishes it to `payments-dead`.
///             b.include(reconcile.batch(nonzero!(50)))
///                 .max_attempts(nonzero!(5u32))
///                 .dead_letter("payments-dead");
///         },
///     )
/// }
/// # }
/// # fn main() {}
/// ```
impl SubscriptionSource<ConnectedPubSubBroker> for GooglePubSub {
    type Subscriber = PubSubSubscriber;
    type Copies = BrokerMoves;

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

    async fn subscribe(
        self,
        connected: &ConnectedPubSubBroker,
    ) -> Result<PubSubSubscriber, PubSubError> {
        connected.subscribe_descriptor(self).await
    }

    /// Takes in the registration's cap and dead-letter destination, which become the
    /// subscription's dead-letter policy when [`subscribe`](Self::subscribe) opens it: the topic
    /// is the policy's `deadLetterTopic` and the cap its `maxDeliveryAttempts`.
    ///
    /// Only recorded here. The declaration turns into topology where the connection exists, and
    /// a half declaration is refused there rather than opening a subscription that honours
    /// neither half.
    fn declare_retry(self, declaration: &RetryDeclaration) -> Self {
        self.take_declaration(declaration)
    }
}

/// What the bare-name registrations of one connection declared about their retries.
///
/// A bare name carries no descriptor to declare on, so the broker takes the declaration through
/// [`Subscribe::declare_retry`](ruststream::Subscribe::declare_retry) and keeps it here until
/// the subscription of that name opens. The connection holds one over either transport, so
/// `#[subscriber("orders-workers")]` gets the dead-letter policy the mount site declared in a test
/// as in production.
#[derive(Debug, Default)]
pub(crate) struct DeclaredRetries(Mutex<HashMap<String, RetryDeclaration>>);

impl DeclaredRetries {
    /// Takes what a registration declared for the subscription `name` opens.
    ///
    /// Refuses what a Pub/Sub dead-letter policy cannot express - half a declaration, a cap
    /// outside the range the service accepts - on the same check the descriptor runs, and
    /// refuses a second registration on the same subscription that declares something else: one
    /// subscription carries one policy, and the last writer winning would silently drop what the
    /// other registration asked for.
    pub(crate) fn take(
        &self,
        name: &str,
        declaration: &RetryDeclaration,
    ) -> Result<(), DeclareRetryError> {
        GooglePubSub::new(name)
            .take_declaration(declaration)
            .validate()
            .map_err(refusal)?;
        let conflicts = {
            let mut taken = self.0.lock().unwrap_or_else(PoisonError::into_inner);
            match taken.get(name) {
                Some(already) if already != declaration => true,
                _ => {
                    taken.insert(name.to_owned(), declaration.clone());
                    false
                }
            }
        };
        if conflicts {
            return Err(refusal(PubSubError::InvalidDescriptor(format!(
                "subscription '{name}' is mounted twice with different retry \
                 declarations, and one subscription carries one dead-letter policy",
            ))));
        }
        Ok(())
    }

    /// The descriptor the bare name `name` opens: the subscription it names, carrying what the
    /// registration declared for it.
    pub(crate) fn source(&self, name: &str) -> GooglePubSub {
        let source = GooglePubSub::new(name);
        match self
            .0
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .get(name)
        {
            Some(declaration) => source.take_declaration(declaration),
            None => source,
        }
    }
}

/// A startup refusal, with the crate's own error as its cause: the runtime prints the reason on
/// one line and fails the registration that declared it.
fn refusal(err: PubSubError) -> DeclareRetryError {
    DeclareRetryError::Broker(Box::new(err))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn empty_subscription_name_is_rejected_before_io() {
        assert!(matches!(
            GooglePubSub::new("").validate(),
            Err(PubSubError::InvalidDescriptor(_))
        ));
    }

    #[test]
    fn empty_topic_name_is_rejected_before_io() {
        assert!(matches!(
            GooglePubSub::new("s").create_with_topic("").validate(),
            Err(PubSubError::InvalidDescriptor(_))
        ));
    }

    /// What a registration declaring `attempts` and `destination` states at its mount site.
    fn declaration(attempts: Option<u32>, destination: Option<&str>) -> RetryDeclaration {
        let mut declaration = RetryDeclaration::new();
        if let Some(attempts) = attempts {
            declaration =
                declaration.with_max_attempts(NonZeroU32::new(attempts).expect("non-zero"));
        }
        if let Some(destination) = destination {
            declaration = declaration.with_dead_letter(destination.to_owned());
        }
        declaration
    }

    /// Builds the descriptor a registration declaring `attempts` and `destination` produces.
    fn declared(attempts: Option<u32>, destination: Option<&str>) -> GooglePubSub {
        SubscriptionSource::<ConnectedPubSubBroker>::declare_retry(
            GooglePubSub::new("orders-workers"),
            &declaration(attempts, destination),
        )
    }

    /// Both halves are one resource field, so the descriptor carries them as one policy.
    #[test]
    fn a_full_declaration_becomes_the_subscriptions_dead_letter_policy() {
        let source = declared(Some(5), Some("orders-dead"));
        assert_eq!(source.dead_letter_policy(), Some(("orders-dead", 5)));
    }

    /// A registration that declared nothing leaves the subscription's own topology alone.
    #[test]
    fn an_empty_declaration_leaves_the_subscription_untouched() {
        let source = declared(None, None);
        assert_eq!(source.dead_letter_policy(), None);
        assert!(source.validate().is_ok());
    }

    /// Half a declaration cannot be honoured, and the subscription says so before it opens
    /// rather than running without the cap the registration asked for.
    #[test]
    fn half_a_declaration_is_refused_before_io() {
        for (attempts, destination) in [(Some(5), None), (None, Some("orders-dead"))] {
            let source = declared(attempts, destination);
            assert!(
                matches!(source.validate(), Err(PubSubError::InvalidDescriptor(_))),
                "declaring {attempts:?} / {destination:?} must be refused",
            );
        }
    }

    /// Pub/Sub bounds `maxDeliveryAttempts`, so a cap outside it is named here and not by the
    /// admin call that would reject it.
    #[test]
    fn a_cap_outside_the_services_range_is_refused_before_io() {
        for attempts in [1, 4, 101, 1_000] {
            let source = declared(Some(attempts), Some("orders-dead"));
            assert!(
                matches!(source.validate(), Err(PubSubError::InvalidDescriptor(_))),
                "max_attempts({attempts}) must be refused",
            );
        }
        for attempts in [5, 100] {
            let source = declared(Some(attempts), Some("orders-dead"));
            assert!(
                source.validate().is_ok(),
                "max_attempts({attempts}) must be accepted",
            );
        }
    }

    /// A cap too large for the wire type saturates onto a value the range check refuses, so it
    /// is rejected rather than quietly reduced to something the service accepts.
    #[test]
    fn a_cap_past_the_wire_type_is_refused_rather_than_reduced() {
        let source = declared(Some(u32::MAX), Some("orders-dead"));
        assert_eq!(source.declared_attempts(), Some(i32::MAX));
        assert!(matches!(
            source.validate(),
            Err(PubSubError::InvalidDescriptor(_))
        ));
    }

    /// A dead-letter topic is a resource name, so an empty one is refused with the rest.
    #[test]
    fn an_empty_dead_letter_topic_is_refused_before_io() {
        let source = declared(Some(5), Some(""));
        assert!(matches!(
            source.validate(),
            Err(PubSubError::InvalidDescriptor(_))
        ));
    }

    /// A bare name has no descriptor to hold the declaration, so the broker's ledger holds it
    /// and the subscription that name opens carries the policy.
    #[test]
    fn a_declaration_for_a_bare_name_reaches_the_subscription_it_opens() {
        let taken = DeclaredRetries::default();

        taken
            .take("orders-workers", &declaration(Some(5), Some("orders-dead")))
            .expect("a full declaration is one a dead-letter policy carries");

        assert_eq!(
            taken.source("orders-workers").dead_letter_policy(),
            Some(("orders-dead", 5)),
        );
    }

    /// A name nothing declared for opens the subscription as it stands.
    #[test]
    fn a_bare_name_with_no_declaration_opens_the_subscription_untouched() {
        let taken = DeclaredRetries::default();

        taken
            .take("orders-workers", &RetryDeclaration::new())
            .expect("a registration may declare nothing");

        assert_eq!(
            taken.source("orders-workers"),
            GooglePubSub::new("orders-workers")
        );
    }

    /// The ledger holds a bare name to the same policy the descriptor is held to, and the
    /// refusal carries the crate's own reason so the startup line says what to fix.
    #[test]
    fn a_bare_name_is_refused_what_a_dead_letter_policy_cannot_express() {
        let taken = DeclaredRetries::default();

        for (attempts, destination) in [
            (Some(5), None),
            (None, Some("orders-dead")),
            (Some(3), Some("orders-dead")),
        ] {
            let err = taken
                .take("orders-workers", &declaration(attempts, destination))
                .expect_err("the declaration must be refused");
            assert!(
                matches!(err, DeclareRetryError::Broker(_)),
                "declaring {attempts:?} / {destination:?} must refuse as a broker rejection",
            );
        }
    }

    /// One subscription carries one dead-letter policy, so a second registration declaring
    /// something else for the same name is refused rather than silently overwriting the first.
    #[test]
    fn one_subscription_declared_twice_differently_is_refused() {
        let taken = DeclaredRetries::default();
        taken
            .take("orders-workers", &declaration(Some(5), Some("orders-dead")))
            .expect("the first declaration is taken");

        let err = taken
            .take(
                "orders-workers",
                &declaration(Some(10), Some("orders-dead")),
            )
            .expect_err("a contradicting declaration must be refused");

        assert!(matches!(err, DeclareRetryError::Broker(_)));
        assert_eq!(
            taken.source("orders-workers").dead_letter_policy(),
            Some(("orders-dead", 5)),
            "the refusal leaves the declaration already taken alone",
        );
    }

    /// Two registrations that declare the same thing are no contradiction: the subscription
    /// opens with the policy both asked for.
    #[test]
    fn one_subscription_declared_twice_alike_is_taken_once() {
        let taken = DeclaredRetries::default();
        let declared = declaration(Some(5), Some("orders-dead"));

        taken.take("orders-workers", &declared).expect("the first");
        taken.take("orders-workers", &declared).expect("the second");

        assert_eq!(
            taken.source("orders-workers").dead_letter_policy(),
            Some(("orders-dead", 5)),
        );
    }
}