Replay and seeking

This guide reads a stream again from its oldest record or from a moment, and moves a running subscription from inside its handler.

Kinesis keeps a record after a consumer has read it, until the stream's retention period ends. A service can read it again: to rebuild a read model, to backfill a new consumer, or to skip records it cannot use. A subscription takes its start position at the mount site, and a handler moves its own subscription through the delivery's context.

The read models

The service keeps two read models in memory: the balance of every account and the spend of every card. Each posting carries the ledger's revision. A correction of past postings is published as a new revision, and a read model that meets one rebuilds itself from the oldest record of the stream.

Macros
Manual

src/main.rs
/// One posting of the ledger. `revision` is the ledger's revision: a correction of past
/// postings is published as a new revision, and a read model meeting one rebuilds itself.
#[derive(Debug, PartialEq, Deserialize)]
struct Posting {
    id: u64,
    account: String,
    amount_cents: i64,
    revision: u32,
}

/// A read model kept in memory: the balance of every account.
#[derive(Debug, Default)]
struct Projection {
    revision: u32,
    balances: HashMap<String, i64>,
}

impl Projection {
    /// Applies `posting`, or clears the model and answers `false` when the posting belongs to a
    /// newer revision, which the model has to be rebuilt for from the start of the log.
    fn apply(&mut self, posting: &Posting) -> bool {
        if posting.revision > self.revision {
            self.revision = posting.revision;
            self.balances.clear();
            return false;
        }
        *self.balances.entry(posting.account.clone()).or_default() += posting.amount_cents;
        true
    }
}

/// The service's state: one read model per subscription that builds one.
#[derive(Clone, Default)]
struct Projections {
    balances: Arc<Mutex<Projection>>,
    spend: Arc<Mutex<Projection>>,
}

src/main.rs
/// One posting of the ledger. `revision` is the ledger's revision: a correction of past
/// postings is published as a new revision, and a read model meeting one rebuilds itself.
#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
struct Posting {
    id: u64,
    account: String,
    amount_cents: i64,
    revision: u32,
}

/// A read model kept in memory: the balance of every account.
#[derive(Debug, Default)]
struct Projection {
    revision: u32,
    balances: HashMap<String, i64>,
}

impl Projection {
    /// Applies `posting`, or clears the model and answers `false` when the posting belongs to a
    /// newer revision, which the model has to be rebuilt for from the start of the log.
    fn apply(&mut self, posting: &Posting) -> bool {
        if posting.revision > self.revision {
            self.revision = posting.revision;
            self.balances.clear();
            return false;
        }
        *self.balances.entry(posting.account.clone()).or_default() += posting.amount_cents;
        true
    }
}

/// The service's state: one read model per subscription that builds one.
#[derive(Clone, Default)]
struct Projections {
    balances: Arc<Mutex<Projection>>,
    spend: Arc<Mutex<Projection>>,
}

Start from the oldest record

The balances live in memory, so after a restart they are built again from the whole stream.

Macros
Manual

