Seek

A handler moves its running subscription through the SeekHandle key; a sequence position moves one shard, the others move every shard.

Ctx<SeekHandle> binds a KinesisSeeker for the subscription the delivery came from. Its seek(position) takes a KinesisPosition and returns once the shards it moves read from there. The horizon, the tip and a timestamp move every shard this instance reads. A sequence position moves only the shard it names, and that shard needs a live reader in this instance: a shard another instance reads, or one already finished, is refused with KinesisError::Read. Ctx<Position> gives the delivery's own sequence position to aim with.

A seek moves this instance and no other. It writes no checkpoint: it starts the checkpoint bookkeeping of the shards it moves afresh, so an acknowledgement of a record delivered before the seek no longer moves the checkpoint. The next acknowledgement after the seek checkpoints from the new position. Records from that position onward are delivered again, which at-least-once delivery allows. A seek after the service shut down returns KinesisError::NotConnected. A batch handler reaches the same key through its KinesisBatchContext. A position that has to apply on every startup belongs in start_at instead.

Example

The handler replays the retained backlog, and the marker job moves the subscription to the tip, so the rest of the backlog is skipped.

Macros

examples/kinesis_seek.rs

/// Replays the retained backlog; on the marker job it abandons the rest of it and follows the
/// tip instead.
#[subscriber(KinesisStream::new("jobs"), start_at(KinesisPosition::horizon()))]
async fn replay(job: &Job, Ctx(seeker): Ctx<SeekHandle>) -> HandlerOutcome {
    if job.id == MARKER && seeker.seek(KinesisPosition::latest()).await.is_err() {
        return HandlerOutcome::retry();
    }
    tracing::info!(job.id, "job replayed");
    HandlerOutcome::ack()
}

Manual

examples/manual/kinesis_seek.rs

/// Replays the retained backlog; on the marker job it abandons the rest of it and follows the
/// tip instead.
struct Replay;

impl Handle<Job, (), (), KinesisContext> for Replay {
    async fn handle(
        &self,
        job: &Job,
        _outs: &(),
        ctx: &mut Context<'_, KinesisContext>,
    ) -> Result<(), HandlerOutcome> {
        if job.id == MARKER {
            ctx.context(SeekHandle)
                .seek(KinesisPosition::latest())
                .await
                .map_err(|_| HandlerOutcome::retry())?;
        }
        tracing::info!(job.id, "job replayed");
        Ok(())
    }
}