Skip to main content

Consumer groups and crash recovery

This guide shares a Redis stream between workers and between services, and recovers the entries a stopped worker left unfinished.

The example is a payments service. Payments arrive on the stream payments. The ledger group posts each payment, and the fraud group screens it.

A group of workers​

src/main.rs
/// Longer than the slowest posting ever takes: an entry idle this long belongs to no live worker.
const MIN_IDLE: Duration = Duration::from_secs(30);

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    account: String,
    amount_cents: u64,
}

#[subscriber(
    RedisStream::claiming("payments", MIN_IDLE).group("ledger"),
    workers(8, by_key)
)]
async fn post_payment(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let earlier_deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("0");
    if payment.account.starts_with("migrating-") {
        tracing::warn!(
            payment.id,
            earlier_deliveries,
            "account locked, payment left pending"
        );
        return HandlerOutcome::retry();
    }
    tracing::info!(payment.id, earlier_deliveries, "payment posted");
    HandlerOutcome::ack()
}

A Redis stream is read through a consumer group. The group holds a cursor into the stream, and every entry past the cursor goes to one consumer of the group. Each subscription is one consumer, so every running instance of the service adds a consumer to ledger, and the instances share the payments. A consumer gets a generated name, ruststream-0, ruststream-1 and so on in each process. consumer names it, for example after the host, so XINFO CONSUMERS tells the instances apart.

An entry handed to a consumer goes into the group's pending entries list. It leaves the list when the handler acknowledges it. If the worker stops first, a crash or a deploy, the entry stays pending: Redis keeps it, and no other consumer receives it on its own.

RedisStream::claiming reads both sets in one call: entries pending for at least min_idle, then new ones. So the subscription that takes new payments also takes over the payments of a worker that stopped. It needs Redis 8.4 or later; on an older server the service does not start, and the error names the subscription and both versions.

A retry on this subscription does not append a copy. The entry stays pending under its own id, and the next read claims it once it has been idle min_idle. post_payment uses that for an account that is being migrated: the payment waits in the pending list and comes back every 30 seconds.

A claimed entry carries DELIVERY_COUNT_HEADER: the number of deliveries before this one, zero for a new entry. IDLE_MS_HEADER carries how long the entry sat idle. Retries, delays and dead letters caps the deliveries on this count.

Choosing min_idle​

min_idle has no default. An entry idle that long is treated as abandoned, so set it above the longest time a handler ever runs. A shorter value takes an entry away from a worker that is still on it, and the payment is posted twice. A longer value delays recovery after a crash by the same amount.

A second group​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments.screened")]
struct Screened {
    payment_id: u64,
    suspicious: bool,
}

#[subscriber(
    RedisStream::new("payments")
        .group("fraud")
        .start_id(StreamStart::Beginning),
    reply
)]
async fn screen_payment(payment: &Payment) -> Screened {
    Screened {
        payment_id: payment.id,
        suspicious: payment.amount_cents > 1_000_000,
    }
}

Each group has its own cursor and its own pending list, so each group reads every entry. The fraud group receives every payment the ledger group receives, and neither slows the other. In practice each service that reads a stream has a group of its own.

The service creates a group at startup when it does not exist yet. start_id says where a new group starts: StreamStart::New, the default, reads only entries added after that, and StreamStart::Beginning reads every entry the stream still holds. The fraud group was added after the ledger, so it starts at the beginning and screens the payments already on the stream. An existing group keeps its cursor, and start_id does not move it; start_at does.

The service​

src/main.rs
#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(post_payment);
            b.include(screen_payment).out_reply(Publish);
        },
    )
}

screen_payment replies with a Screened on the stream payments.screened.

Order per account​

workers(8, by_key) runs post_payment on eight lanes. Payments that share a partition key go to the same lane, one after another, so the payments of one account are posted in the order they were added. Redis has no partitions, so the key travels in the PARTITION_KEY_HEADER header. The producer sets it with the partition_key step of its publish. A payment without a key goes to any lane. The order holds inside one consumer; two instances of the service each post their own share of the payments.

Recovery before Redis 8.4​

src/main.rs
#[subscriber(RedisStream::new("payments").group("ledger"))]
async fn post_new_payment(payment: &Payment) -> HandlerOutcome {
    tracing::info!(payment.id, "payment posted");
    HandlerOutcome::ack()
}

#[subscriber(RedisStream::reclaim("payments", MIN_IDLE).group("ledger"))]
async fn recover_payment(payment: &Payment, ctx: &mut Context<'_>) -> HandlerOutcome {
    let deliveries = ctx.headers().get_str(DELIVERY_COUNT_HEADER).unwrap_or("1");
    tracing::warn!(
        payment.id,
        deliveries,
        "payment recovered from a stopped worker"
    );
    HandlerOutcome::ack()
}

fn app_before_8_4() -> impl App {
    RustStream::new(AppInfo::new("payments", "0.1.0")).with_broker(
        RedisBroker::standalone("redis://localhost:6379"),
        |b| {
            b.include(post_new_payment);
            b.include(recover_payment);
            b.include(screen_payment).out_reply(Publish);
        },
    )
}

An older server has no claiming read, so recovery is a second subscription on the same group. RedisStream::new reads new entries. RedisStream::reclaim reads only entries pending for at least min_idle, with XAUTOCLAIM, so it takes over what a stopped worker held. A reclaimed delivery counts itself in DELIVERY_COUNT_HEADER, so the first recovery reports at least 1.

On a new subscription, retry appends a copy of the entry to the stream and acknowledges the original. The copy comes back at once, behind the entries added meanwhile. A wait such as the migration above takes retry_after there, which Retries, delays and dead letters covers.

Testing recovery​

The tests in the example file run the service in process on a paused clock. One publishes a payment for an account under migration, moves the clock by min_idle with tb.advance, and sees the ledger receive the payment a second time, with no copy on the stream. Testing a Redis service covers the harness.