Rust

Install

One crate, snappea, with features. Production code needs only the emitter, which compiles no checker and depends only on serde and serde_json:

[dependencies]
snappea = { version = "0.1", default-features = false, features = ["emit"] }

[dev-dependencies]
snappea = "0.1"   # default features: test and cli

Quick start

The examples on every language page check one spec: an inventory where reservations never exceed the stock and releases match reservations.

// The inventory contract, checked against the P model in this folder and against every example application.
event eStocked: (units: int);
event eReserved: (order: string, units: int);
event eReleased: (order: string, units: int);

// Applications give each product its own monitor through emit's key; the P model has one product.
spec InventoryCorrect observes eStocked, eReserved, eReleased {
  var available: int;
  var reservations: map[string, int];

  start state Unstocked {
    on eStocked do (p: (units: int)) {
      assert p.units >= 0, "negative initial stock";
      available = p.units;
      goto Ready;
    }
    on eReserved do { assert false, "reservation before stock setup"; }
    on eReleased do { assert false, "release before stock setup"; }
  }

  state Ready {
    on eStocked do { assert false, "product stocked twice"; }
    on eReserved do (p: (order: string, units: int)) {
      assert p.units > 0, "reservation must be positive";
      assert !(p.order in reservations), "duplicate reservation";
      available = available - p.units;
      assert available >= 0, "reservation exceeds available stock";
      reservations[p.order] = p.units;
    }
    on eReleased do (p: (order: string, units: int)) {
      assert p.order in reservations, "release without a reservation";
      assert p.units == reservations[p.order], "released quantity differs from reservation";
      available = available + p.units;
      reservations -= p.order;
    }
  }
}

The service emits an event after each change:

use snappea::emit::{emit, json};
use std::collections::HashMap;

#[derive(Default)]
pub struct Inventory {
    stock: HashMap<String, i64>,
    reservations: HashMap<(String, String), i64>,
}

impl Inventory {
    pub fn add_product(&mut self, sku: &str, units: i64) {
        self.stock.insert(sku.into(), units);
        emit("eStocked", sku, json!({"units": units}));
    }

    pub fn reserve(&mut self, sku: &str, order: &str, units: i64) -> Result<(), String> {
        let available = self.stock.get_mut(sku).ok_or("unknown product")?;
        if units > *available {
            return Err("insufficient stock".into());
        }
        *available -= units;
        self.reservations.insert((sku.into(), order.into()), units);
        emit("eReserved", sku, json!({"order": order, "units": units})); // after the change
        Ok(())
    }
}

Point the crate at the specs, then wrap a test body in check:

[package.metadata.snappea]
spec_files = ["specs"]

use inventory::Inventory;
use snappea::test::Model;
use std::sync::LazyLock;

// Compiled once, from [package.metadata.snappea] in Cargo.toml.
static MODEL: LazyLock<Model> = LazyLock::new(|| Model::load().unwrap());

#[test]
fn two_orders_for_the_last_units() {
    MODEL.check(|_recording| {
        let mut inventory = Inventory::default();
        inventory.add_product("coffee", 3);
        inventory.reserve("coffee", "order-1", 2).unwrap();
        let _ = inventory.reserve("coffee", "order-2", 2); // rejecting order-2 is fine too
    });
}

cargo test passes. Now break the service: delete the stock check from reserve. The test's own code still passes, because it accepts either outcome for the second order. The spec does not:

---- two_orders_for_the_last_units stdout ----

thread 'two_orders_for_the_last_units' panicked at tests/inventory.rs:10:11:
P spec InventoryCorrect was broken: reservation exceeds available stock

  Events (key "coffee"):
      1  eStocked   units=3                src/lib.rs:13
      2  eReserved  order=order-1 units=2  src/lib.rs:20
      3  eReserved  order=order-2 units=2  src/lib.rs:20  <- broke the spec here

  Spec: specs/inventory.p:27

Test setup

A checked fixture

model.check(|recording| ...) records while the closure runs, then checks and panics with the failure output. Cargo has no global per-test hook, so each test opts in by calling it; put it in your existing fixture factory.

• Put setup and cleanup inside the closure, and join worker threads there. Nothing is checked in Drop.
• A panic in the closure propagates unchanged, with no second check.
• Never share a recording between tests: Cargo runs tests in parallel.
• Compile the Model once (in a LazyLock) and share it; clones are cheap.
• SNAPPEA_MODE=off runs the closures without checking.

Model::load() reads [package.metadata.snappea]. Model::from_files(paths) and Model::compile(&[(name, text)]) (for include_str!) set up explicitly. Builders override any setting: .specs([..]), .exclude([..]), .strict(..), .check_hot(..), .unknown_events(..), .coverage(path).

