Error handling

A consumer in production meets three very different kinds of failure: the database is down, one particular key hit a conflict, or an event can never be applied. wakuwaku makes you say which one happened, and reacts accordingly. Your handler returns a HandleError, and its ErrorClass decides what the consumer does next.

HandleErrorCopy Anchor Link

HandleError wraps any error that converts into anyhow::Error (every std::error::Error + Send + Sync + 'static does) together with a class. Build it with one of the constructors:

HandleError::transient(error)      // ErrorClass::Transient
HandleError::retryable(error)      // ErrorClass::Retryable
HandleError::unrecoverable(error)  // ErrorClass::Unrecoverable
HandleError::new(ErrorClass::Retryable, error)

Both types live in wakuwaku::iggy::error. HandleError::class and HandleError::source give the class and the wrapped error back, which is handy in tests.

The three classesCopy Anchor Link

TransientCopy Anchor Link

Use Transient when nothing would succeed right now: the database or a required service is unreachable. The consumer calls your handler again with the same chunk, sleeping between attempts according to ConsumerConfig::retry_delays: retry_delays[0] after the first failure, retry_delays[1] after the second, and so on; once the list is exhausted its last delay repeats forever. With the defaults that is 2 s, 4 s, 8 s, 16 s, 16 s, 16 s, …

There is no limit on the number of attempts. While the chunk is retried, its key cannot advance, the batch cannot finish, and the partition task polls neither the main topic nor the retry topic. Other keys of the same batch still finish their current work, then wait. This is deliberate: when the database is down, moving on would only fail every other key too.

The consumer's own Iggy calls (polling, storing offsets, writing retry records) are retried the same way, with the same delays.

RetryableCopy Anchor Link

Use Retryable when this key failed but others are fine, and a later attempt may succeed: a deadlock, a serialization failure, a row that another stream has not created yet.

When a run fails with Retryable while the consumer processes the main topic:

1. The key stops. Its other runs in the same stage still finish (body runs are concurrent), but later stages and segments of the key are not started.

2. Unfinished work goes to the retry topic. The consumer writes to partition p of retry_<topic> (for main partition p):

   • the failed chunk and every later chunk of the failed run;
   • every run of the later stages of the failed segment, and every later segment of the key.

   Runs that completed are not written again. Everything is written already reduced, with orderings rewritten so that planning the records again keeps every order the original plan enforced.

3. The key is quarantined. While a key has records pending in the retry topic, its new events from the main topic are not executed; they are diverted to the retry topic behind the failed ones. The per-key order from Ordering is therefore preserved.

4. The key is retried. Before each poll of the main topic, the partition task runs one attempt for every retry key that is due. An attempt plans and executes all of the key's pending records, as if they were a fresh batch.

The schedule comes from retry_delays again: the first attempt is due retry_delays[0] after the failure, and after each failed attempt the next one is due after the next delay. A key is therefore retried once after each delay. With the default DEFAULT_RETRY_DELAYS of 2, 4, 8 and 16 seconds:

Attempts run no earlier than that, between polls of the main topic. If the fourth attempt fails, the consumer gives up: it drops the records of the runs that failed and logs giving up retry at error level. The key's remaining pending records (for example events diverted while it was quarantined) are not dropped: they are attempted right away, and the attempt counter starts over. Once a key has no pending records, its quarantine is lifted.

A few details worth knowing:

• The retry topic is created on start with the main topic's partition count. If it already exists with a different count, start fails with Error::PartitionCountMismatch.
• The failure time is stored in each retry record's header, so after a restart the first attempt is still due retry_delays[0] after the original failure (immediately if that has passed). The attempt counter itself is kept in memory: after a restart, pending keys get the full schedule again.
• ConsumerConfig::max_pending_retry (default 100 000) caps the records waiting in one retry partition. When a newly failing key would exceed it, that key's failed events are dropped and logged (retry backlog full) instead of being written. Events of keys that are already quarantined are still diverted.

UnrecoverableCopy Anchor Link

Use Unrecoverable when retrying can never help: the event is invalid, refers to something that will never exist, or hits a bug. The consumer drops the chunk, logs dropping chunk after unrecoverable handler failure at error level, and continues with the next chunk of the run. The key is not stopped.

Records that never reach your handlerCopy Anchor Link

Some records are dropped by the consumer before any handler sees them. Each is logged with tracing at error level:

• Bad headers: a missing or malformed type tag, key or properties header, or an unknown header version.
• Unknown type tag: no handler is registered for the record's type.
• Algebra mismatch: the algebra in the header differs from the marker of the registered type (see Algebra pitfalls).
• Undecodable body: MessageDe::from_bytes fails. The rest of the run is still processed.
• Unserializable retry record: an event that has to be written to the retry topic can't be serialized again.

Since these are dropped rather than retried, test your bodies' (de)serialization and keep schema changes backward compatible (see Defining events).

Classifying common errorsCopy Anchor Link

Rows that "will exist later" deserve a remark: events of different keys are never ordered, and events of different topics even less so. If the stock row for a SKU is created by a ProductCreated event keyed by something else, a StockAdjusted event can arrive first. Retryable covers that gap, as long as it closes within the retry schedule.

A classifier for sqlx errors on PostgreSQL might look like this:

/// Sort a database error into the class the consumer should react with.
fn classify(error: sqlx::Error) -> HandleError {
    match &error {
        // The database is unreachable: every key would fail the same way.
        sqlx::Error::PoolTimedOut
        | sqlx::Error::PoolClosed
        | sqlx::Error::Io(_)
        | sqlx::Error::Tls(_) => HandleError::transient(error),
        // Serialization failure, deadlock or lock timeout
        // (PostgreSQL SQLSTATE 40001 / 40P01 / 55P03):
        // this key lost a race and may well succeed later.
        sqlx::Error::Database(db)
            if matches!(db.code().as_deref(), Some("40001" | "40P01" | "55P03")) =>
        {
            HandleError::retryable(error)
        }
        // Anything else is a bug or bad data; retrying won't change it.
        _ => HandleError::unrecoverable(error),
    }
}

And a handler that uses it, turning "no row updated" into a Retryable domain error (StockAdjusted is the event from the Algebra page):

/// The event names a SKU that has no stock row (yet).
#[derive(Debug)]
struct UnknownSku(String);

impl fmt::Display for UnknownSku {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "no stock row for SKU {}", self.0)
    }
}