src/main.rs
#[subscriber(
    KinesisStream::new("ledger").create_if_missing(1),
    start_at(KinesisPosition::horizon())
)]
async fn project_balance(
    posting: &Posting,
    ctx: &mut Context<'_, KinesisContext, Projections>,
    Ctx(seeker): Ctx<SeekHandle>,
) -> HandlerOutcome {
    let applied = ctx
        .state()
        .balances
        .lock()
        .unwrap_or_else(PoisonError::into_inner)
        .apply(posting);
    if applied {
        tracing::info!(posting.id, "balance updated");
        return HandlerOutcome::ack();
    }
    tracing::info!(
        posting.id,
        posting.revision,
        "new revision, rebuilding the balances"
    );
    if seeker.seek(KinesisPosition::horizon()).await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

src/main.rs
/// Declaring `KinesisContext` is what puts the seek handle in the context.
struct ProjectBalance;

impl Handle<Posting, (), (), KinesisContext, Projections> for ProjectBalance {
    async fn handle(
        &self,
        posting: &Posting,
        _outs: &(),
        ctx: &mut Context<'_, KinesisContext, Projections>,
    ) -> Result<(), HandlerOutcome> {
        let applied = ctx
            .state()
            .balances
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .apply(posting);
        if applied {
            tracing::info!(posting.id, "balance updated");
            return Ok(());
        }
        tracing::info!(
            posting.id,
            posting.revision,
            "new revision, rebuilding the balances"
        );
        if ctx
            .context(SeekHandle)
            .seek(KinesisPosition::horizon())
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

start_at(KinesisPosition::horizon()) opens every shard at its trim horizon, the oldest record the stream still keeps. It applies on every start, to the shards the subscription opens with, ahead of any stored checkpoint. Without start_at, a shard resumes after its checkpoint and opens at its tip when it has none.

KinesisPosition has four positions:

Position	Where the subscription reads from
`horizon()`	the oldest record of every shard
`latest()`	the tip of every shard, so only new records are read
`timestamp(ms)`	the first record of every shard that arrived at or after `ms`
`sequence(shard, seq)`	one record of one shard, and the records after it

A stream keeps its records for 24 hours by default, and up to 365 days with extended retention. What is past the retention period is gone, and the trim horizon moves forward with it.

Seek from a handler

The same handler rebuilds the balances when a posting of a newer revision arrives. Ctx<SeekHandle> gives the handler the KinesisSeeker of its own subscription. seek moves the subscription to the oldest record, and the handler reads every posting again. The posting that started the rebuild comes back as well, and this time the projection applies it.

A seek takes any of the positions above. A position for the whole stream moves every shard this instance holds. Ctx<Position> holds the delivery's own position, a sequence position: a seek there reads that shard again from the record and moves no other shard. It needs this instance to hold the shard; a seek to a shard that another instance reads, or that is finished, fails.

Seek from a batch handler

Card postings arrive in volume, so the spend is built in batches of up to 500.

Macros
Manual

src/main.rs
#[subscriber(KinesisStream::new("card-postings").create_if_missing(1))]
async fn project_spend(
    postings: &[Posting],
    ctx: &mut Context<'_, KinesisBatchContext, Projections>,
) -> HandlerOutcome {
    let applied = {
        let mut spend = ctx
            .state()
            .spend
            .lock()
            .unwrap_or_else(PoisonError::into_inner);
        postings.iter().all(|posting| spend.apply(posting))
    };
    if applied {
        tracing::info!(postings = postings.len(), "card spend updated");
        return HandlerOutcome::ack();
    }
    tracing::info!("new revision, rebuilding the card spend");
    if ctx
        .context(SeekHandle)
        .seek(KinesisPosition::horizon())
        .await
        .is_err()
    {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

src/main.rs
/// A batch body declares `KinesisBatchContext`, which carries the same seek handle.
struct ProjectSpend;

impl Handle<[Posting], (), (), KinesisBatchContext, Projections> for ProjectSpend {
    async fn handle(
        &self,
        postings: &[Posting],
        _outs: &(),
        ctx: &mut Context<'_, KinesisBatchContext, Projections>,
    ) -> Result<(), Vec<HandlerOutcome>> {
        let applied = {
            let mut spend = ctx
                .state()
                .spend
                .lock()
                .unwrap_or_else(PoisonError::into_inner);
            postings.iter().all(|posting| spend.apply(posting))
        };
        if applied {
            tracing::info!(postings = postings.len(), "card spend updated");
            return Ok(());
        }
        tracing::info!("new revision, rebuilding the card spend");
        if ctx
            .context(SeekHandle)
            .seek(KinesisPosition::horizon())
            .await
            .is_err()
        {
            return Err(postings.iter().map(|_| HandlerOutcome::retry()).collect());
        }
        Ok(())
    }
}

A batch handler declares KinesisBatchContext instead of the per-delivery context. It carries the same SeekHandle key and no position, because a batch spans many records. A body that needs the place of one record reads its kinesis-shard-id and kinesis-sequence-number headers.

Start from a moment

The payout review reads the payouts of the last day on every start.

Macros
Manual

src/main.rs
/// A payout as the payments service publishes it.
#[derive(Debug, PartialEq, Deserialize)]
struct Payout {
    id: u64,
    amount_cents: i64,
}

#[subscriber(KinesisStream::new("payouts").create_if_missing(1))]
async fn review_payout(payout: &Payout) -> HandlerOutcome {
    tracing::info!(payout.id, payout.amount_cents, "payout reviewed");
    HandlerOutcome::ack()
}

src/main.rs
/// A payout as the payments service publishes it.
#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
struct Payout {
    id: u64,
    amount_cents: i64,
}

/// The body reads no context, but runs in an app whose state is `Projections`.
struct ReviewPayout;

impl Handle<Payout, (), (), (), Projections> for ReviewPayout {
    fn handle(
        &self,
        payout: &Payout,
        _outs: &(),
        _ctx: &mut Context<'_, (), Projections>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(payout.id, payout.amount_cents, "payout reviewed");
        ready(Ok(()))
    }
}

The moment is known only when the service starts, so the mount site names the position instead of the attribute.

Macros
Manual

src/main.rs
/// Milliseconds since the epoch, the unit of a Kinesis arrival timestamp.
fn epoch_millis(at: SystemTime) -> u64 {
    at.duration_since(UNIX_EPOCH).map_or(0, |since| {
        u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
    })
}

fn service(broker: KinesisBroker, review_from: SystemTime) -> impl App<State = Projections> {
    RustStream::new(AppInfo::new("ledger", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Projections::default()))
        .with_broker(broker, |b| {
            b.include(project_balance);
            b.include(project_spend.batch(nonzero!(500usize)));
            b.include(
                review_payout.start_at(KinesisPosition::timestamp(epoch_millis(review_from))),
            );
        })
}

/// The review reads the payouts of the last day on every start, the stream's default retention.
const REVIEW_WINDOW: Duration = Duration::from_hours(24);

#[ruststream::app]
fn app() -> impl App<State = Projections> {
    let review_from = SystemTime::now() - REVIEW_WINDOW;
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        review_from,
    )
}

src/main.rs
/// Milliseconds since the epoch, the unit of a Kinesis arrival timestamp.
fn epoch_millis(at: SystemTime) -> u64 {
    at.duration_since(UNIX_EPOCH).map_or(0, |since| {
        u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
    })
}

fn service(broker: KinesisBroker, review_from: SystemTime) -> impl App<State = Projections> {
    RustStream::new(AppInfo::new("ledger", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Projections::default()))
        .with_broker(broker, |b| {
            b.include(
                subscriber(
                    KinesisStream::new("ledger").create_if_missing(1),
                    ProjectBalance,
                )
                .start_at(KinesisPosition::horizon())
                .build(),
            );
            b.include(
                subscriber(
                    KinesisStream::new("card-postings").create_if_missing(1),
                    ProjectSpend,
                )
                .batch(nonzero!(500usize))
                .build(),
            );
            b.include(
                subscriber(
                    KinesisStream::new("payouts").create_if_missing(1),
                    ReviewPayout,
                )
                .start_at(KinesisPosition::timestamp(epoch_millis(review_from)))
                .build(),
            );
        })
}

/// The review reads the payouts of the last day on every start, the stream's default retention.
const REVIEW_WINDOW: Duration = Duration::from_hours(24);

fn app() -> impl App<State = Projections> {
    let review_from = SystemTime::now() - REVIEW_WINDOW;
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
        review_from,
    )
}

fn main() -> ExitCode {
    run_main(app)
}

KinesisPosition::timestamp takes milliseconds since the epoch. Kinesis matches it against the arrival timestamp it stamped on each record. That stamp is approximate, so records that arrived around the same moment are not told apart exactly.

Seeks and checkpoints

A seek clears the checkpoint bookkeeping of every shard it moves. A delivery read before the seek settles into nothing, so the ack of the posting that started the rebuild moves no checkpoint. A later checkpoint therefore never passes a record that the seek replayed and no handler has handled yet. A seek moves this instance only: other instances keep reading the shards they hold, and an instance that takes a shard over resumes from its checkpoint.

The tests in the example file run the app in process. The in-process Kinesis keeps the whole log of each stream, so a seek reads it again there too.

The read models​

Start from the oldest record​

Seek from a handler​

Seek from a batch handler​

Start from a moment​