Consuming events

A consumer is a list of event handlers that all read one Iggy topic. You write one handler per event type, register them, and call start; wakuwaku spawns one task per assigned partition and runs the pipeline described on the overview.

Event handlersCopy Anchor Link

A handler implements EventHandler<E> for an event type E:

pub trait EventHandler<E: Event>: Send + Sync + 'static {
    const MAX_BATCH: NonZeroUsize;

    fn handle(&self, run: &[E::Target]) -> impl Future<Output = Result<(), HandleError>> + Send;
}

• run holds events of one key, in the order they must be applied, already reduced with E's algebra. The element type is E::Target, the decoded body; for an event that is its own body (the usual case) that is simply E.
• MAX_BATCH caps the length of run. Longer runs are passed in consecutive chunks of at most MAX_BATCH events, in order. Use NonZeroUsize::MIN (1) to get one event per call.
• The return value decides what happens on failure, through the ErrorClass of the HandleError; see below.

You can write handle as an async fn. Its future must be Send, because handlers of different keys run concurrently on Tokio, so do not hold a std::sync::MutexGuard across an .await.

Here are the two handlers for the events from Defining events. Both are built from an OrderStore, a stand-in for your database pool:

/// Writes new orders.
pub struct RecordOrder {
    store: OrderStore,
}

impl ServiceCreation for RecordOrder {
    type Dep = OrderStore;
    fn create_service(store: OrderStore) -> Self {
        Self { store }
    }
}

impl EventHandler<OrderPlaced> for RecordOrder {
    const MAX_BATCH: NonZeroUsize = NonZeroUsize::new(100).unwrap();

    async fn handle(&self, run: &[OrderPlaced]) -> Result<(), HandleError> {
        for order in run {
            self.store.insert(order.order_id, order.total_cents).await;
        }
        Ok(())
    }
}

/// Applies payments. A run arrives already folded into one payment.
pub struct ApplyPayment {
    store: OrderStore,
}

impl ServiceCreation for ApplyPayment {
    type Dep = OrderStore;
    fn create_service(store: OrderStore) -> Self {
        Self { store }
    }
}

impl EventHandler<PaymentReceived> for ApplyPayment {
    const MAX_BATCH: NonZeroUsize = NonZeroUsize::MIN;

    async fn handle(&self, run: &[PaymentReceived]) -> Result<(), HandleError> {
        for payment in run {
            self.store
                .pay(payment.order_id, payment.amount_cents)
                .await
                .map_err(HandleError::retryable)?;
        }
        Ok(())
    }
}

PaymentReceived is Associative, so the payments of one order that fall into the same run reach ApplyPayment as a single, summed event. Because OrderPlaced is Acquire, a payment published after its order is never applied before it.

Handlers are plain values, so a unit test calls handle directly, without Iggy:

#[cfg(test)]
mod tests {
    use super::*;
    use wakuwaku::iggy::error::ErrorClass;

    #[tokio::test]
    async fn payment_for_unknown_order_is_retryable() {
        let handler = ApplyPayment::create_service(OrderStore::default());
        let payment = PaymentReceived { order_id: 7, amount_cents: 500 };
        let error = handler.handle(&[payment]).await.unwrap_err();
        assert_eq!(error.class(), ErrorClass::Retryable);
    }
}

Registering through the service builderCopy Anchor Link

Handlers usually need shared dependencies: a database pool, an HTTP client, a Publisher. Implement ServiceCreation for the handler, as above, and let ServiceBuilder::iggy_consumer build it from a dependency already registered in the service builder (a handler that needs several services is composed instead):

use std::sync::Arc;
use wakuwaku::ServiceBuilder;
use wakuwaku::deps::iggy::prelude::{Client, IggyClient};
use wakuwaku::iggy::ConsumerConfig;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Arc::new(IggyClient::from_connection_string(
        "iggy://iggy:iggy@localhost:8090",
    )?);
    client.connect().await?;

    // Consume both event types of the `orders` topic with one consumer.
    // The topic has one partition; Iggy partition ids are zero-based.
    let (_services, consumers) = ServiceBuilder::new(OrderStore::default())
        .iggy_consumer::<RecordOrder, _, _>()
        .iggy_consumer::<ApplyPayment, _, _>()
        .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
        .await?;

    tokio::signal::ctrl_c().await?;
    consumers.shutdown().await;
    Ok(())
}

The three type parameters are <Handler, Event, Position>:

• Handler is the handler type to build.
• Event can be _ when the handler implements EventHandler for exactly one event. Name it when one handler type handles several events: .iggy_consumer::<MyHandler, MyEvent, _>().
• Position is where the handler's Dep sits in the builder. _ works when that type is registered once; otherwise name it with Here / There, as for any provide call. A dependency that was never registered is a compile error.

