Decoding

A claimed row that does not decode is settled by the subscription's decode policy before the handler runs, and still spends its attempt.

A row decodes in two steps: the driver reads its columns into the struct, and in payload mode the codec reads the payload into the handler's type. A failure at either step reaches the subscription's on_failure(decode = ..) policy, set with FailurePolicies::with_decode. The policy settles the row, and the subscription goes on with the next one. The default policy drops the row: it is deleted, or marked where the table has processed_at. A policy that retries it keeps the row in the table, after a delay with FailurePolicy::RetryAfter.

Such a row still reports its attempt: the claim reads the attempt column alone, so max_attempts(n) spends it like any other row, and a retrying policy keeps it only up to the cap. When the attempt column itself does not decode, the row reports no attempt. A claimed id that the service's own fetch returned no row for settles the same way. A row whose id does not decode fails the claim, and every claim that reaches it fails the same way until the row is fixed; the error names the subscription and the table.

Example

A row that does not decode stays in the table and comes back every ten minutes, until a release that reads it ships.

Macros
Manual

examples/sqlx_messages_decoding.rs
/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    // A row that does not decode stays in the table and comes back every ten minutes, until a
    // release that reads it ships.
    let keep = FailurePolicy::RetryAfter(Duration::from_secs(600));
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<SendEmail>("emails"), |b| {
            b.include(send.on_failure(FailurePolicies::default().with_decode(keep)));
        })
}

examples/manual/sqlx_messages_decoding.rs
/// The service on the database `url` names.
fn service(url: &str) -> RustStream {
    let schema_url = url.to_owned();
    // Nothing connects here: the broker builds its pool when the service starts.
    let broker = SqlxBroker::from_url(url, SqlitePoolOptions::new());
    // A row that does not decode stays in the table and comes back every ten minutes, until a
    // release that reads it ships.
    let keep = FailurePolicy::RetryAfter(Duration::from_secs(600));
    RustStream::new(AppInfo::new("mailer", "0.1.0"))
        // The table, created before the subscription checks it.
        .on_startup(async move |()| {
            let mut conn = SqliteConnection::connect(&schema_url).await?;
            sqlx::raw_sql(SCHEMA).execute(&mut conn).await?;
            conn.close().await
        })
        .with_broker(broker.route::<SendEmail>("emails"), |b| {
            let send = subscriber("emails", Sender)
                .map_source(|_| InboxQueue::<SendEmail>::new("emails"))
                .on_failure(FailurePolicies::default().with_decode(keep));
            b.include(send.build());
        })
}

Example​

Example