DynamoLeaseStore

DynamoLeaseStore keeps leases in a DynamoDB table, so several instances share the shards; the table needs only a string key lease_key.

It needs the dynamodb-lease feature of the crate. DynamoLeaseStore::new takes the AWS configuration and the table name. The table has a string partition key named lease_key and nothing else; on-demand billing fits it. One row holds one shard of one stream, under the key stream:shard, so one table serves every stream of a service. The row holds the owner, the lease's expiry, the checkpoint and a counter bumped on every write. Every write is conditional, so two instances never both hold a shard, and a checkpoint lands only while its writer holds the lease. The checkpoints outlive the process, so a restarted instance resumes each shard where its acknowledgements stopped. The broker is built from the same configuration with KinesisBroker::from_config, so both reach one account and region. The service's credentials need dynamodb:GetItem and dynamodb:UpdateItem on the table. Each instance needs an owner id of its own; a table written by an earlier version of this crate is read through legacy_stream.

Example

Every instance of the fulfilment service runs this code against one table.

Macros
Manual

examples/kinesis_leases.rs
/// Fulfils each order. Every instance runs this handler, and each shard is read by the one
/// instance that holds its lease in the table.
#[subscriber(KinesisStream::new("fulfilment"))]
async fn fulfil(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "fulfilled");
    HandlerOutcome::ack()
}

/// The lease store and the broker share one AWS configuration: one account, one region.
fn broker(config: &SdkConfig, table: &str) -> KinesisBroker {
    let leases = Arc::new(DynamoLeaseStore::new(config, table));
    KinesisBroker::from_config(config.clone()).lease_store(leases)
}

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

examples/manual/kinesis_leases.rs
/// Fulfils each order. Every instance runs this handler, and each shard is read by the one
/// instance that holds its lease in the table.
struct Fulfil;

impl Handle<Order> for Fulfil {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(order.id, "fulfilled");
        ready(Ok(()))
    }
}

/// The lease store and the broker share one AWS configuration: one account, one region.
fn broker(config: &SdkConfig, table: &str) -> KinesisBroker {
    let leases = Arc::new(DynamoLeaseStore::new(config, table));
    KinesisBroker::from_config(config.clone()).lease_store(leases)
}

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

Example​

Example