Struct DynamoLeaseStore 

pub struct DynamoLeaseStore { /* private fields */ }

A DynamoDB-backed lease store.

The table needs a string partition key named lease_key and nothing else; create it with on-demand billing. Instances race leases with conditional writes and steal only expired ones. One row holds one shard of one stream, under the key stream:shard, so one table serves every stream a service consumes.

Rows from earlier versions

Earlier versions of this crate keyed a row by the shard id alone (shardId-000000000000), and a table written by them still holds those rows. A row keyed by a bare shard id does not say which stream it belongs to, so the store reads one only for the stream named by legacy_stream: when a shard has no checkpoint under its own key, the bare row’s checkpoint is used and copied under the new key. For any other stream the bare rows are ignored. With no stream named, a shard whose bare row holds a checkpoint is refused with an error naming the table, the shard and the stream that asked, and it is not read until the operator names the stream or deletes the row.

The two versions lease different rows, so an instance of each can read the same shard at once: stop the old instances before starting the new ones, or accept the duplicates.

Implementations

impl DynamoLeaseStore

pub fn new(config: &SdkConfig, table: impl Into<String>) -> Self

Uses table in the account and region of config.

Examples

use std::error::Error;
use std::sync::Arc;

use aws_config::BehaviorVersion;
use ruststream_kinesis::prelude::*;

pub async fn run() -> Result<(), Box<dyn Error>> {
    let config = aws_config::defaults(BehaviorVersion::latest()).load().await;
    let leases = Arc::new(DynamoLeaseStore::new(&config, "orders-leases"));
    let broker = KinesisBroker::from_config(config).lease_store(leases);

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(broker, |b| {
            b.include(handle);
        })
        .run()
        .await?;
    Ok(())
}

pub fn legacy_stream(self, stream: impl Into<String>) -> Self

Names the stream the table’s rows keyed by a bare shard id belong to: the rows an earlier version of this crate wrote, before a row was kept per stream and shard.

A shard of stream with no checkpoint under its own key resumes from its bare row, and the first read copies that checkpoint under the new key. Other streams ignore the bare rows. Once the upgraded service has started a subscription on every stream it consumes, every bare row that held progress has its copy, and the bare rows and this setting can go.

A table that served more than one stream before the upgrade holds, in each bare row, whichever stream checkpointed last: no stream can resume from it safely. Delete those rows and open each stream at a chosen position instead.

Examples

use std::error::Error;
use std::sync::Arc;

use aws_config::BehaviorVersion;
use ruststream_kinesis::prelude::*;

pub async fn run() -> Result<(), Box<dyn Error>> {
    let config = aws_config::defaults(BehaviorVersion::latest()).load().await;
    // The table was written by an earlier version for the `orders` stream only.
    let leases =
        Arc::new(DynamoLeaseStore::new(&config, "orders-leases").legacy_stream("orders"));
    let broker = KinesisBroker::from_config(config).lease_store(leases);

    RustStream::new(AppInfo::new("orders", "0.1.0"))
        .with_broker(broker, |b| {
            b.include(handle);
        })
        .run()
        .await?;
    Ok(())
}

Trait Implementations

impl Clone for DynamoLeaseStore

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for DynamoLeaseStore

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl LeaseStore for DynamoLeaseStore

fn acquire<'a>( &'a self, key: &'a LeaseKey, owner: &'a str, ttl: Duration, ) -> BoxFuture<'a, Result<bool, LeaseError>>

Attempts to take the shard’s lease for owner, valid for ttl. Returns false when another live owner holds it.

fn renew<'a>( &'a self, key: &'a LeaseKey, owner: &'a str, ttl: Duration, ) -> BoxFuture<'a, Result<bool, LeaseError>>

Heartbeats the lease. Returns false when the lease is no longer held (fenced).

fn checkpoint<'a>( &'a self, key: &'a LeaseKey, owner: &'a str, sequence: &'a str, ) -> BoxFuture<'a, Result<bool, LeaseError>>

Records sequence as the shard’s checkpoint, conditional on owner holding the lease. Returns false when fenced.

fn read<'a>( &'a self, key: &'a LeaseKey, ) -> BoxFuture<'a, Result<LeaseState, LeaseError>>

Reads the shard’s persisted state (regardless of ownership).

fn release<'a>( &'a self, key: &'a LeaseKey, owner: &'a str, ) -> BoxFuture<'a, Result<(), LeaseError>>

Releases the lease so another owner can take it without waiting for expiry.