Explicit checks

The explicit form returns the result instead of panicking:

let recording = MODEL.recording();
let mut app = App::new(recording.sink());
app.reserve("coffee", "o1", 2)?;
recording.check()?; // Result<CheckReport, CheckError>

check consumes the recording; join worker threads and finish cleanup first.

Negative controls

Prove the spec catches the bug it is meant to catch:

let violation = recording.expect_violation("InventoryCorrect", "exceeds available stock");
assert_eq!(violation.key, "coffee");

It panics unless exactly that spec is broken with a message matching the regular expression.

Property tests

Create the recording inside the property body, which runs once per case and shrink attempt, and report with prop_assert! so proptest shrinks the input:

proptest! {
    #[test]
    fn ops(commands in commands()) {
        let recording = MODEL.recording();
        // ... run commands against a fresh application ...
        let checked = recording.check();
        prop_assert!(checked.is_ok(), "{}", checked.unwrap_err());
    }
}

Emitting events

use snappea::emit::{emit, emit_at, json};

emit("eReserved", "coffee", json!({"order": "o1", "units": 2}));
emit_at(row.commit_time, "eReserved", sku, json!({"order": order, "units": units})); // when it happened

• Payload: anything Serialize: json!, a #[derive(Serialize)] struct, a scalar, or () for none. Named tuples are objects; enums go by name.
• Key: anything Display. It splits events into independent streams, such as one per product; each stream gets its own instance of every spec.
• Time: emit_at takes a SystemTime, or i64 epoch nanoseconds for a logical clock. Each key's events are checked in time order.
• Emit after the change commits, or specs see states that never existed.
• The call site is recorded with #[track_caller] and shown in failure reports.

Three ways to connect application code to a recording:

• The global snappea::emit::emit: a recording binds the thread that creates it. Bind threads the test spawns with let _guard = recording.sink().bind();. With several recordings open and an unbound thread, the emit fails them all, so bind or inject.
• recording.sink(): a cloneable EventSink to inject, as in the explicit check above. Its emit returns an error after the check.
• recording.emitter(): a snappea::emit::Emitter, for code that takes one and gets a production Emitter otherwise.

In production

use snappea::emit::Emitter;

snappea::emit::set_sink(Emitter::file("/var/log/app/events/inventory.jsonl")?);
// or Emitter::stdout(), Emitter::writer(w), Emitter::callback(|event| ...)
// ... at shutdown:
snappea::emit::close(); // writes what is queued

snappea observe specs/inventory.p '/var/log/app/events/*.jsonl' --spec InventoryCorrect

Emitting never panics, returns an error, or blocks on I/O. Writers queue up to 16,384 lines and drop events when the queue is full or a write fails; snappea::emit::dropped() counts them. With no sink, emit costs a few atomic loads, but the payload is still built; code holding an Emitter can skip a costly one with is_on().

Other processes

let recording = MODEL.recording();
Command::new(server_bin).envs(recording.env()).status()?;   // sets SNAPPEA_EVENTS_DIR
recording.add_events_dir("/shared/events");                  // a folder the test did not create
recording.add_log("logs/events.jsonl");                      // JSON event lines
recording.add_log_with("logs/app.log", |line| parse(line));  // -> Option<Value>
recording.check()?;

A child process that calls snappea::emit::emit writes to recording.events_dir() through SNAPPEA_EVENTS_DIR; anything else can write the same JSON lines there. Stop or join children before the check. A failed check keeps the folder and names it.

Failures

A failure shows the spec's assert message, the events of the key that broke it (the last 20) with where each was emitted, and the assert in the P file. Model::check panics with it; check() returns CheckError::Violation(SpecViolation) or CheckError::Setup(SetupError). Assert on fields, not text:

let Err(CheckError::Violation(v)) = recording.check() else { panic!() };
assert_eq!((v.spec.as_str(), v.key.as_str()), ("InventoryCorrect", "coffee"));
// v.message, v.location (P assert file:line), v.events (last 20, with src),
// v.liveness (hot state under check_hot), v.others (more violations)

SetupError.kind is one of:

Reference

Settings

Set in [package.metadata.snappea] of the crate's Cargo.toml (paths relative to it), or with the Model builders. Precedence: builder, environment variable, Cargo.toml, default.

Environment variables

Suite coverage

Set coverage (or SNAPPEA_COVERAGE) and every check merges the P spec handlers, statements and branches it reached into that file, under a lock, so parallel tests and test binaries share it. The summary prints when coverage changes (cargo test -- --nocapture shows it). Delete a stale file before a run.

API

• snappea::test on docs.rs
• snappea::emit on docs.rs

Runnable projects: Rust examples. The checker itself is also a library: using the checker from Rust.