start_at

start_at seeks a subscription to a position on every startup, rewinding its durable cursor, unlike Pulsar's initial position.

It is a clause of the attribute, start_at(PulsarPosition::earliest()), or .start_at(position) among the mount-site settings. A Pulsar subscription is durable state on the broker: its cursor records what was acknowledged, and it outlives the service. The seek lands before the subscription's first delivery, so the handler sees the log from the position on. start_at(PulsarPosition::earliest()) therefore replays the retained backlog each time the service starts, acknowledged messages included. Pulsar's own initial position is different: the broker applies it only when it creates the subscription. A subscription without the clause resumes from its cursor. It suits a read model rebuilt from the retained log on each boot.

Replay and seeking rebuilds account balances from a ledger on every startup.

Example

The audit projection is rebuilt from the oldest retained entry on every startup, although its subscription acknowledged everything it read before.

Macros

examples/pulsar_positions_start_at.rs

/// Rebuilds the audit projection from the oldest retained entry on every startup.
#[subscriber(
    PulsarSubscription::new("ledger.audit", "audit"),
    start_at(PulsarPosition::earliest())
)]
async fn replay(entry: &AuditEntry) -> HandlerOutcome {
    tracing::info!(entry.id, "projected");
    HandlerOutcome::ack()
}

Manual

examples/manual/pulsar_positions_start_at.rs

/// Projects every audit entry it reads.
struct Replay;

impl Handle<AuditEntry> for Replay {
    fn handle(
        &self,
        entry: &AuditEntry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(entry.id, "projected");
        ready(Ok(()))
    }
}

/// The service on the Pulsar cluster `url` names.
fn service(url: &str) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(PulsarBroker::new(url), |b| {
        // The projection is rebuilt from the oldest retained entry on every startup.
        b.include(
            subscriber(PulsarSubscription::new("ledger.audit", "audit"), Replay)
                .start_at(PulsarPosition::earliest())
                .build(),
        );
    })
}