replay

replay reads the whole retained file once and completes at its end or at the end-of-stream mark, instead of following writes.

FileStream::new("orders").replay() turns the subscription into one pass over what the file holds under its key. It starts at the first message the file retains and delivers every message in order. It completes at the end of the file, or at the end-of-stream mark a writer left with end_with_eos, and a message appended after that point is not read. That suits a job over a finished log: an audit, a migration, an index rebuilt from scratch. start_at(FilePosition::beginning()) on a plain descriptor reads the same messages and then goes on following new writes. A seek from a handler still moves a replay, one that has reached the end included, and the replay goes on from the new position (Seek). A seek to a position past the tip is refused, and the replay stays where it was. A retry_after copy reaches a replay while the replay is still short of the end, and is left unread once the replay has passed it. A service that only reads a finished file opens it with existing_only, so a wrong path stops it at startup instead of creating an empty file. The guide An event log on disk totals a finished file with a replay.

Example

The audit reads the three orders of a finished day and completes; an order appended later is not read.

Macros
Manual

examples/file_replay.rs
/// Audits a finished day of orders in one pass.
#[subscriber(FileStream::new("orders").replay())]
async fn audit(order: &Order) -> HandlerOutcome {
    tracing::info!(order.id, "audited");
    HandlerOutcome::ack()
}

/// The audit of the day's file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(file, |b| {
        b.include(audit);
    })
}

fn main() -> ExitCode {
    // The shop wrote the day's file; the audit reads it and never creates one.
    run_main(|| app(FileBroker::new("/var/lib/shop/2026-10-17.ss").existing_only()))
}

examples/manual/file_replay.rs
/// Audits a finished day of orders in one pass.
struct Audit;

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

/// The audit of the day's file `main` hands it.
fn app(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(file, |b| {
        b.include(subscriber(FileStream::new("orders").replay(), Audit).build());
    })
}

fn main() -> ExitCode {
    // The shop wrote the day's file; the audit reads it and never creates one.
    run_main(|| app(FileBroker::new("/var/lib/shop/2026-10-17.ss").existing_only()))
}

Example​

Example