Tracking

Outbox::layer and publish_layer track messages: the publish records then sends, and the delivery's outcome settles its record by its id.

publish_layer returns a TrackingPublishLayer, mounted with RustStream::publish_layer. It reaches everything a handler publishes, replies and Out slots alike. A message under a registered name is recorded through the record's Publish, then sent with the record's id in OUTBOX_ID_HEADER. A record that fails is not sent, and the publish fails with OutboxError::Record.

layer returns a TrackingLayer, mounted with RustStream::layer. A delivery under a registered name that carries an id is taken into work by the record's fetch, on a connection the layer returns before the handler runs, so the handler can publish tracked messages of its own. The handler's outcome then settles the record:

Outcome	Event	Default statement
acknowledged	`outbox::Ack`	marks the record processed
asked again (`retry`, `retry_after`)	`outbox::Retry`	none: the record waits for the next startup
dropped	`outbox::Discard`	marks the record processed

A record without processed_at is deleted where the default marks it. A record the fetch finds processed already acknowledges the delivery without running the handler, which is how a copy sent twice reaches its consumer once. A fetch that fails retries the delivery without the handler. A settlement that fails logs a warning and keeps the handler's outcome, and the record stays for the next startup. A delivery without the header, with a header that is not an id, or under a name the registry does not hold runs untracked.

Both layers pass a message whose name is not registered after one comparison per registered name.

Example

fulfil acknowledges an order in stock, whose record is marked, and asks again for one out of stock, whose record waits; the subscription takes one attempt, so the broker does not deliver that order again.

Macros
Manual

examples/sqlx_outbox_tracking.rs
use ruststream::memory::prelude::*;
use ruststream_sqlx::outbox;
use sqlx::SqlitePool;

#[subscriber("checkout", reply)]
async fn place(cmd: &PlaceOrder) -> OrderPlaced {
    OrderPlaced {
        id: cmd.id,
        in_stock: cmd.in_stock,
    }
}

/// The outcome settles the record: an acknowledgement marks it, a retry leaves it.
#[subscriber("orders")]
async fn fulfil(order: &OrderPlaced) -> HandlerOutcome {
    if order.in_stock {
        tracing::info!(order = order.id, "fulfilling the order");
        HandlerOutcome::ack()
    } else {
        HandlerOutcome::retry()
    }
}

fn service(pool: SqlitePool) -> impl App {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Takes a delivery that carries a record id into work and settles the record.
        .layer(tracking.layer())
        // Records a message under a registered name, then sends it with the record's id.
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(place).out_reply(Publish);
            // One attempt: an order out of stock is not delivered again, and its record waits.
            b.include(fulfil).max_attempts(nonzero!(1u32));
        })
}

examples/manual/sqlx_outbox_tracking.rs
use ruststream::memory::prelude::*;
use ruststream_sqlx::outbox::Outbox;
use sqlx::SqlitePool;

struct Place;

impl Handle<PlaceOrder, OrderPlaced> for Place {
    fn handle(
        &self,
        cmd: &PlaceOrder,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<OrderPlaced, HandlerOutcome>> {
        ready(Ok(OrderPlaced {
            id: cmd.id,
            in_stock: cmd.in_stock,
        }))
    }
}

/// The outcome settles the record: an acknowledgement marks it, a retry leaves it.
struct Fulfil;

impl Handle<OrderPlaced> for Fulfil {
    fn handle(
        &self,
        order: &OrderPlaced,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if order.in_stock {
            tracing::info!(order = order.id, "fulfilling the order");
            ready(Ok(()))
        } else {
            ready(Err(HandlerOutcome::retry()))
        }
    }
}

/// The name `OrderPlaced` is published under.
struct Orders;

impl TrackedName for Orders {
    const NAME: &'static str = "orders";
}

fn service(pool: SqlitePool) -> impl App {
    let tracking = Outbox::new(pool).track::<OrderOutbox, Orders>();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Takes a delivery that carries a record id into work and settles the record.
        .layer(tracking.layer())
        // Records a message under a registered name, then sends it with the record's id.
        .publish_layer(tracking.publish_layer())
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("checkout", Place).reply().build())
                .out_reply(Publish);
            // One attempt: an order out of stock is not delivered again, and its record waits.
            b.include(subscriber("orders", Fulfil).build())
                .max_attempts(nonzero!(1u32));
        })
}

Example​

Example