Skip to main content

Publishing all or nothing

This guide publishes several messages from one handler together or not at all.

AMQP 1.0 has transactions in the protocol: a sender posts messages inside a transaction, and the broker makes them visible on the commit. The transaction feature turns this on:

Cargo.toml
[dependencies]
ruststream-amqp = { version = "0.7", features = ["transaction"] }

The example is a billing service. It turns each order into ledger entries and an invoice, and either all of them reach the broker or none does.

The handler​

src/main.rs
#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "billing.orders")]
struct Order {
    id: u64,
    lines: Vec<Line>,
}

#[derive(Debug, PartialEq, Serialize, Deserialize)]
struct Line {
    sku: String,
    cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "ledger")]
struct LedgerEntry {
    order: u64,
    sku: String,
    cents: u64,
}

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "invoices")]
struct Invoice {
    order: u64,
    total_cents: u64,
}

#[derive(OutSlot)]
#[publishes(LedgerEntry, Invoice)]
struct Billing;

#[subscriber(AmqpAddress::queue("billing.orders"))]
async fn bill(
    order: &Order,
    Out(billing): Out<impl TransactionalPublisher, Billing>,
) -> HandlerOutcome {
    if billing.begin_transaction().await.is_err() {
        return HandlerOutcome::retry();
    }
    for line in &order.lines {
        if line.cents == 0 {
            tracing::warn!(
                order.id,
                line.sku,
                "a line without a price, nothing is billed"
            );
            // A failed abort has still ended the transaction, so there is nothing left to undo.
            let _ = billing.abort().await;
            return HandlerOutcome::drop();
        }
        let entry = LedgerEntry {
            order: order.id,
            sku: line.sku.clone(),
            cents: line.cents,
        };
        if billing.message(&entry).publish().await.is_err() {
            let _ = billing.abort().await;
            return HandlerOutcome::retry();
        }
    }
    let invoice = Invoice {
        order: order.id,
        total_cents: order.lines.iter().map(|line| line.cents).sum(),
    };
    if billing.message(&invoice).publish().await.is_err() {
        let _ = billing.abort().await;
        return HandlerOutcome::retry();
    }
    if billing.commit().await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(order.id, invoice.total_cents, "order billed");
    HandlerOutcome::ack()
}

The Billing slot is bounded Out<impl TransactionalPublisher, _>: the body says it needs transactions and names no broker. begin_transaction opens a transaction on the slot's publisher. Every publish after it is posted inside the transaction, and the broker holds it back. commit makes the ledger entries and the invoice visible together. A line without a price makes the order unbillable, and abort discards what was posted, so nothing reaches ledger or invoices.

A commit or an abort that fails has still ended the transaction, so the next begin_transaction starts a fresh one. bill retries the order when the commit fails.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("billing", "0.1.0")).with_broker(
        AmqpBroker::new("amqp://artemis:artemis@localhost:5672"),
        |b| {
            b.include(bill).out(Billing, TransactionalPublish).build();
        },
    )
}

TransactionalPublish is the policy that carries transactions, AmqpTransactionalPublish under its prelude name. Its publisher holds one transaction at a time. A second begin_transaction while one is open returns an error and leaves the open one untouched. Each registration gets a publisher of its own, so two handlers can each hold a transaction at the same time. A publish on this publisher outside a transaction goes out at once.

A transaction that starts the service, with no delivery to handle, runs from a startup hook: b.after_startup(TransactionalPublish, ..) hands the hook the same publisher.

The order's own delivery​

The transaction covers what the handler publishes. The order's delivery is settled after the handler returns, outside the transaction. A crash between the commit and the settlement delivers the order again, and its ledger entries and invoice are published a second time. Give the messages an id the readers can use to drop a repeat, such as the order's id here.

Testing the transaction​

The tests in the example file run the service in process. The in-process broker publishes nothing before the commit and discards the messages of an abort, so the tests read the outcome from what was published. Testing an AMQP service covers the harness.