impl std::error::Error for UnknownSku {}

pub struct ApplyStock {
    pool: PgPool,
}

impl EventHandler<StockAdjusted> for ApplyStock {
    const MAX_BATCH: NonZeroUsize = NonZeroUsize::MIN;

    async fn handle(&self, run: &[StockAdjusted]) -> Result<(), HandleError> {
        for adjustment in run {
            let updated = sqlx::query("UPDATE stock SET quantity = quantity + $1 WHERE sku = $2")
                .bind(adjustment.delta)
                .bind(&adjustment.sku)
                .execute(&self.pool)
                .await
                .map_err(classify)?;
            if updated.rows_affected() == 0 {
                // The `ProductCreated` projection may simply not have caught up.
                return Err(HandleError::retryable(UnknownSku(adjustment.sku.clone())));
            }
        }
        Ok(())
    }
}

The imports used by the snippets on this page:

use std::fmt;
use std::num::NonZeroUsize;
use std::time::Duration;
use wakuwaku::deps::sqlx::{self, PgPool};
use wakuwaku::iggy::error::{ErrorClass, HandleError};
use wakuwaku::iggy::{ConsumerConfig, EventHandler};

(sqlx is available as wakuwaku::deps::sqlx with the sqlx-pg or sqlx-mysql feature; see sqlx.)

Tuning the retry scheduleCopy Anchor Link

Both schedules, in-place Transient retries and Retryable attempts, come from the same list. Set it, and the backlog cap, on the ConsumerConfig:

const RETRY_DELAYS: &[Duration] = &[
    Duration::from_secs(1),
    Duration::from_secs(10),
    Duration::from_secs(60),
    Duration::from_secs(300),
];

pub fn config() -> ConsumerConfig {
    let mut config = ConsumerConfig::new("shop", "inventory", vec![0, 1, 2]);
    config.retry_delays = RETRY_DELAYS;
    config.max_pending_retry = 10_000;
    config
}

With the schedule above, a Retryable key gets four attempts over about six minutes, and a Transient outage is retried every five minutes once the first three delays have passed.

Design handlers to be idempotentCopy Anchor Link

Even without bugs, a handler can see the same events more than once:

• Redelivery after a crash. In the default DeliveryMode::NoEventLose, offsets are stored after a batch is done, so events processed since the last stored offset are processed again after a crash.
• Transient retries. The same chunk is passed again. Whatever your handler wrote before failing is written again.
• Retry attempts. If a run fails in a retry attempt after some of its chunks succeeded, the next attempt applies those chunks again.

So:

• Make each handle call all-or-nothing, for example one database transaction per call. Then a failed call leaves nothing behind to repeat.
• Make the effect idempotent where you can: upserts (INSERT … ON CONFLICT DO NOTHING), "set to" instead of "add", conditional updates on a version column.
• Where you can't (counters, outgoing e-mails), carry a unique id in the event body and record processed ids in the same transaction. Note that a unique id makes no two events equal, which turns off Idempotent collapsing, and that an Associative combine must then decide which ids the folded event stands for.
• If losing events on a crash is acceptable but duplicates are not, DeliveryMode::NoDoubleConsume stores offsets before processing instead. See Consuming events.

Next stepsCopy Anchor Link