Повторное чтение и перемотка

Это руководство читает поток заново с самой старой записи или с заданного момента и перематывает работающую подписку из её обработчика.

Kinesis хранит запись и после того, как потребитель её прочитал, пока не истечёт срок хранения потока. Сервис может прочитать её снова: чтобы перестроить модель чтения, заполнить данными нового потребителя или пропустить записи, которые ему не годятся. Подписка получает начальную позицию в точке монтирования, а обработчик перематывает свою подписку через контекст доставки.

Модели чтения

Сервис держит в памяти две модели чтения: баланс каждого счёта и расходы по каждой карте. Каждая проводка содержит ревизию реестра. Исправление прошлых проводок публикуется новой ревизией, и модель чтения, которая её встречает, перестраивается с самой старой записи потока.

Макросы

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>>,
}

Чтение с самой старой записи

Балансы хранятся в памяти, поэтому после перезапуска они строятся заново по всему потоку.

Макросы

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()) открывает каждый сегмент с TRIM_HORIZON, самой старой записи, которую поток ещё хранит. Эта позиция действует при каждом запуске, на сегменты, с которыми открывается подписка, и имеет приоритет над любой сохранённой контрольной точкой. Без start_at сегмент продолжает после своей контрольной точки, а если её нет, открывается с конца.

У KinesisPosition четыре позиции:

Позиция	Откуда читает подписка
horizon()	с самой старой записи каждого сегмента
latest()	с конца каждого сегмента, поэтому читаются только новые записи
timestamp(ms)	с первой записи каждого сегмента, поступившей в момент ms или позже
sequence(shard, seq)	с одной записи одного сегмента и записей после неё

По умолчанию поток хранит записи 24 часа, а с продлённым сроком хранения - до 365 дней. Записи старше срока хранения удаляются, и TRIM_HORIZON сдвигается вперёд вслед за ними.

Перемотка из обработчика

Тот же обработчик перестраивает балансы, когда приходит проводка более новой ревизии. Ctx<SeekHandle> даёт обработчику KinesisSeeker его собственной подписки. seek перематывает подписку на самую старую запись, и обработчик читает все проводки заново. Проводка, с которой началась перестройка, тоже возвращается, и на этот раз проекция её применяет.

Перемотка принимает любую из позиций выше. Позиция для всего потока перематывает каждый сегмент, который держит этот экземпляр. Ctx<Position> хранит собственную позицию доставки, позицию sequence: перемотка на неё читает этот сегмент заново с этой записи и не трогает остальные сегменты. Для неё этот экземпляр должен держать сегмент. Перемотка на сегмент, который читает другой экземпляр или который уже прочитан до конца, возвращает ошибку.

Перемотка из пакетного обработчика

Проводки по картам идут большим объёмом, поэтому расходы строятся пакетами до 500 записей.

Макросы

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(())
    }
}

Пакетный обработчик объявляет KinesisBatchContext вместо контекста одной доставки. В нём тот же ключ SeekHandle и нет позиции, потому что пакет охватывает много записей. Тело, которому нужно место одной записи, читает её заголовки kinesis-shard-id и kinesis-sequence-number.

Чтение с заданного момента

Проверка выплат при каждом запуске читает выплаты за последние сутки.

Макросы

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(()))
    }
}

Момент известен только при запуске сервиса, поэтому позицию задаёт точка монтирования, а не атрибут.

Макросы

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 принимает миллисекунды от начала эпохи. Kinesis сравнивает это значение с меткой времени поступления, которую он ставит на каждую запись. Эта метка приблизительная, поэтому записи, поступившие примерно в один момент, точно не различаются.

Перемотка и контрольные точки

Перемотка сбрасывает учёт контрольных точек каждого сегмента, который она переносит. Доставка, прочитанная до перемотки, завершается впустую, поэтому ack проводки, с которой началась перестройка, не двигает контрольную точку. Так следующая контрольная точка никогда не обгоняет запись, которую перемотка прочитала заново и которую ещё не обработал ни один обработчик. Перемотка двигает только этот экземпляр. Остальные экземпляры продолжают читать свои сегменты, а экземпляр, который забирает сегмент, продолжает с его контрольной точки.

Тесты в файле примера запускают приложение внутри процесса. Kinesis внутри процесса хранит весь журнал каждого потока, поэтому перемотка и там читает его заново.