Dead letters

max_attempts(n) and dead_letter(topic) at the mount site become the subscription's own dead-letter policy, written at startup.

A Pub/Sub subscription moves a spent message itself. After n deliveries the service stops redelivering the message and publishes it to the dead-letter topic, with its payload and its attributes. The crate publishes nothing: it writes the declaration onto the subscription as its dead-letter policy. A subscription that create_with_topic creates carries the policy from the start, and the dead-letter topic is created with it. A subscription that already exists gets the policy written onto it at startup, whether the handler names it by string or by descriptor.

The policy is one field of the subscription with both halves, so the mount site declares both. A cap without a topic, a topic without a cap, or a cap outside the 5 to 100 Pub/Sub accepts fails the startup. Two registrations on one subscription that declare different policies fail the startup too. .out_retry(..) does not compile on this broker, because no retry copy is ever published.

The dead-letter topic is an ordinary topic, read through a subscription of its own. A message published to a topic with no subscription is lost, so the reader's subscription exists before the first message is spent. The Pub/Sub service account of the project needs permission to publish to the dead-letter topic and to subscribe to the source subscription. Pub/Sub counts deliveries on a best-effort basis, so a message may get a few more or fewer than n.

Example

Five deliveries of a declined payment; after the fifth, Pub/Sub publishes it to payments-dead.

Macros
Manual

examples/pubsub_settlement_dead_letters.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // Five deliveries of one payment; after the fifth Pub/Sub publishes it to
            // `payments-dead`, payload and attributes as they arrived.
            b.include(charge)
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments-dead");
            b.include(on_dead_letter);
        },
    )
}

examples/manual/pubsub_settlement_dead_letters.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        PubSubBroker::new("my-project"),
        |b| {
            // Five deliveries of one payment; after the fifth Pub/Sub publishes it to
            // `payments-dead`, payload and attributes as they arrived.
            b.include(subscriber(payments(), Charge).build())
                .max_attempts(nonzero!(5u32))
                .dead_letter("payments-dead");
            b.include(subscriber(dead_letters(), OnDeadLetter).build());
        },
    )
}

The dead-letter topic is read by an ordinary subscription:

Macros
Manual

examples/pubsub_settlement_dead_letters.rs
/// The dead-letter topic is an ordinary topic, read through a subscription of its own.
#[subscriber(GooglePubSub::new("payments-dead-audit").create_with_topic("payments-dead"))]
async fn on_dead_letter(payment: &Payment) -> HandlerOutcome {
    tracing::error!(payment.id, "payment dead-lettered");
    HandlerOutcome::ack()
}

examples/manual/pubsub_settlement_dead_letters.rs
/// The dead-letter topic is an ordinary topic, read through a subscription of its own.
struct OnDeadLetter;

impl Handle<Payment> for OnDeadLetter {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::error!(payment.id, "payment dead-lettered");
        ready(Ok(()))
    }
}

fn dead_letters() -> GooglePubSub {
    GooglePubSub::new("payments-dead-audit").create_with_topic("payments-dead")
}

Example​

Example