Skip to main content

Batch processing

This guide hands a handler many messages at once, for work that is cheaper per batch than per message, such as one write to a warehouse.

Pub/Sub delivers messages one by one and settles each on its own. The service collects the deliveries of one subscription into a batch on its side, calls the handler once for the batch, and then settles every message of it.

The example loads page clicks into an analytics warehouse.

The batch handler​

src/main.rs
/// Loads a batch of clicks into the warehouse in one write. A click from a user the warehouse
/// does not know yet comes back on its own; a click without a page is dropped.
#[subscriber(
    GooglePubSub::new("clicks-loader")
        .create_with_topic("clicks")
        .batch_wait(Duration::from_millis(500))
)]
async fn load(clicks: &[Click]) -> Vec<HandlerOutcome> {
    tracing::info!(count = clicks.len(), "clicks loaded");
    clicks.iter().map(settle).collect()
}

fn settle(click: &Click) -> HandlerOutcome {
    if click.page.is_empty() {
        HandlerOutcome::drop()
    } else if click.user == UNKNOWN_USER {
        HandlerOutcome::retry()
    } else {
        HandlerOutcome::ack()
    }
}

A slice parameter, &[Click], makes load a batch handler. batch_wait on the descriptor is how long a batch short of its size waits for more messages, half a second here. A batch is handed over as soon as it is full, or when the wait is over with what it has.

load returns one outcome per click, in the order of the batch. A click from a user the warehouse does not know yet is retried: Pub/Sub delivers that click again, and it arrives on its own or in a later batch. A click without a page is dropped, and the rest are acknowledged. A single HandlerOutcome settles the whole batch the same way. In the manual form Ok(()) acknowledges the batch, and Err carries the outcomes click by click.

The mount site​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("clicks", "0.1.0")).with_broker(
        PubSubBroker::new("my-project").emulator("localhost:8085"),
        |b| {
            // At most a hundred clicks per call; a click refused five times goes to `clicks-dead`.
            b.include(load.batch(nonzero!(100)))
                .max_attempts(nonzero!(5u32))
                .dead_letter("clicks-dead");
        },
    )
}

batch(nonzero!(100)) is the batch size, and a batch handler does not compile without it. Every message of a batch is leased until the batch is settled, so the batch size stays under the subscription's max_outstanding, 1000 by default. The dead-letter declaration counts each click on its own: after five deliveries a click goes to clicks-dead, however many batches it rode in. Acknowledgement and retries covers the dead-letter topic.

Testing​

The tests publish clicks together with tokio::join!, so they meet in one batch, and read the batches back with assert_batch_sizes and batches. A retried click shows up as a batch of one: the batch of two, then the stranger's click alone four more times, and then it is on clicks-dead. Testing a Pub/Sub service covers the harness.