Replay and seeking

This guide reads a Pulsar topic's retained log again: from the start on every startup, or from a position a handler seeks to.

A subscription is a cursor on the topic's log, and the server keeps it. A seek moves that cursor, and the subscription reads on from its new place.

The example is a search service. It keeps account balances built from a ledger, and a search index of orders. Search stands in for the client of the search cluster.

Positions

PulsarPosition names a place in a topic's log:

Position	Where
`PulsarPosition::earliest()`	the oldest retained message
`PulsarPosition::latest()`	the end of the log: only messages published after the seek
`PulsarPosition::timestamp(millis)`	the first message published at or after that time, in milliseconds since the Unix epoch
`PulsarPosition::MessageId(..)`	a delivered message, read from its `Position`

Every startup from the start

Macros
Manual

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 seeks the subscription each time the service starts. balances reads the ledger from its oldest retained entry on every start, so the balances are rebuilt from scratch. Without start_at, an existing subscription resumes where its consumers stopped, and a new one starts at the end of the topic.

Pulsar has an initial position of its own, but it applies only when a subscription is created. start_at applies to the subscription that already exists too. Every instance that starts moves the shared cursor, so the instances of a service with start_at replay the log at each start of any of them.

A seek from the handler

Macros
Manual

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 is the subscription's own seeker, and index reads it from the Pulsar context. When the search cluster loses its index, index seeks to the oldest retained order, and every order is indexed again. In the manual form the handler names PulsarContext as its context and reads the seeker with ctx.context(SeekHandle).

A seek moves the whole subscription: every consumer of it, every topic it reads and every partition. Deliveries already queued for the subscription are discarded, and the log is read from the new place.

Two other seeks come up often:

PulsarPosition::timestamp(millis) skips a stretch of bad data: the subscription resumes at the first message published after it.
A delivery's position, read with Ctx<Position>, delivers exactly that message again, then the rest of the log in order.

A batch handler seeks the same way through PulsarBatchContext, which carries the seeker and no position, because a batch spans many messages.

What a replay reaches

A replay reads what the topic still holds. By default Pulsar deletes a message once every subscription has acknowledged it. A retention policy on the namespace keeps acknowledged messages for a size or a time:

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

A non-persistent topic keeps nothing, so a seek there replays nothing.

The app

Macros
Manual

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 builds the app on the broker it is given, so a test can take a publisher off that broker before the app starts.

Testing a seek

The tests in the example file run the app in process. The in-process Pulsar keeps each topic's log, so a seek there replays it as a server does. The test publishes three orders in one burst through broker.publisher(), so all three wait in the subscription before the first one is handled. tb.settle() then waits until the subscription is idle. A backlog left by an earlier run of the service exists only on a server, so start_at over such a backlog is checked by the crate's live tests.

Positions​

Every startup from the start​

A seek from the handler​

What a replay reaches​

The app​