ruststream-gcp-pubsub 0.7.0

Google Cloud Pub/Sub broker implementation for the RustStream messaging framework.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
//! [`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::{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 registration declared 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 {
///     println!("got order {}", order.id);
///     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 changes no resource 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. The local
    /// emulator does not, so a run against it sees no back-pressure.
    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));
    ///     }
    ///     println!("sent invoice {}", invoice.id);
    ///     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();
    ///     println!("indexed {ids:?}");
    ///     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 {
///     println!("got order {}", order.id);
///     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())
    }
}

/// 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.
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)),
        );
    }
}