OutboxError

OutboxError names what the outbox could not do: record a publish, read the unprocessed records, or republish one, a publish with no pool yet, and a record a handler's own SQLite transaction locks out.

Each variant names the registered name, and the record type where one is involved:

Variant	When	What follows
`NoPool`	a message under a registered name was published before the registry had a pool	the message is not sent
`Record`	the record's `Publish` failed; its source is the database's error	the message is not sent
`WriteLockHeld`	on SQLite, a transactional handler published after its first write through `Tx`, and its own delivery transaction holds the write lock the record needs (Interplay)	the message is not sent
`Recover`	the startup republish could not read a name's unprocessed records	startup fails
`Republish`	the startup republish could not send a record; it names the record's id	startup fails

A publish through the publish middleware fails with NoPool, Record or WriteLockHeld, as the error of the publisher a handler holds. A publish through wrap fails with TrackedPublishError: Outbox carries the OutboxError, and Publish the wrapped publisher's own error after the record was written, which leaves the record for the next startup. The publish builder wraps either in its own PublishError::Publish. PoolAlreadySet is what a second set_pool returns.

Example

An HTTP endpoint answers each order by what its publish through wrap returned: 202 once the record is written, 503 when it is not, so the client sends the order again.

Macros
Manual

examples/sqlx_outbox_errors.rs
use ruststream::runtime::PublishError;
use ruststream_sqlx::outbox::TrackedPublishError;

/// The status an HTTP endpoint answers an order with.
fn status<E: Display>(published: Result<(), PublishError<TrackedPublishError<E>>>) -> u16 {
    match published {
        Ok(()) => 202,
        // Recorded but not sent: the next startup sends it.
        Err(PublishError::Publish(TrackedPublishError::Publish(error))) => {
            tracing::warn!(%error, "the order waits for the next startup");
            202
        }
        // Neither recorded nor sent: the client sends the order again.
        Err(PublishError::Publish(TrackedPublishError::Outbox(error))) => {
            tracing::warn!(%error, "the order was not recorded");
            503
        }
        Err(error) => {
            tracing::error!(%error, "the order was not published");
            500
        }
    }
}
async fn serve(pool: SqlitePool, orders: &[u64]) -> Result<Vec<u16>, Box<dyn Error + Send + Sync>> {
    let tracking = outbox! { pool: pool, "orders" => OrderOutbox };
    let broker = MemoryBroker::new().bindable();
    let egress = broker.bind(Publish);
    let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(broker, |_b| {});
    let running = app.start().await?;
    let publisher = tracking.wrap(running.publisher(egress).await?);
    let mut answers = Vec::with_capacity(orders.len());
    for &id in orders {
        answers.push(status(
            publisher.message(&OrderPlaced { id }).publish().await,
        ));
    }
    running.shutdown().await?;
    Ok(answers)
}

examples/manual/sqlx_outbox_errors.rs
use ruststream::runtime::PublishError;
use ruststream_sqlx::outbox::TrackedPublishError;

/// The status an HTTP endpoint answers an order with.
fn status<E: Display>(published: Result<(), PublishError<TrackedPublishError<E>>>) -> u16 {
    match published {
        Ok(()) => 202,
        // Recorded but not sent: the next startup sends it.
        Err(PublishError::Publish(TrackedPublishError::Publish(error))) => {
            tracing::warn!(%error, "the order waits for the next startup");
            202
        }
        // Neither recorded nor sent: the client sends the order again.
        Err(PublishError::Publish(TrackedPublishError::Outbox(error))) => {
            tracing::warn!(%error, "the order was not recorded");
            503
        }
        Err(error) => {
            tracing::error!(%error, "the order was not published");
            500
        }
    }
}
async fn serve(pool: SqlitePool, orders: &[u64]) -> Result<Vec<u16>, Box<dyn Error + Send + Sync>> {
    let tracking = Outbox::new(pool).track::<OrderOutbox, Orders>();
    let broker = MemoryBroker::new().bindable();
    let egress = broker.bind(Publish);
    let app = RustStream::new(AppInfo::new("orders", "0.1.0"))
        .publish_layer(tracking.publish_layer())
        .with_broker(broker, |_b| {});
    let running = app.start().await?;
    let publisher = tracking.wrap(running.publisher(egress).await?);
    let mut answers = Vec::with_capacity(orders.len());
    for &id in orders {
        answers.push(status(
            publisher.message(&OrderPlaced { id }).publish().await,
        ));
    }
    running.shutdown().await?;
    Ok(answers)
}

Example​

Example