Контрольные точки и гарантии доставки

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

Kinesis не хранит состояния для каждой записи и каждого читателя. Читатель сам запоминает, докуда дошёл в каждом сегменте, и в следующий раз начинает с этого места. Это место - его контрольная точка. В RustStream контрольную точку двигает исход обработчика.

Сервис

Пример отправляет заказы через перевозчика. Отправление без адреса не уйдёт никогда. Отправление для перевозчика, которого эта версия сервиса не знает, ждёт версии, которая его знает.

Макросы

src/main.rs

/// The carriers this release of the service books with.
const CARRIERS: [&str; 2] = ["dhl", "ups"];

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "shipments")]
struct Shipment {
    id: u64,
    address: String,
    carrier: String,
}

#[subscriber(KinesisStream::new("shipments").create_if_missing(1))]
async fn ship(shipment: &Shipment) -> HandlerOutcome {
    if shipment.address.is_empty() {
        tracing::warn!(shipment.id, "shipment without an address retired");
        return HandlerOutcome::drop();
    }
    if !CARRIERS.contains(&shipment.carrier.as_str()) {
        tracing::warn!(
            shipment.id,
            shipment.carrier,
            "unknown carrier, shipment held"
        );
        return HandlerOutcome::retry();
    }
    tracing::info!(shipment.id, shipment.carrier, "shipment booked");
    HandlerOutcome::ack()
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(ship);
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

Вручную

src/main.rs

/// The carriers this release of the service books with.
const CARRIERS: [&str; 2] = ["dhl", "ups"];

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Shipment {
    id: u64,
    address: String,
    carrier: String,
}

impl OutgoingDestination for Shipment {
    type Form = FixedName;

    const DESTINATION: &'static str = "shipments";
}

impl MessageHeaders for Shipment {
    type Contract = NoHeaders;
}

struct Ship;

impl Handle<Shipment> for Ship {
    fn handle(
        &self,
        shipment: &Shipment,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if shipment.address.is_empty() {
            tracing::warn!(shipment.id, "shipment without an address retired");
            return ready(Err(HandlerOutcome::drop()));
        }
        if !CARRIERS.contains(&shipment.carrier.as_str()) {
            tracing::warn!(
                shipment.id,
                shipment.carrier,
                "unknown carrier, shipment held"
            );
            return ready(Err(HandlerOutcome::retry()));
        }
        tracing::info!(shipment.id, shipment.carrier, "shipment booked");
        ready(Ok(()))
    }
}

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("shipping", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber(KinesisStream::new("shipments").create_if_missing(1), Ship).build());
    })
}

fn main() -> ExitCode {
    run_main(|| {
        service(
            KinesisBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
        )
    })
}

Что делает каждый исход

Исход	Запись	Контрольная точка сегмента
ack	обработана	проходит за неё, когда обработаны все более ранние записи сегмента
drop	снята, больше не читается	как при ack
retry без предела в точке монтирования	остаётся необработанной	останавливается перед ней

Контрольная точка покрывает все записи перед ней, поэтому продвигается только по непрерывному ряду обработанных записей. Крейт записывает её в хранилище аренд, когда она сдвигается. Запись, обработанная не по порядку, в параллельной партиции воркеров или в пакете, ничего не сдвигает, пока не обработаны и записи перед ней.

ship снимает отправление без адреса через drop: новая попытка получит тот же ответ.

retry просит Kinesis доставить запись снова, а сегмент не может сделать это для одной записи: это журнал, который читается по порядку. Доставка отвечает AckError::Unsupported, среда выполнения пишет об этом предупреждение в журнал, и запись остаётся необработанной. Так ship задерживает отправление для неизвестного перевозчика. Читатель продолжает: следующие отправления доставляются и обрабатываются, а контрольная точка ждёт перед задержанным. Ждёт только его сегмент. Контрольные точки остальных сегментов продвигаются дальше. При пределе, объявленном в точке монтирования, retry вместо этого публикует копию (Повторы и потоки мёртвых писем).

Что читается заново

Гарантия доставки - at-least-once. Читатель, который берёт сегмент, начинает после его контрольной точки, поэтому всё, начиная с первой необработанной записи, доставляется заново: задержанное отправление и все отправления за ним, обработанные или нет. Так происходит, когда сервис перезапускается и когда сегмент забирает другой экземпляр (Несколько экземпляров). Поэтому обработчик, который может увидеть запись дважды, пишется идемпотентным: отправление, проведённое дважды, учитывается один раз.

Контрольная точка записывается в хранилище аренд, и от хранилища зависит, что переживёт перезапуск. MemoryLeaseStore по умолчанию хранится в процессе и стартует пустым, поэтому перезапущенный сервис открывает каждый сегмент с конца и читает только то, что опубликовано с этого момента. DynamoLeaseStore хранит контрольные точки в таблице, и перезапущенный сервис продолжает с того места, где остановился предыдущий. Сегмент без контрольной точки открывается там, куда указывает start_at, а если ничего не указано - с конца (Повторное чтение и перемотка).

Задержанная запись не возвращается, пока сегмент держит тот же читатель. Чтобы прочитать её снова без перезапуска, обработчик перематывает подписку назад к ней.

Пакеты и закрытые сегменты

Пакетный обработчик, который возвращает один исход, завершает им каждую запись пакета. Обработчик, который возвращает Vec<HandlerOutcome>, завершает каждую запись отдельно, и каждая двигает контрольную точку своего сегмента.

Сегмент, который закрыло разделение или слияние, читается до конца. Его контрольная точка становится SHARD_END, только когда завершена каждая доставка из него, и лишь тогда запускаются его дочерние сегменты. Поэтому задержанная запись в закрытом сегменте задерживает и его дочерние сегменты.

Проверка исходов

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