Statements

#[outbox(custom(fetch, ack, retry, discard, recover))] replaces a record's default statement with the service's own, for a status of its own or a saga.

Each event is a trait of its own, and the registry runs the record's impl in place of the default:

Event	Trait	Default
a delivery takes the record into work	`Fetch`	reads the record while it is unprocessed
the handler acknowledged	`Ack`	marks the record processed
the handler asked again	`Retry`	runs nothing
the handler dropped the message	`Discard`	marks the record processed
the startup republish	`Recover`	reads the unprocessed records of one name

Fetch returns Option<Self>: None means the record is processed or taken already, and the delivery is acknowledged without its handler. So a fetch of the service's own decides what "taken" means, a stamp, a status column or a lease, and the other events release or finish it. Publish has no default: it is always the service's own and is never listed. A default Retry takes no connection; a Retry of the service's own takes one, as the other events do.

By hand, the record's OutboxSpec names each event with .own::<own::Fetch>() and the others, and its type Table lists the same markers of outbox::spec::own. A record type that names an event and lacks its trait does not register, and the error names the missing trait.

Example

OrderOutbox stamps taken_at when a consumer takes the record, and its retry releases the record and counts the attempt; the acknowledgement keeps the default mark. The repository's examples/outbox takes its records with such a fetch on Postgres.

Macros
Manual

examples/sqlx_outbox_statements.rs
use ruststream_sqlx::{Outbox, outbox};
use sqlx::{Sqlite, SqliteConnection};

/// One record of `order_outbox`, taken and released by statements of the service's own.
#[derive(Outbox, sqlx::FromRow)]
#[outbox(table = "order_outbox", custom(fetch, retry))]
struct OrderOutbox {
    #[field(id)]
    id: i64,
    #[field(name)]
    name: String,
    #[field(payload)]
    payload: Vec<u8>,
    #[field(processed_at)]
    processed_at: Option<DateTime<Utc>>,
}

impl outbox::Publish<Sqlite> for OrderOutbox {
    async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO order_outbox (name, payload) VALUES (?, ?) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

/// Takes the record into work, or finds it taken or processed already (`None`).
impl outbox::Fetch<Sqlite> for OrderOutbox {
    async fn fetch(conn: &mut SqliteConnection, id: &i64) -> sqlx::Result<Option<Self>> {
        sqlx::query_as(
            "UPDATE order_outbox SET taken_at = CURRENT_TIMESTAMP
             WHERE id = ? AND taken_at IS NULL AND processed_at IS NULL
             RETURNING id, name, payload, processed_at",
        )
        .bind(id)
        .fetch_optional(conn)
        .await
    }
}

/// Releases a record its consumer asked for again, and counts the attempt.
impl outbox::Retry<Sqlite> for OrderOutbox {
    async fn retry(conn: &mut SqliteConnection, id: &i64) -> sqlx::Result<()> {
        sqlx::query("UPDATE order_outbox SET taken_at = NULL, attempts = attempts + 1 WHERE id = ?")
            .bind(id)
            .execute(conn)
            .await
            .map(drop)
    }
}

examples/manual/sqlx_outbox_statements.rs
use ruststream_sqlx::dialect::Column;
use ruststream_sqlx::outbox::spec::{ProcessedAt, own};
use ruststream_sqlx::outbox::{self, OutboxSpec, OutboxTable};
use sqlx::{Sqlite, SqliteConnection};

/// One record of `order_outbox`, taken and released by statements of the service's own.
#[derive(sqlx::FromRow)]
struct OrderOutbox {
    id: i64,
    name: String,
    payload: Vec<u8>,
}

impl OutboxTable for OrderOutbox {
    type Id = i64;
    type Table = OutboxSpec<(ProcessedAt, own::Fetch, own::Retry)>;
    const TABLE: Self::Table = OutboxSpec::new(
        "order_outbox",
        Column::new("id"),
        Column::new("name"),
        Column::new("payload"),
    )
    .processed_at(Column::new("processed_at"))
    .own::<own::Fetch>()
    .own::<own::Retry>();

    fn id(&self) -> &i64 {
        &self.id
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn payload(&self) -> &[u8] {
        &self.payload
    }
}

impl outbox::Publish<Sqlite> for OrderOutbox {
    async fn publish(conn: &mut SqliteConnection, msg: &OutgoingMessage<'_>) -> sqlx::Result<i64> {
        sqlx::query_scalar("INSERT INTO order_outbox (name, payload) VALUES (?, ?) RETURNING id")
            .bind(msg.name())
            .bind(msg.payload())
            .fetch_one(conn)
            .await
    }
}

/// Takes the record into work, or finds it taken or processed already (`None`).
impl outbox::Fetch<Sqlite> for OrderOutbox {
    async fn fetch(conn: &mut SqliteConnection, id: &i64) -> sqlx::Result<Option<Self>> {
        sqlx::query_as(
            "UPDATE order_outbox SET taken_at = CURRENT_TIMESTAMP
             WHERE id = ? AND taken_at IS NULL AND processed_at IS NULL
             RETURNING id, name, payload",
        )
        .bind(id)
        .fetch_optional(conn)
        .await
    }
}

/// Releases a record its consumer asked for again, and counts the attempt.
impl outbox::Retry<Sqlite> for OrderOutbox {
    async fn retry(conn: &mut SqliteConnection, id: &i64) -> sqlx::Result<()> {
        sqlx::query("UPDATE order_outbox SET taken_at = NULL, attempts = attempts + 1 WHERE id = ?")
            .bind(id)
            .execute(conn)
            .await
            .map(drop)
    }
}

Example​

Example