start_at

start_at opens the subscription at a position you name on every startup, ahead of the stored checkpoints.

It is a clause of the attribute, start_at(KinesisPosition::horizon()), or .start_at(position) among the mount-site settings. The position lands on the shards the subscription opens with, ahead of their first read, whatever their checkpoints say. A reader that starts later in the run, after a failure or a takeover, resumes from its shard's checkpoint. A child shard of a split or a merge with no checkpoint of its own opens at its oldest record, so no record of a resharded key is skipped. A sequence position on a shard this instance does not read is refused, and the service stops at startup. It suits a read model rebuilt from the retained stream on each boot.

Example

The audit projection is rebuilt from the oldest retained entry on every startup, although it checkpointed every entry it read before.

Macros

examples/kinesis_positions_start_at.rs

/// Rebuilds the audit projection from the oldest retained entry on every startup.
#[subscriber(
    KinesisStream::new("ledger.audit"),
    start_at(KinesisPosition::horizon())
)]
async fn project(entry: &AuditEntry) -> HandlerOutcome {
    tracing::info!(entry.id, "entry projected");
    HandlerOutcome::ack()
}

Manual

examples/manual/kinesis_positions_start_at.rs

/// Projects every audit entry it reads.
struct Project;

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

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(broker, |b| {
        // The projection is rebuilt from the oldest retained entry on every startup.
        b.include(
            subscriber(KinesisStream::new("ledger.audit"), Project)
                .build()
                .start_at(KinesisPosition::horizon()),
        );
    })
}