LeaseStore

A LeaseStore holds each shard's owner and checkpoint: acquire, renew, checkpoint, read and release, with SHARD_END marking a finished shard.

Kinesis keeps no record of who reads a shard or how far a reader got, so the subscription keeps both in a store. KinesisBroker::lease_store installs one, and every subscription of the broker uses it. Without the call the broker uses a MemoryLeaseStore, correct for one instance of a service. Several instances share the shards through a store they all reach, such as DynamoLeaseStore. A subscription acquires a shard for its instance before it reads it, and a shard another live instance holds is left to that instance. A lease lasts ten seconds, and the reader renews it every three. A failed renewal means another instance has taken the shard, and the reader stops at once without writing a checkpoint. An expired lease is free to take, so the shards of an instance that stopped move to the others. Every acknowledgement that moves a shard's checkpoint is one checkpoint call, accepted only while the instance holds the lease. A reader starts at the checkpoint the store reads back as LeaseState, and at the tip when there is none. A shard read to its end is checkpointed as SHARD_END, which lets its children start, and then released. A reader that stops also releases its shard, so the next owner need not wait for the lease to expire. Each call names the shard by its LeaseKey, so one store serves every stream of a broker. A store that fails a call reports a LeaseError; the subscription reports it as KinesisError::Lease naming the stream and the shard, and tries the shard again at its next listing. The in-process transport of a TestApp test consults no store.

Example

The service takes its store from the code that starts it; two instances on one store split the shards of inventory.

Macros
Manual

examples/kinesis_leases_store.rs
/// Applies each stock change. Every instance of the service runs this handler; the lease store
/// decides which instance reads which shard.
#[subscriber(KinesisStream::new("inventory"))]
async fn apply(change: &StockChange) -> HandlerOutcome {
    tracing::info!(change.sku, change.delta, "stock changed");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker, leases: Arc<dyn LeaseStore>) -> RustStream {
    RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(
        broker.lease_store(leases),
        |b| {
            b.include(apply);
        },
    )
}

examples/manual/kinesis_leases_store.rs
/// Applies each stock change. Every instance of the service runs this handler; the lease store
/// decides which instance reads which shard.
struct Apply;

impl Handle<StockChange> for Apply {
    fn handle(
        &self,
        change: &StockChange,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(change.sku, change.delta, "stock changed");
        ready(Ok(()))
    }
}

fn service(broker: KinesisBroker, leases: Arc<dyn LeaseStore>) -> RustStream {
    RustStream::new(AppInfo::new("inventory", "0.1.0")).with_broker(
        broker.lease_store(leases),
        |b| {
            b.include(subscriber(KinesisStream::new("inventory"), Apply).build());
        },
    )
}

Example​

Example