consumer-service

A long-lived consumer service.

Consuming is a service, not a component. A wasmCloud component is instantiated per request and torn down after it, which leaves nowhere to hold a record stream, and — more importantly — nothing to drive the group heartbeat between requests. A consumer that stops reading is a consumer the group evicts.

A service instead exports wasi:cli/run, which the host calls once at startup and expects to block indefinitely. That single long-lived call is what owns the stream<consumed-record> and keeps the membership alive.

The shape of run is a loop over the stream:

let consumer = consumer::open(config).await?;
consumer.subscribe(topics).await?;
let (records, done) = consumer.records();
loop {
    select {
        rec = records.next() => { process(rec); consumer.commit(..).await?; }
        ev  = rebalances.next() => { /* commit before revoke */ }
    }
}

Handle rebalances in the same loop. Records and assignment changes arrive on separate streams, and committing an offset for a partition that has already been revoked is where duplicate processing starts.