Each iggy_consumer call builds its handler immediately: it clones the Dep out of the builder, calls create_service, and wraps the handler in an Arc. Calls to iggy_consumer and compose_iggy_consumer chain in any order, and the chain ends with either:

• start(client, config), which starts consuming and returns the service builder together with the running ConsumerRuntime, so you can keep building the rest of the service; or
• into_parts(), which returns the service builder and the unstarted IggyConsumerRegisterCenter, to start later:

let (services, orders) = ServiceBuilder::new(OrderStore::default())
    .iggy_consumer::<RecordOrder, _, _>()
    .iggy_consumer::<ApplyPayment, _, _>()
    .into_parts();

// ... finish setting up the service from `services`, then:
let consumers = orders
    .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
    .await?;

Handlers that need several servicesCopy Anchor Link

iggy_consumer builds a handler from one registered dependency. A handler that needs several, or one you construct by hand, is registered with compose_iggy_consumer instead. Its closure borrows the service builder, takes what it needs with provide, and returns the handler in an Arc. The handler needs no ServiceCreation impl, and it is not registered as a service, so it can't be another component's dependency.

Say recording an order should also mail a confirmation. The handler now needs the OrderStore and a Mailer (another stand-in, a cheap-to-clone handle):

/// Writes new orders and mails a confirmation.
pub struct RecordAndConfirmOrder {
    pub store: OrderStore,
    pub mailer: Mailer,
}

impl EventHandler<OrderPlaced> for RecordAndConfirmOrder {
    const MAX_BATCH: NonZeroUsize = NonZeroUsize::new(100).unwrap();

    async fn handle(&self, run: &[OrderPlaced]) -> Result<(), HandleError> {
        for order in run {
            self.store.insert(order.order_id, order.total_cents).await;
            self.mailer
                .confirm(&order.customer, order.order_id)
                .await
                .map_err(HandleError::retryable)?;
        }
        Ok(())
    }
}

let (_services, consumers) = ServiceBuilder::new(OrderStore::default())
    .push(Mailer::default())
    .compose_iggy_consumer(|s| {
        Arc::new(RecordAndConfirmOrder {
            store: s.provide::<OrderStore, _>().clone(),
            mailer: s.provide::<Mailer, _>().clone(),
        })
    })
    .iggy_consumer::<ApplyPayment, _, _>()
    .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
    .await?;

The handler and event types are inferred from the returned Arc. When the handler implements EventHandler for several events, name the event: .compose_iggy_consumer::<_, MyEvent>(|s| …). A handler built before the builder, or shared with other code, is moved in as is: .compose_iggy_consumer(move |_| handler). The other compose_* methods of the service builder work the same way; see Composing components with a closure.

Registering directlyCopy Anchor Link

ServiceBuilder::iggy_consumer is a convenience over IggyConsumerRegisterCenter, a type-level list of handlers. You can use it directly, for example when your service has no ServiceBuilder:

pub async fn start_by_hand(
    client: Arc<IggyClient>,
    store: OrderStore,
) -> Result<ConsumerRuntime, Error> {
    IggyConsumerRegisterCenter::new(Arc::new(RecordOrder::create_service(store.clone())))
        .push(Arc::new(ApplyPayment::create_service(store)))
        .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
        .await
}

new starts the list with its first handler and every push adds one. Handlers are passed as Arc, so one handler value can serve several event types. When the handler type implements EventHandler for more than one event, name the event. Here Audit implements both EventHandler<OrderPlaced> and EventHandler<PaymentReceived>:

let audit = Arc::new(Audit);
let handlers = IggyConsumerRegisterCenter::<_, OrderPlaced>::new(audit.clone())
    .push::<_, PaymentReceived>(audit);

One main topic per consumerCopy Anchor Link

A consumer reads exactly one main topic: the SUPER_PARTITION_NAME of its events' key type. Every handler in one list must therefore handle events keyed by key types with the same topic name. start checks this, together with the rest of the setup, and returns an error instead of consuming:

To consume events of another key type, start a second consumer. Records whose tag no registered handler matches are dropped and logged, and the offset moves past them. A consumer that registers only some of a topic's event types therefore simply skips the others, which is fine on purpose but a silent loss by mistake: when you add an event type to a topic, also register its handler in the topic's consumer.

ConfigurationCopy Anchor Link

ConsumerConfig::new(stream, consumer_name, partitions) sets the three required fields and defaults the rest. All fields are public:

Override fields with struct update syntax. retry_delays is a &'static slice, so declare custom delays as a const:

use std::time::Duration;
use wakuwaku::iggy::{ConsumerConfig, DeliveryMode};

const RETRY_DELAYS: &[Duration] = &[
    Duration::from_secs(1),
    Duration::from_secs(5),
    Duration::from_secs(30),
];

