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

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

Подписка - это курсор в журнале топика, и хранит его сервер. Перемотка переставляет этот курсор, и подписка продолжает чтение с нового места.

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

Позиции

PulsarPosition называет место в журнале топика:

Позиция	Где
PulsarPosition::earliest()	самое старое хранимое сообщение
PulsarPosition::latest()	конец журнала: только сообщения, опубликованные после перемотки
PulsarPosition::timestamp(millis)	первое сообщение, опубликованное в этот момент или позже, в миллисекундах от начала эпохи Unix
PulsarPosition::MessageId(..)	доставленное сообщение, его позицию даёт Position

С начала при каждом запуске

Макросы

src/main.rs

/// Rebuilds the account balances from the oldest retained entry on every startup.
#[subscriber(
    PulsarSubscription::new("ledger", "balances"),
    start_at(PulsarPosition::earliest())
)]
async fn project(entry: &Entry) -> HandlerOutcome {
    tracing::info!(entry.account, entry.amount_cents, "balance updated");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

/// Rebuilds the account balances; the mount site starts it at the oldest retained entry.
struct Project;

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

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

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

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

Макросы

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
}

/// Indexes each order; a lost index sends the subscription back to the oldest retained order.
#[subscriber(PulsarSubscription::new("orders", "search-index"))]
async fn index(
    order: &Order,
    Ctx(seeker): Ctx<SeekHandle>,
    State(search): State<Arc<Search>>,
) -> HandlerOutcome {
    match search.index(order).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(SearchError::IndexLost) => {
            tracing::warn!(order.id, "index lost, indexing every order again");
            if seeker.seek(PulsarPosition::earliest()).await.is_err() {
                return HandlerOutcome::retry();
            }
            HandlerOutcome::ack()
        }
    }
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Serialize, Deserialize, Outgoing, JsonSchema)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
}

/// Indexes each order; a lost index sends the subscription back to the oldest retained order.
struct Index;

impl Handle<Order, (), (), PulsarContext, Clients> for Index {
    async fn handle(
        &self,
        order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, PulsarContext, Clients>,
    ) -> Result<(), HandlerOutcome> {
        match ctx.state().search.index(order).await {
            Ok(()) => Ok(()),
            Err(SearchError::IndexLost) => {
                tracing::warn!(order.id, "index lost, indexing every order again");
                ctx.context(SeekHandle)
                    .seek(PulsarPosition::earliest())
                    .await
                    .map_err(|_| HandlerOutcome::retry())
            }
        }
    }
}

SeekHandle - ручка перемотки самой подписки, и index получает её из контекста Pulsar. Когда поисковый кластер теряет индекс, index перематывает подписку на самый старый хранимый заказ, и каждый заказ индексируется заново. В ручной форме обработчик называет своим контекстом PulsarContext и получает ручку перемотки через ctx.context(SeekHandle).

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

Часто нужны ещё две перемотки:

• PulsarPosition::timestamp(millis) пропускает участок плохих данных: подписка продолжает с первого сообщения, опубликованного после него.
• Позиция доставки, полученная через Ctx<Position>, доставляет ровно это сообщение снова, а за ним остаток журнала по порядку.

Пакетный обработчик перематывает так же, через PulsarBatchContext. В нём есть ручка перемотки и нет позиции, потому что пакет охватывает много сообщений.

Что доступно для повторного чтения

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

docker compose exec pulsar bin/pulsar-admin namespaces set-retention public/default \
  --size 1G --time 3d

Нехранимый топик ничего не хранит, поэтому перемотка в нём ничего не перечитывает.

Приложение

Макросы

src/main.rs

fn service(broker: PulsarBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("search", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            b.include(project);
            b.include(index);
        })
}

#[ruststream::app]
fn app() -> impl App<State = Clients> {
    service(PulsarBroker::new("pulsar://localhost:6650"))
}

Вручную

src/main.rs

fn service(broker: PulsarBroker) -> impl App<State = Clients> {
    RustStream::new(AppInfo::new("search", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Clients::default()))
        .with_broker(broker, |b| {
            b.include(
                subscriber(PulsarSubscription::new("ledger", "balances"), Project)
                    .start_at(PulsarPosition::earliest())
                    .build(),
            );
            b.include(subscriber(PulsarSubscription::new("orders", "search-index"), Index).build());
        })
}

fn app() -> impl App<State = Clients> {
    service(PulsarBroker::new("pulsar://localhost:6650"))
}

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

service строит приложение на переданном ему брокере, поэтому тест может взять у этого брокера издателя до запуска приложения.

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

Тесты в файле примера запускают приложение внутри процесса. Pulsar внутри процесса хранит журнал каждого топика, поэтому перемотка там перечитывает его, как на сервере. Тест публикует три заказа одной серией через broker.publisher(), поэтому все три ждут в подписке, прежде чем обработается первый. Затем tb.settle() ждёт, пока у подписки не кончится работа. Очередь, которую оставил прошлый запуск сервиса, существует только на сервере, поэтому start_at над такой очередью проверяют живые тесты крейта.