Skip to main content

Running beside an HTTP server, outbox

The storefront takes orders over HTTP and stores them with their events; a relay hands the events to the broker, and the shop runs beside the server.

When the service also serves HTTP, the HTTP server owns the process and the shop runs beside it. start starts the shop in the background and returns once it is running, so a broker that refuses to connect stops the process before the server takes a request. An HTTP request should not publish to the broker itself: a broker outage would fail the request, and an order stored without its event would never reach the shop. A transactional outbox solves both. The request writes the order and its event in one transaction, and a relay publishes the stored events later. A broker outage only delays the events, and none is lost; a crash between a publish and its removal publishes the event again, so the shop sees each order at least once.

The store​

With a real database, the orders and the outbox are two tables written in one transaction. The mutex over the pair plays that role here. Only a real database lets the events survive a crash: these rows live in memory, and a crash loses the ones not yet published.

http.rs
/// The orders the storefront took, and the events the shop still owes the broker. With a real
/// database both writes run in one transaction; the mutex over the pair plays that role here.
#[derive(Default)]
struct Store {
    orders: Vec<Order>,
    outbox: VecDeque<(RequestMeta, Order)>,
}

The HTTP side and the outbox are plain functions, so both tabs show the same code.

The endpoint​

POST /orders takes the order as JSON and the request id from the x-request-id header. It only writes to the store, so it answers 202 Accepted without waiting on the broker. An order past the shop's bounds answers 400 Bad Request and is not stored.

http.rs
/// Takes an order over HTTP. The request only writes to the store, so no broker I/O can fail or
/// stall the response.
async fn place_order(
    State(store): State<Arc<Mutex<Store>>>,
    headers: HeaderMap,
    Json(order): Json<Order>,
) -> (StatusCode, &'static str) {
    let request_id = headers
        .get("x-request-id")
        .and_then(|value| value.to_str().ok())
        .unwrap_or("unknown")
        .to_owned();
    if total_cents(&order.lines).is_none() {
        return (
            StatusCode::BAD_REQUEST,
            "the order is past the shop's bounds\n",
        );
    }
    {
        let mut store = store.lock().await;
        store.orders.push(order.clone());
        store.outbox.push_back((RequestMeta { request_id }, order));
    }
    (StatusCode::ACCEPTED, "accepted\n")
}

The relay​

The relay publishes the oldest event first, with its header contract, and removes it only after the publish succeeded. A refused publish leaves the row for the next round. The relay runs until the process stops it, then runs one last round. The publisher is any Publisher, so the relay does not change with the broker.

http.rs
/// Hands the outbox to the broker, oldest event first. A row leaves the outbox only after its
/// publish succeeded, so a broker outage delays events and loses none.
async fn relay_pending(store: &Mutex<Store>, egress: &impl Publisher) {
    loop {
        let Some((meta, order)) = store.lock().await.outbox.front().cloned() else {
            return;
        };
        let sent = egress
            .message(&order)
            .with_headers(&meta)
            .with_codec(JsonCodec)
            .publish()
            .await;
        if sent.is_err() {
            // The row stays for the next round.
            return;
        }
        store.lock().await.outbox.pop_front();
    }
}

/// Runs the relay every 200 milliseconds until `stop` completes, then runs it once more.
async fn relay_outbox(
    store: Arc<Mutex<Store>>,
    egress: impl Publisher,
    stop: impl Future<Output = ()>,
) {
    let mut rounds = tokio::time::interval(Duration::from_millis(200));
    tokio::pin!(stop);
    loop {
        tokio::select! {
            biased;
            () = &mut stop => break,
            _ = rounds.tick() => relay_pending(&store, &egress).await,
        }
    }
    // The last round runs once no request writes a row any more, so it empties the outbox.
    relay_pending(&store, &egress).await;
}

The process​

bindable lets the process take a publisher token before the shop starts, and publisher turns it into a live publisher once the broker is connected. The server stops on Ctrl+C, or when stopping reports that the shop stopped on its own. Once HTTP has drained, the process stops the relay, and its last round hands over the rows still pending while the shop runs. The shop then drains with shutdown. MemoryBroker keeps its queues in the process's memory as well: an order the last round hands over that the shop has not taken by the time it drains is lost with the process, and a broker that stores its messages keeps it for the next start. A hand-written main installs the logger itself with ruststream::logging::init().

http.rs
#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn Error>> {
    ruststream::logging::init()?;
    let broker = MemoryBroker::new().bindable();
    // A token now, a live publisher once the shop has connected the broker.
    let egress = broker.bind(Publish);
    // The shop starts first: a broker that refuses to connect stops the process before HTTP
    // takes a request.
    let running = shop(broker).start().await?;
    let egress = running.publisher(egress).await?;

    let store = Arc::new(Mutex::new(Store::default()));
    let stop_relay = CancellationToken::new();
    let relay = tokio::spawn(relay_outbox(
        Arc::clone(&store),
        egress,
        stop_relay.clone().cancelled_owned(),
    ));
    let router = axum::Router::new()
        .route("/orders", post(place_order))
        .with_state(store);

    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    tracing::info!("storefront on http://127.0.0.1:8080/orders");
    // HTTP stops on Ctrl+C, or when the shop stops on its own; the shop then drains.
    let stopping = running.stopping();
    axum::serve(listener, router)
        .with_graceful_shutdown(async move {
            tokio::select! {
                _ = tokio::signal::ctrl_c() => {}
                () = stopping => {}
            }
        })
        .await?;
    // HTTP has drained, so no request writes a row any more: the relay's last round hands the
    // pending rows over while the shop still runs, and only then does the shop drain.
    stop_relay.cancel();
    relay.await?;
    running.shutdown().await?;
    Ok(())
}

Run it and place an order:

cargo run --example guide_http --features testing,macros,memory,json,cbor,asyncapi,logging
curl -X POST http://127.0.0.1:8080/orders -H 'content-type: application/json' \
  -H 'x-request-id: req-1' \
  -d '{"id":1,"customer":"ada@example.com","lines":[{"sku":"mug","quantity":12,"price_cents":1200}]}'

The log of chapter 5 follows the order through the shop. A health endpoint reports the shop's state to an orchestrator from the same server.