PulsarPosition

PulsarPosition names a place in a topic's retained log: either end, a delivered message, or a publish time.

Pulsar keeps a topic's messages as a log, so a subscription can be moved through it, at startup with start_at or from a handler with a seek. PulsarPosition::earliest() is the oldest message the topic still retains, and every retained message is delivered again from there. PulsarPosition::latest() is the tip of the log: only messages published after the move are delivered. PulsarPosition::timestamp(millis) is the first message published at or after that time, in milliseconds since the Unix epoch. A delivery's own place is a position as well: Ctx<Position> binds it through the Position key, a copy the handler can keep. A seek to such a position delivers that message again, then the rest of the log in order. A position applies to every topic and every partition the subscription reads. A non-persistent topic retains nothing, so a move over it replays nothing.

Replay and seeking rebuilds balances from a ledger on every startup and indexes every order again with a seek.

Example

A projection bookmarks the entry that closes a day; a rebuild request moves the ledger subscription back to it.

Macros
Manual

examples/pulsar_positions.rs
/// Where the projection last closed a day: the ledger subscription's seeker and the position
/// of the entry that closed it.
#[derive(Clone, Default)]
struct Bookmark(Arc<Mutex<Option<(PulsarSeeker, PulsarPosition)>>>);

#[derive(FromRef)]
struct AppState {
    bookmark: Bookmark,
}

/// Projects each ledger entry and bookmarks the entry that closes a day.
#[subscriber(PulsarSubscription::new("ledger", "projection"))]
async fn project(
    entry: &Entry,
    Ctx(position): Ctx<Position>,
    Ctx(seeker): Ctx<SeekHandle>,
    State(bookmark): State<Bookmark>,
) -> HandlerOutcome {
    tracing::info!(entry.id, "projected");
    if entry.closes_day {
        *bookmark.0.lock().await = Some((seeker, position));
    }
    HandlerOutcome::ack()
}

/// Replays the ledger from the last closed day: the entry at the bookmark comes again, then
/// every entry after it, in order.
#[subscriber(PulsarSubscription::new("ledger.rebuild", "projection"))]
async fn rebuild(request: &Rebuild, State(bookmark): State<Bookmark>) -> HandlerOutcome {
    let Some((seeker, position)) = bookmark.0.lock().await.clone() else {
        return HandlerOutcome::ack();
    };
    tracing::info!(reason = %request.reason, "rebuilding from the last closed day");
    match seeker.seek(position).await {
        Ok(()) => HandlerOutcome::ack(),
        Err(err) => {
            tracing::warn!(%err, "the ledger cannot be rewound");
            HandlerOutcome::retry()
        }
    }
}

examples/manual/pulsar_positions.rs
/// Where the projection last closed a day: the ledger subscription's seeker and the position
/// of the entry that closed it.
#[derive(Clone, Default)]
struct Bookmark(Arc<Mutex<Option<(PulsarSeeker, PulsarPosition)>>>);

struct AppState {
    bookmark: Bookmark,
}

/// Projects each ledger entry and bookmarks the entry that closes a day.
struct Project;

impl Handle<Entry, (), (), PulsarContext, AppState> for Project {
    async fn handle(
        &self,
        entry: &Entry,
        _outs: &(),
        ctx: &mut Context<'_, PulsarContext, AppState>,
    ) -> Result<(), HandlerOutcome> {
        tracing::info!(entry.id, "projected");
        if entry.closes_day {
            let mark = (
                ctx.context(SeekHandle).clone(),
                ctx.context(Position).clone(),
            );
            *ctx.state().bookmark.0.lock().await = Some(mark);
        }
        Ok(())
    }
}

/// Replays the ledger from the last closed day: the entry at the bookmark comes again, then
/// every entry after it, in order.
struct Rewind;

impl Handle<Rebuild, (), (), PulsarContext, AppState> for Rewind {
    async fn handle(
        &self,
        request: &Rebuild,
        _outs: &(),
        ctx: &mut Context<'_, PulsarContext, AppState>,
    ) -> Result<(), HandlerOutcome> {
        let Some((seeker, position)) = ctx.state().bookmark.0.lock().await.clone() else {
            return Ok(());
        };
        tracing::info!(reason = %request.reason, "rebuilding from the last closed day");
        seeker.seek(position).await.map_err(|err| {
            tracing::warn!(%err, "the ledger cannot be rewound");
            HandlerOutcome::retry()
        })
    }
}

fn app() -> impl App<State = AppState> {
    RustStream::new(AppInfo::new("ledger", "0.1.0"))
        .on_startup(async move |()| {
            Ok::<_, Infallible>(AppState {
                bookmark: Bookmark::default(),
            })
        })
        .with_broker(PulsarBroker::new("pulsar://localhost:6650"), |b| {
            b.include(subscriber(PulsarSubscription::new("ledger", "projection"), Project).build());
            b.include(
                subscriber(
                    PulsarSubscription::new("ledger.rebuild", "projection"),
                    Rewind,
                )
                .build(),
            );
        })
}

Example​

Example