Settlement

An ack is a per-shard checkpoint: it moves only when every earlier record is handled; drop retires a record, and a requeue is unsupported.

A shard is an ordered log, and the service keeps no per-record state for a reader. So the subscription remembers one position per shard, the checkpoint, in the lease store. HandlerOutcome::ack marks the record handled. The checkpoint moves past it once every earlier record of the shard is handled too, and the lease store records the new position. HandlerOutcome::drop marks the record handled the same way, which is how a record the handler can never process is retired. HandlerOutcome::retry asks for a requeue, which a shard cannot perform; the runtime logs the refusal as a warning. The record stays unhandled, and the checkpoint stops in front of it while later records are still delivered and handled. It is read again, with every record after it, when the shard's lease is next taken: after a restart or when another instance takes the shard over. That needs a store that keeps checkpoints across a restart, such as DynamoLeaseStore; the default MemoryLeaseStore forgets them, and a restarted service opens each shard at the tip. Within one run the record does not come back. A registration that declares max_attempts turns retry into a counted copy published back to the stream, as retry_after does after a delay. Delivery is at-least-once: a record handled after the held one is delivered again when the shard is read from the checkpoint. A seek starts the checkpoint bookkeeping of the shards it moves afresh. In a TestApp test the in-process transport consults no lease store, and a held record does not come back there either.

Example

An entry is booked, an empty one is retired, and a refund holds the checkpoint until the payment it returns is booked.

Macros
Manual

examples/kinesis_settlement_outcomes.rs
/// Books each entry. An empty entry is retired; a refund waits for the payment it returns, and
/// holds the shard's checkpoint in front of it until then.
#[subscriber(KinesisStream::new("ledger"))]
async fn book(entry: &Entry) -> HandlerOutcome {
    match entry.amount {
        0 => HandlerOutcome::drop(),
        amount if amount < 0 => HandlerOutcome::retry(),
        amount => {
            tracing::info!(entry.id, amount, "booked");
            HandlerOutcome::ack()
        }
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
        b.include(book);
    })
}

examples/manual/kinesis_settlement_outcomes.rs
/// Books each entry. An empty entry is retired; a refund waits for the payment it returns, and
/// holds the shard's checkpoint in front of it until then.
struct Book;

impl Handle<Entry> for Book {
    fn handle(
        &self,
        entry: &Entry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        ready(match entry.amount {
            0 => Err(HandlerOutcome::drop()),
            amount if amount < 0 => Err(HandlerOutcome::retry()),
            amount => {
                tracing::info!(entry.id, amount, "booked");
                Ok(())
            }
        })
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("ledger", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(KinesisStream::new("ledger"), Book).build());
    })
}

Example​

Example