Skip to main content

Resuming and seeking

This guide makes a service resume where its last run stopped, and moves a running subscription to another place in the file.

A stream file keeps no record of how far a reader got. ack and nack report Unsupported on this transport (Acknowledgement), so a restarted subscription opens wherever its mount site says. A service that must not handle an order twice stores its own checkpoint.

The example is an invoicing service that reads the shop's orders and invoices each one once, across restarts.

The checkpoint​

src/checkpoint.rs
/// The sequence of the last order the service invoiced, kept in a file of its own.
#[derive(Debug, Clone)]
struct Checkpoint {
    path: Arc<Path>,
}

impl Checkpoint {
    fn new(path: impl AsRef<Path>) -> Self {
        Self {
            path: path.as_ref().into(),
        }
    }

    /// The stored sequence; `None` until the first order is invoiced.
    fn load(&self) -> Option<u64> {
        fs::read_to_string(&self.path).ok()?.trim().parse().ok()
    }

    fn store(&self, sequence: u64) -> io::Result<()> {
        fs::write(&self.path, sequence.to_string())
    }
}

/// The app's state: each field reaches a handler as `State<FieldType>`.
#[derive(Debug, Clone, FromRef)]
struct Invoicing {
    checkpoint: Checkpoint,
}

The checkpoint is the sequence of the last order the service invoiced, in a file of its own. Invoicing is the app's state, and the handler takes the checkpoint from it. Any store the service already has works the same way: a database row, a key in a cache.

Resuming at the checkpoint​

src/main.rs
#[subscriber(FileStream::new("orders"))]
async fn invoice(
    order: &Order,
    Ctx(at): Ctx<Position>,
    State(checkpoint): State<Checkpoint>,
) -> HandlerOutcome {
    let FilePosition::Sequence(sequence) = at else {
        return HandlerOutcome::drop();
    };
    // The run opens at the stored sequence, so the order the last run ended on comes again.
    if checkpoint.load().is_some_and(|stored| sequence <= stored) {
        return HandlerOutcome::drop();
    }
    tracing::info!(order.id, sequence, "invoiced");
    if let Err(error) = checkpoint.store(sequence) {
        tracing::error!(order.id, sequence, %error, "checkpoint not stored");
    }
    HandlerOutcome::ack()
}

fn invoicing(file: FileBroker, checkpoint: Checkpoint) -> impl App<State = Invoicing> {
    let start = checkpoint
        .load()
        .map_or(FilePosition::beginning(), FilePosition::sequence);
    RustStream::new(AppInfo::new("invoicing", "0.1.0"))
        .on_startup(async move |()| Ok::<_, Infallible>(Invoicing { checkpoint }))
        .with_broker(file, |b| {
            b.include(invoice.start_at(start));
        })
}

fn main() -> ExitCode {
    run_main(|| {
        invoicing(
            FileBroker::new("/var/lib/shop/orders.ss"),
            Checkpoint::new("/var/lib/shop/invoicing.checkpoint"),
        )
    })
}

invoicing reads the checkpoint before the service starts and passes it to start_at at the mount site. The first run has no checkpoint and reads the file from the beginning. FilePosition::sequence(n) delivers message n itself, so a run opened at the stored sequence gets the order the last run ended on once more. invoice reads the delivery's sequence with Ctx<Position> and drops an order at or below the checkpoint. Every other order is invoiced, and its sequence becomes the new checkpoint.

The checkpoint is written after the invoice. A run that stops between the two invoices that order again on the next start, so the invoice itself must tolerate a repeat. Writing the checkpoint first gives the opposite trade: an order is never invoiced twice, and one can be lost.

FilePosition::timestamp(ms) opens at the first message written after a moment, in milliseconds since the Unix epoch. It suits a service that rebuilds its view from a point in time instead of from a stored sequence.

Moving a running subscription​

A handler moves its own subscription with Ctx<SeekHandle>. seek(FilePosition::end()) skips the backlog and goes on from the tip; seek(FilePosition::sequence(n)) goes back to message n or forward past a run of messages known to fail. The deliveries already queued from the old place are discarded, and the next delivery comes from the new one. A seek to a sequence past the next one the file writes is refused with an error: a live subscription then ends, and a replay stays where it was. A batch handler reaches the same handle through FileBatchContext.

Testing the resume​

The in-process tests record five orders before the service starts and store a checkpoint of three. The service reads orders three to five, drops the third and invoices the other two. The restart test runs the service twice on one real file with TestApp::start_live: the second run opens at the order the first one ended on and drops it. Testing a file or pipeline service covers both modes.