Failures

What a subscriber does when its handler panics or a payload cannot decode, set per subscriber with on_failure.

A subscriber carries two policies: one for a handler panic, one for a delivery that cannot decode. A panic fails fast by default, and a test reads it with panicked: a loud error names the subscription, the service shuts down gracefully, and run returns an error, so an orchestrator restarts it. Fail fast takes effect under an unwinding panic profile; with panic = "abort" the process is already gone. A decode failure drops the delivery by default: one malformed payload does not stop the consumer, and assert_last_failed_to_decode checks it in a test. A decode failure is a payload the codec rejects, a header contract that does not parse, a missing carried value, or a delivery the broker reports unreadable. Each policy is a FailurePolicy: fail_fast, drop, retry, retry_after(delay) or skip. drop, retry and retry_after(delay) settle the way the handler outcomes of the same name do. skip acks the delivery to move past it; the message is gone and was never handled. retry and retry_after bring the delivery back, so a payload that never decodes keeps coming back. The attribute takes on_failure(panic = .., decode = ..), with either key alone; the manual form takes a FailurePolicies value.

Example

Macros

examples/failure_policy.rs

/// No clause: a payload that cannot decode is dropped, and a panic stops the service.
#[subscriber("orders")]
async fn fulfil(order: &Order) -> HandlerOutcome {
    assert_ne!(order.id, 0, "an order without an id reached fulfilment");
    HandlerOutcome::ack()
}

Manual

examples/manual/failure_policy.rs

/// No settings step: a payload that cannot decode is dropped, and a panic stops the service.
struct Fulfil;

impl Handle<Order> for Fulfil {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        assert_ne!(order.id, 0, "an order without an id reached fulfilment");
        ready(Ok(()))
    }
}

Move past every failure

Macros

examples/failure_policy.rs

/// An audit trail moves past anything it cannot process: both failures ack the delivery.
#[subscriber("orders", on_failure(panic = skip, decode = skip))]
async fn audit(order: &Order) -> HandlerOutcome {
    assert_ne!(order.id, 0, "an order without an id reached the audit");
    HandlerOutcome::ack()
}

Manual

examples/manual/failure_policy.rs

/// An audit trail moves past anything it cannot process: both failures ack the delivery.
struct Audit;

impl Handle<Order> for Audit {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        assert_ne!(order.id, 0, "an order without an id reached the audit");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            subscriber("orders", Audit)
                .on_failure(
                    FailurePolicies::default()
                        .with_panic(FailurePolicy::Skip)
                        .with_decode(FailurePolicy::Skip),
                )
                .build(),
        );
    })
}

A payload to read later

Macros

examples/failure_policy.rs

const SCHEMA_ROLLOUT: Duration = Duration::from_secs(60);

/// Producers roll a schema change out first: a payload this version cannot read yet comes
/// back in a minute instead of being dropped. A panic still stops the service.
#[subscriber("orders", on_failure(decode = retry_after(SCHEMA_ROLLOUT)))]
async fn accept(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "accepted");
    HandlerOutcome::ack()
}

Manual

examples/manual/failure_policy.rs

const SCHEMA_ROLLOUT: Duration = Duration::from_secs(60);

/// Producers roll a schema change out first: a payload this version cannot read yet comes
/// back in a minute instead of being dropped. A panic still stops the service.
struct Accept;

impl Handle<Order> for Accept {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "accepted");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(MemoryBroker::new(), |b| {
        b.include(
            subscriber("orders", Accept)
                .on_failure(
                    FailurePolicies::default()
                        .with_decode(FailurePolicy::RetryAfter(SCHEMA_ROLLOUT)),
                )
                .build(),
        );
    })
}