Skip to main content

Acknowledgement

ZeroMQ settles nothing: ack and nack report Unsupported, a plain retry delivers nothing again, and every retry is a copy this service publishes.

ZMTP has no frame that settles a delivery. A message reaches the subscription once, and the socket keeps no copy of it. So ack and nack on a delivery report AckError::Unsupported, and the runtime logs a warning naming the subscription. An ack and a drop both end the delivery, and nothing comes back after either.

A retry asks the transport to deliver the message again, and the transport has nothing to deliver: the handler is not called again. A retry_after(delay) is served by the runtime instead. It publishes a copy of the delivery once the delay is over, and the copy arrives as a new message, as Retries describes. Under a cap a plain retry travels as such a copy too. Where the copies go follows the side of the endpoint, as out_retry describes.

A delivery reports no redelivery_count: the framework counts its copies in a header of its own. Delivery is at most once, as Settlement draws. A process that stops loses what its subscriptions had read and not yet handled, and every copy still waiting out its delay.

Retries and dead letters retries jobs and parks failed invoices on a transport that delivers at most once.

Example​

A payment of nothing is dropped. A payment the gateway fails comes back as a copy 10 seconds later, through the queue the service binds.

examples/zmq_settlement_at_most_once.rs
#[subscriber("payments")]
async fn charge(payment: &Payment) -> HandlerOutcome {
    if payment.amount == 0 {
        // Nothing to charge. The delivery is gone either way, so it is dropped on purpose.
        tracing::warn!(payment.id, "a payment of nothing, dropped");
        return HandlerOutcome::drop();
    }
    match gateway(payment) {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            // The socket keeps no copy, so the runtime publishes one after the delay.
            tracing::warn!(payment.id, %err, "the gateway failed, trying again later");
            HandlerOutcome::retry_after(Duration::from_secs(10))
        }
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        ZmqQueue::new(ZmqEndpoint::bind("tcp://0.0.0.0:5555")),
        |b| {
            b.include(charge);
        },
    )
}