Skip to main content

An event log on disk

This guide records a shop's events in a stream file, follows them as they are written and totals a finished file in one pass.

A stream file is a log: a publish appends a message under a stream key, and every reader of the key sees it in order. The example is a shop that records a receipt for each order, and an audit job that totals the receipts of a finished file.

Recording and following​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "orders")]
struct Order {
    id: u64,
    amount_cents: u64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "receipts")]
struct Receipt {
    order: u64,
    amount_cents: u64,
    /// Where the order sits under `orders`, so a reader can move back to it.
    sequence: u64,
}

/// Records a receipt for each order, appended to the same file under `receipts`.
#[subscriber(FileStream::new("orders"), reply)]
async fn checkout(order: &Order, Ctx(at): Ctx<Position>) -> Receipt {
    let sequence = match at {
        FilePosition::Sequence(sequence) => sequence,
        // A delivery always sits at a sequence; the other forms name places to start from.
        FilePosition::Beginning | FilePosition::End | FilePosition::Timestamp(_) => 0,
    };
    Receipt {
        order: order.id,
        amount_cents: order.amount_cents,
        sequence,
    }
}

/// Follows the receipts as the shop writes them.
#[subscriber(FileStream::new("receipts"))]
async fn notify(receipt: &Receipt) -> HandlerOutcome {
    tracing::info!(receipt.order, "customer notified");
    HandlerOutcome::ack()
}

fn shop(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("shop", "0.1.0")).with_broker(file, |b| {
        b.include(checkout);
        b.include(notify);
    })
}

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

checkout answers each order with a Receipt, and the reply is appended to the same file under receipts, the key the type declares. One file holds every stream key, so the path is named once, in FileBroker::new. A publish appends the message and flushes it, so the bytes reach the operating system before the call returns. The file is synced to the device when the broker shuts down.

notify follows receipts from the end of the file and reads each receipt as checkout appends it. Another process that opens the same path follows the key the same way. A subscription with no position starts at the end, so a restarted notify reads only the receipts written after it started.

Each stream key is numbered from one. The receipt keeps the order's sequence, which Ctx<Position> hands the handler, so a reader can move back to that order later. A later run that appends to the same file carries the numbering on: after a restart the next order is number three, not one.

The file client caps a stream file at 16 GiB. A publish that reaches the cap fails, and so does every later publish to that file. Running in production covers what that means for a long-lived file.

A one-pass job​

src/bin/audit.rs
/// Totals the receipts of a finished file, up to 64 at a time; one outcome settles the batch.
#[subscriber(FileStream::new("receipts").replay())]
async fn total(batch: &[Receipt]) -> HandlerOutcome {
    let cents: u64 = batch.iter().map(|receipt| receipt.amount_cents).sum();
    tracing::info!(receipts = batch.len(), cents, "receipts totalled");
    HandlerOutcome::ack()
}

/// The audit job: it reads the shop's file and never creates one.
fn audit(file: FileBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("audit", "0.1.0")).with_broker(file.existing_only(), |b| {
        b.include(total.batch(nonzero!(64)));
    })
}

replay() reads what the file holds under receipts from the first message and completes at the end of the file. A receipt appended after that point is not read, so the job does its pass and has nothing more to wait for. existing_only makes startup fail when the file is missing, so a mistyped path stops the job instead of creating an empty file.

total takes a slice, and batch(nonzero!(64)) at the mount site caps a batch at 64 receipts. A batch takes what the file already holds and does not wait to fill up; one outcome settles the whole batch.

A writer marks a file finished with end_with_eos, which appends an end-of-stream mark at shutdown. A replay completes at the mark, and a live subscription such as notify ends on it.

Testing the log​

The tests in the example file run the shop and the audit in process: an order gets its receipt and notify reads it, and the audit reads three recorded receipts in one batch and ignores a fourth. The audit's receipts are recorded on a clone of its broker before it starts, the way the shop would have left them. The restart test runs the shop twice on one real file with TestApp::start_live and checks that the second run numbers its order three. Testing a file or pipeline service covers both modes.