pub fn config(partitions: Vec<u32>) -> ConsumerConfig {
    ConsumerConfig {
        mode: DeliveryMode::NoEventLose,
        retry_delays: RETRY_DELAYS,
        poll_count: 500,
        poll_interval: Duration::from_millis(250),
        ..ConsumerConfig::new("shop", "order-projection", partitions)
    }
}

Partitions and tasksCopy Anchor Link

start spawns one Tokio task per entry in partitions. Partitions are processed in parallel with each other; inside a partition, the keys of one polled batch are processed concurrently, and the next batch is polled only when every key of the current one is done, failed over to the retry topic, or dropped.

partitions is an explicit assignment, by design: the consumer reads the partitions you list and no others. There is intentionally no "all partitions" mode, and the list should not be read from the topic either. A consumer that silently follows the topic's partition count can't be split safely across replicas, and partitions added to the topic later would be picked up without anyone assigning them.

Iggy partition ids are zero-based: for the Iggy server versions wakuwaku supports, a topic with n partitions has the ids 0..n. Publishers spread events over every partition of the topic by key hash, so a partition that no consumer lists is not consumed at all.

One partition for a small deploymentCopy Anchor Link

A small deployment creates its topic with one partition (see What must exist on the server) and consumes vec![0], as the examples on this page do. One partition task already processes the keys of each batch concurrently, which is enough for most services.

Splitting across replicasCopy Anchor Link

When one partition no longer keeps up, create the topic with more partitions, or grow it (which moves keys to other partitions, see Publishing), and assign the partitions to replicas explicitly. Each replica reads its assignment from its own configuration, for example an environment variable, and passes it to config:

use std::env;
use std::error::Error;

/// Partitions assigned to this replica, e.g. `ORDER_PARTITIONS=0,2`.
pub fn assigned_partitions() -> Result<Vec<u32>, Box<dyn Error>> {
    env::var("ORDER_PARTITIONS")?
        .split(',')
        .map(|id| Ok(id.trim().parse::<u32>()?))
        .collect()
}

With four partitions and two replicas, replica i of n taking the ids with id % n == i gives:

The pipeline of one partitionCopy Anchor Link

Each partition task repeats:

1. Drain the retry partition. Read partition p of retry_<topic> to its end and queue the records by key.
2. Run due retries. Re-plan and re-run every quarantined key whose next retry delay has elapsed.
3. Poll. Fetch up to poll_count records of partition p of the main topic, starting after the stored offset. If none arrive, sleep for poll_interval and start over.
4. Plan. Parse the headers, group the records by key hash, split each key into segments at its ordering fences and group segments into runs by type. See Ordering.
5. Divert quarantined keys. A key that still has records waiting in the retry topic is not executed; its new records are appended to the retry topic behind the old ones, so they stay in order.
6. Execute. Run the other keys concurrently. For each run: decode the bodies, reduce them with the algebra, and call the handler chunk by chunk.
7. Write failures to retry. Keys that failed with a retryable error are written to the retry topic from where they failed, and become quarantined.
8. Commit. Store the main offset as the delivery mode requires.

Failures of Iggy itself (polling, storing an offset, writing to the retry topic) do not stop the task: each call is retried with retry_delays until it succeeds, with a warning logged through tracing.

Delivery modesCopy Anchor Link

DeliveryMode decides when offsets are stored, and therefore what a crash costs:

With NoEventLose, write handlers to be idempotent: applying the same event twice must be harmless. Also note that a retry attempt that fails after some chunks of a run succeeded applies those chunks again on the next attempt. Choose NoDoubleConsume only when losing events is cheaper than repeating them, such as for best-effort notifications.

When a handler failsCopy Anchor Link

Return a HandleError built with one of three constructors; its class decides what happens:

Any error convertible into anyhow::Error can be wrapped, which includes every std::error::Error + Send + Sync + 'static. Retry scheduling, quarantine and the retry topic are covered in Error handling.

ShutdownCopy Anchor Link

A ConsumerRuntime stops in one of three ways:

What a stop costs depends on the delivery mode:

shutdown waits as long as the work in flight takes. A handler that keeps returning Transient is retried until it succeeds, and Iggy calls (storing an offset, writing to the retry topic) are retried while the server is unreachable. Bound the wait with a timeout that fits your platform's grace period (Kubernetes, for example, kills a pod 30 seconds after asking it to stop, by default):

use std::time::Duration;

tokio::signal::ctrl_c().await?;
if tokio::time::timeout(Duration::from_secs(20), consumers.shutdown())
    .await
    .is_err()
{
    eprintln!("consumer shutdown timed out; the remaining work was aborted");
}

When the timeout fires, the unfinished shutdown future is dropped together with the runtime it owns, which aborts the tasks still running: the work that did not finish costs what an abort costs. Use abort directly only when the work in flight can't be waited for at all.

Next stepsCopy Anchor Link