Merge dbsp/step4-aggregates into dbsp/step5-formulas

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdfEGXA7aJ6ftMsCc9PJze

crates/zset-ffi/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "zset-ffi"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+description = "A C ABI over zset (byte-string keys, i64 weights) for CFFI callers: cdylib, staticlib and rlib."
+publish = false
+
+# Native only, and deliberately not in the wasm bundle: `zset` stays std-only. The
+# library is named zset_ffi, so the artifacts are libzset_ffi.{so,dylib,a}.
+[lib]
+crate-type = ["cdylib", "staticlib", "rlib"]
+
+[dependencies]
+zset = { path = "../zset" }
+
+[dev-dependencies]
+libloading = "0.8"

crates/zset-ffi/include/zset_ffi.h

@@ -0,0 +1,150 @@
+/* zset_ffi.h: the C ABI of crates/zset-ffi (libzset_ffi.{so,dylib,a}).
+ * Hand-written; tests/header.rs checks it against src/lib.rs.
+ * Reasoning: context/code/zset-ffi.md.
+ *
+ * Conventions
+ *  - Every function but the frees, zs_abi_version and zs_last_error returns int32_t: 0 is
+ *    success, anything else is a ZS_ERR_* code.
+ *    Results come back through out-pointers, which are checked FIRST (a NULL out is
+ *    ZS_ERR_NULL_ARGUMENT with no state changed) and written only on success.
+ *    On failure zs_last_error() describes it (per thread, valid until the next failure).
+ *  - Handles (ZsZSet, ZsCircuit, ZsRebuild) are opaque. Each is created by a function that
+ *    takes ZsX **out and is released by exactly one zs_X_free. Freeing NULL is a no-op;
+ *    freeing twice or using a freed handle is undefined.
+ *  - Keys are byte strings: (pointer, length). The library copies them; the caller keeps
+ *    ownership of every buffer it passes, and may reuse it after the call returns.
+ *    A zero length accepts a NULL pointer.
+ *  - The library never returns memory the caller must free, except handles.
+ *  - Weights are int64_t. A weight that would leave int64_t is ZS_ERR_OVERFLOW, never a
+ *    wrong value, and never an unwinding exception.
+ *  - Not thread-safe. A ZsCircuit and every ZsRebuild made from it form ONE group (they share
+ *    a reference count): keep the whole group on one thread at a time. Z-sets are separate
+ *    handles. Streams, inputs and sinks are uint32_t ids local to their circuit; a sink id is
+ *    never a stream id (they are refused as ZS_ERR_BAD_HANDLE), but an id from another circuit
+ *    is not detected. A ZsRebuild is stepped only on the circuit that made it (else
+ *    ZS_ERR_BAD_HANDLE), even if another circuit now lives at the old address.
+ *  - Keys of 4 GiB or longer are refused with ZS_ERR_BAD_VALUE: the wire form has u32 lengths.
+ */
+#ifndef ZSET_FFI_H
+#define ZSET_FFI_H
+
+#include <stddef.h>
+#include <stdint.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define ZS_ABI_VERSION 1
+
+#define ZS_OK 0
+#define ZS_ERR_NULL_ARGUMENT 1
+#define ZS_ERR_OVERFLOW 2
+#define ZS_ERR_PANIC 3
+#define ZS_ERR_BAD_HANDLE 4
+#define ZS_ERR_BUFFER_TOO_SMALL 5
+#define ZS_ERR_BAD_VALUE 6
+#define ZS_ERR_POISONED 7
+#define ZS_ERR_CALLBACK 8
+#define ZS_ERR_INTERRUPTED 9
+
+#define ZS_AGG_COUNT 0
+#define ZS_AGG_SUM 1
+#define ZS_AGG_MIN 2
+#define ZS_AGG_MAX 3
+
+typedef struct ZsZSet ZsZSet;
+typedef struct ZsCircuit ZsCircuit;
+typedef struct ZsRebuild ZsRebuild;
+
+/* Callbacks run inside zs_circuit_step and zs_rebuild_step.
+ *  - The function pointer and its ctx are kept by the circuit and must stay valid until
+ *    zs_circuit_free. emit and emit_ctx are valid only during the callback that received them.
+ *  - A callback emits images of a key by calling emit(emit_ctx, bytes, len); the bytes are
+ *    copied during the call.
+ *  - A callback reports failure by its return value: ZsFlatMapFn non-zero, ZsKeepFn and
+ *    ZsExpiredFn negative. The step then fails with ZS_ERR_CALLBACK and the circuit is dead.
+ *  - A callback MUST return normally: no longjmp, no C++ exception, no Lisp non-local exit
+ *    through it (wrap the body in handler-case and return the failure value). If one does,
+ *    the circuit is left marked mid-step and every later call on it is ZS_ERR_POISONED.
+ *  - A callback that calls the library with its own circuit (or the rebuild job being
+ *    stepped) gets ZS_ERR_POISONED and nothing changes. zs_circuit_free on a circuit that is
+ *    mid-step, and zs_rebuild_free on a job being stepped, do nothing (the object is leaked
+ *    rather than freed under the running frames); free after the call returns.
+ *    A circuit whose step was left by a non-local exit stays mid-step, so its free leaks too. */
+typedef void (*ZsEmit)(void *emit_ctx, const uint8_t *bytes, size_t len);
+typedef int32_t (*ZsFlatMapFn)(void *ctx, const uint8_t *key, size_t len, ZsEmit emit, void *emit_ctx);
+/* 1 keep, 0 drop, negative fail. */
+typedef int32_t (*ZsKeepFn)(void *ctx, const uint8_t *key, size_t len);
+/* 1 expired, 0 not yet, negative fail. */
+typedef int32_t (*ZsExpiredFn)(void *ctx);
+
+uint32_t zs_abi_version(void);
+const char *zs_last_error(void);
+
+/* Z-sets. add/sub/neg/scale/decode_add leave the set unchanged when they fail. add and sub
+ * accept the same handle as both arguments. */
+int32_t zs_zset_new(ZsZSet **out);
+void zs_zset_free(ZsZSet *z);
+int32_t zs_zset_clone(const ZsZSet *z, ZsZSet **out);
+int32_t zs_zset_insert(ZsZSet *z, const uint8_t *key, size_t len, int64_t weight);
+int32_t zs_zset_weight(const ZsZSet *z, const uint8_t *key, size_t len, int64_t *out);
+int32_t zs_zset_len(const ZsZSet *z, size_t *out);
+int32_t zs_zset_add(ZsZSet *a, const ZsZSet *b);
+int32_t zs_zset_sub(ZsZSet *a, const ZsZSet *b);
+int32_t zs_zset_neg(ZsZSet *a);
+int32_t zs_zset_scale(ZsZSet *a, int64_t n);
+int32_t zs_zset_distinct(const ZsZSet *z, ZsZSet **out);
+
+/* Encoding: entries ascending by key, each [u32 LE key length][key][i64 LE weight].
+ * *needed is set whenever z is valid; the bytes are written only if cap >= *needed, else
+ * ZS_ERR_BUFFER_TOO_SMALL (call again with a larger buffer). buf may be NULL only when
+ * *needed is zero or to probe with cap too small; otherwise ZS_ERR_NULL_ARGUMENT. */
+int32_t zs_zset_encode(const ZsZSet *z, uint8_t *buf, size_t cap, size_t *needed);
+int32_t zs_zset_decode_add(ZsZSet *z, const uint8_t *buf, size_t len);
+
+/* Circuits. Every stream carries a Z-set of byte-string keys. */
+int32_t zs_circuit_new(ZsCircuit **out);
+void zs_circuit_free(ZsCircuit *c);
+int32_t zs_circuit_input(ZsCircuit *c, uint32_t *out);
+int32_t zs_circuit_output(ZsCircuit *c, uint32_t stream, uint32_t *out);
+int32_t zs_circuit_delay(ZsCircuit *c, uint32_t stream, uint32_t *out);
+int32_t zs_circuit_integrate(ZsCircuit *c, uint32_t stream, uint32_t *out);
+int32_t zs_circuit_differentiate(ZsCircuit *c, uint32_t stream, uint32_t *out);
+int32_t zs_circuit_distinct(ZsCircuit *c, uint32_t stream, uint32_t *out);
+int32_t zs_circuit_plus(ZsCircuit *c, uint32_t a, uint32_t b, uint32_t *out);
+int32_t zs_circuit_filter(ZsCircuit *c, uint32_t stream, ZsKeepFn keep, void *ctx, uint32_t *out);
+int32_t zs_circuit_flat_map(ZsCircuit *c, uint32_t stream, ZsFlatMapFn f, void *ctx, uint32_t *out);
+
+/* index_by: f emits exactly one group key g per element; the output element is
+ * pair(g, element), pair(k, v) = [u32 LE length of k][k][v]. */
+int32_t zs_circuit_index_by(ZsCircuit *c, uint32_t stream, ZsFlatMapFn f, void *ctx, uint32_t *out);
+
+/* join: both inputs carry pair(k, v); the output element is pair(k, pair(a, b)). */
+int32_t zs_circuit_join(ZsCircuit *c, uint32_t a, uint32_t b, uint32_t *out);
+
+/* aggregate: the input carries pair(k, v) with v an 8-byte little-endian int64 (any bytes
+ * for COUNT); the output is pair(k, 8-byte little-endian int64). COUNT is the total weight,
+ * SUM is the weighted sum (a non-empty group always has a row, even at zero), MIN and MAX read the values of positive weight. */
+int32_t zs_circuit_aggregate(ZsCircuit *c, uint32_t stream, int32_t kind, uint32_t *out);
+
+int32_t zs_circuit_push(ZsCircuit *c, uint32_t input, const ZsZSet *z);
+
+/* One clock tick. After a failure in a step the circuit is poisoned: every later call that
+ * uses it returns ZS_ERR_POISONED, except the free. */
+int32_t zs_circuit_step(ZsCircuit *c);
+int32_t zs_circuit_take(ZsCircuit *c, uint32_t sink, ZsZSet **out);
+
+/* Resumable snapshot load: chunk keys per piece (0 reads as 1). The whole load is one
+ * logical tick, and done is 1 as soon as the last chunk ran. While a job is pending it
+ * owns the clock: a zs_circuit_step in between makes the next zs_rebuild_step return
+ * ZS_ERR_INTERRUPTED (the circuit stays usable; drop the job). */
+int32_t zs_rebuild_new(ZsCircuit *c, uint32_t input, const ZsZSet *snapshot, size_t chunk, ZsRebuild **out);
+int32_t zs_rebuild_step(ZsRebuild *job, ZsCircuit *c, ZsExpiredFn expired, void *ctx, int32_t *done);
+void zs_rebuild_free(ZsRebuild *job);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif

crates/zset-ffi/src/engine.rs

@@ -0,0 +1,520 @@
+//! The byte-keyed circuit behind the C ABI, as plain `Result`-returning Rust; panics from
+//! `zset` are caught and classified here. See `context/code/zset-ffi.md`.
+
+use std::any::Any;
+use std::cell::Cell;
+use std::ffi::c_void;
+use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::rc::Rc;
+use zset::{Circuit, Input, Output, Progress, Rebuild, Stream, ZSet};
+
+pub type Bytes = Vec<u8>;
+pub type BZ = ZSet<Bytes>;
+
+/// Declares the error codes once: the enum, the list of all of them and the header's name.
+macro_rules! codes {
+    ($($variant:ident = $n:literal => $name:literal,)*) => {
+        /// The error codes of the ABI; 0 is success and is not a variant.
+        #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+        #[repr(i32)]
+        pub enum Code { $($variant = $n,)* }
+
+        impl Code {
+            pub const ALL: &[Code] = &[$(Code::$variant,)*];
+
+            /// The `#define` this code has in the header.
+            pub fn header_name(self) -> &'static str {
+                match self { $(Code::$variant => $name,)* }
+            }
+        }
+    };
+}
+
+codes! {
+    NullArgument = 1 => "ZS_ERR_NULL_ARGUMENT",
+    Overflow = 2 => "ZS_ERR_OVERFLOW",
+    Panic = 3 => "ZS_ERR_PANIC",
+    BadHandle = 4 => "ZS_ERR_BAD_HANDLE",
+    BufferTooSmall = 5 => "ZS_ERR_BUFFER_TOO_SMALL",
+    BadValue = 6 => "ZS_ERR_BAD_VALUE",
+    Poisoned = 7 => "ZS_ERR_POISONED",
+    Callback = 8 => "ZS_ERR_CALLBACK",
+    Interrupted = 9 => "ZS_ERR_INTERRUPTED",
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct Fail {
+    pub code: Code,
+    pub msg: String,
+}
+
+impl Fail {
+    pub fn new(code: Code, msg: &str) -> Fail {
+        Fail {
+            code,
+            msg: msg.to_string(),
+        }
+    }
+}
+
+type Res<T> = Result<T, Fail>;
+
+const OVERFLOW_PREFIX: &str = "zset: weight overflow";
+
+/// Turns a caught panic into an error: weight overflow is `Overflow`, anything else `Panic`.
+pub fn classify(payload: Box<dyn Any + Send>) -> Fail {
+    let msg = payload
+        .downcast_ref::<String>()
+        .cloned()
+        .or_else(|| payload.downcast_ref::<&str>().map(|s| s.to_string()))
+        .unwrap_or_else(|| "panic with a non-string payload".to_string());
+    let code = if msg.starts_with(OVERFLOW_PREFIX) {
+        Code::Overflow
+    } else {
+        Code::Panic
+    };
+    Fail { code, msg }
+}
+
+/// A length as the wire's `u32`; a longer one is refused rather than truncated.
+pub fn len32(n: usize) -> Res<u32> {
+    u32::try_from(n).map_err(|_| Fail::new(Code::BadValue, "a key is 4 GiB or longer"))
+}
+
+/// `(key, value)` as one byte string: a little-endian `u32` key length, the key, the value.
+pub fn pair(k: &[u8], v: &[u8]) -> Res<Bytes> {
+    let mut out = Vec::with_capacity(4 + k.len() + v.len());
+    out.extend_from_slice(&len32(k.len())?.to_le_bytes());
+    out.extend_from_slice(k);
+    out.extend_from_slice(v);
+    Ok(out)
+}
+
+pub fn unpair(b: &[u8]) -> Option<(&[u8], &[u8])> {
+    let n = u32::from_le_bytes(b.get(..4)?.try_into().ok()?) as usize;
+    let rest = &b[4..];
+    (rest.len() >= n).then(|| rest.split_at(n))
+}
+
+/// Entries as `[u32 LE key length][key][i64 LE weight]`, ascending by key.
+pub fn encode(z: &BZ) -> Res<Bytes> {
+    let mut out = Vec::new();
+    for (k, w) in z.iter() {
+        out.extend_from_slice(&len32(k.len())?.to_le_bytes());
+        out.extend_from_slice(k);
+        out.extend_from_slice(&w.to_le_bytes());
+    }
+    Ok(out)
+}
+
+/// The inverse of `encode`; equal keys sum, and an overflow panics like any addition.
+pub fn decode(mut b: &[u8]) -> Res<BZ> {
+    let bad = || Fail::new(Code::BadValue, "malformed zset encoding");
+    let mut z = ZSet::new();
+    while !b.is_empty() {
+        let n = u32::from_le_bytes(b.get(..4).ok_or_else(bad)?.try_into().unwrap()) as usize;
+        let key = b.get(4..4 + n).ok_or_else(bad)?;
+        let w = b.get(4 + n..12 + n).ok_or_else(bad)?;
+        z.insert(key.to_vec(), i64::from_le_bytes(w.try_into().unwrap()));
+        b = &b[12 + n..];
+    }
+    Ok(z)
+}
+
+pub type Emit = extern "C" fn(*mut c_void, *const u8, usize);
+/// Returns 0, or non-zero to fail the step.
+pub type FlatMapFn = extern "C" fn(*mut c_void, *const u8, usize, Emit, *mut c_void) -> i32;
+/// Returns 0 (drop), 1 (keep), or negative to fail the step.
+pub type KeepFn = extern "C" fn(*mut c_void, *const u8, usize) -> i32;
+/// Returns 0 (still time), 1 (expired), or negative to fail the step.
+pub type ExpiredFn = extern "C" fn(*mut c_void) -> i32;
+
+/// A caller's context pointer; the circuit never dereferences it.
+#[derive(Clone, Copy)]
+struct Ctx(*mut c_void);
+
+type Fault = Rc<Cell<Option<(Code, &'static str)>>>;
+
+/// Records a fault unless one is already recorded: the first one is the cause.
+fn note(fault: &Fault, code: Code, msg: &'static str) {
+    if let Some(first) = fault.take() {
+        fault.set(Some(first));
+    } else {
+        fault.set(Some((code, msg)));
+    }
+}
+
+extern "C" fn collect(ctx: *mut c_void, p: *const u8, n: usize) {
+    // SAFETY: `ctx` is the `Vec` `images` passes below, alive for the callback's duration.
+    let out = unsafe { &mut *(ctx as *mut Vec<Bytes>) };
+    let bytes = if n == 0 || p.is_null() {
+        Vec::new()
+    } else {
+        // SAFETY: the caller of `emit` promises `n` readable bytes at `p`.
+        unsafe { std::slice::from_raw_parts(p, n) }.to_vec()
+    };
+    out.push(bytes);
+}
+
+/// The images a callback emits for `key`; a failing callback records a fault and emits none.
+fn images(f: FlatMapFn, ctx: Ctx, key: &Bytes, fault: &Fault) -> Vec<Bytes> {
+    let mut out: Vec<Bytes> = Vec::new();
+    let rc = f(
+        ctx.0,
+        key.as_ptr(),
+        key.len(),
+        collect,
+        &mut out as *mut Vec<Bytes> as *mut c_void,
+    );
+    if rc != 0 {
+        note(fault, Code::Callback, "a callback reported failure");
+        out.clear();
+    }
+    out
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Agg {
+    Count,
+    Sum,
+    Min,
+    Max,
+}
+
+impl Agg {
+    pub fn from_code(n: i32) -> Res<Agg> {
+        Ok(match n {
+            0 => Agg::Count,
+            1 => Agg::Sum,
+            2 => Agg::Min,
+            3 => Agg::Max,
+            _ => return Err(Fail::new(Code::BadValue, "unknown aggregate kind")),
+        })
+    }
+}
+
+fn checked(v: Option<i64>) -> i64 {
+    v.unwrap_or_else(|| panic!("{OVERFLOW_PREFIX} in aggregate"))
+}
+
+/// Sink ids carry this bit; a stream id is an index far below it, so `streams.get` refuses a
+/// sink id and `take` refuses a stream id.
+const SINK_BIT: u32 = 1 << 31;
+
+/// The circuit, its streams by id, and the first fault a closure recorded.
+pub struct Engine {
+    circuit: Circuit,
+    streams: Vec<Stream<BZ>>,
+    inputs: Vec<Option<Input<Bytes>>>,
+    sinks: Vec<Output<Bytes>>,
+    fault: Fault,
+    poisoned: bool,
+}
+
+impl Default for Engine {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Engine {
+    pub fn new() -> Engine {
+        Engine {
+            circuit: Circuit::new(),
+            streams: Vec::new(),
+            inputs: Vec::new(),
+            sinks: Vec::new(),
+            fault: Rc::default(),
+            poisoned: false,
+        }
+    }
+
+    fn stream(&self, id: u32) -> Res<Stream<BZ>> {
+        self.streams
+            .get(id as usize)
+            .cloned()
+            .ok_or_else(|| Fail::new(Code::BadHandle, "no such stream"))
+    }
+
+    fn register(&mut self, s: Stream<BZ>, input: Option<Input<Bytes>>) -> u32 {
+        self.streams.push(s);
+        self.inputs.push(input);
+        (self.streams.len() - 1) as u32
+    }
+
+    fn alive(&self) -> Res<()> {
+        if self.poisoned {
+            return Err(Fail::new(
+                Code::Poisoned,
+                "the circuit failed in a step and cannot be used again",
+            ));
+        }
+        Ok(())
+    }
+
+    pub fn input(&mut self) -> Res<u32> {
+        self.alive()?;
+        let (i, s) = self.circuit.input::<Bytes>();
+        Ok(self.register(s, Some(i)))
+    }
+
+    pub fn output(&mut self, s: u32) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        self.sinks.push(self.circuit.output(&s));
+        Ok((self.sinks.len() - 1) as u32 | SINK_BIT)
+    }
+
+    pub fn delay(&mut self, s: u32) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let out = self.circuit.delay(&s);
+        Ok(self.register(out, None))
+    }
+
+    pub fn integrate(&mut self, s: u32) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let out = self.circuit.integrate(&s);
+        Ok(self.register(out, None))
+    }
+
+    pub fn differentiate(&mut self, s: u32) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let out = self.circuit.differentiate(&s);
+        Ok(self.register(out, None))
+    }
+
+    pub fn distinct(&mut self, s: u32) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let out = self.circuit.distinct(&s);
+        Ok(self.register(out, None))
+    }
+
+    pub fn plus(&mut self, a: u32, b: u32) -> Res<u32> {
+        self.alive()?;
+        let (a, b) = (self.stream(a)?, self.stream(b)?);
+        let out = self.circuit.plus(&a, &b);
+        Ok(self.register(out, None))
+    }
+
+    pub fn filter(&mut self, s: u32, keep: KeepFn, ctx: *mut c_void) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault));
+        let out = self.circuit.filter(&s, move |k: &Bytes| {
+            match keep(ctx.0, k.as_ptr(), k.len()) {
+                0 => false,
+                n if n > 0 => true,
+                _ => {
+                    note(&fault, Code::Callback, "a callback reported failure");
+                    false
+                }
+            }
+        });
+        Ok(self.register(out, None))
+    }
+
+    pub fn flat_map(&mut self, s: u32, f: FlatMapFn, ctx: *mut c_void) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault));
+        let out = self
+            .circuit
+            .flat_map(&s, move |k: &Bytes| images(f, ctx, k, &fault));
+        Ok(self.register(out, None))
+    }
+
+    /// Each element becomes `pair(g, element)` where the callback emits exactly one `g`.
+    pub fn index_by(&mut self, s: u32, f: FlatMapFn, ctx: *mut c_void) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let (ctx, fault) = (Ctx(ctx), Rc::clone(&self.fault));
+        let out = self.circuit.flat_map(&s, move |k: &Bytes| {
+            let mut gs = images(f, ctx, k, &fault);
+            if gs.len() != 1 {
+                note(
+                    &fault,
+                    Code::BadValue,
+                    "index_by: the key callback must emit exactly one key",
+                );
+                return Vec::new();
+            }
+            match pair(&gs.remove(0), k) {
+                Ok(p) => vec![p],
+                Err(_) => {
+                    note(&fault, Code::BadValue, "a key is 4 GiB or longer");
+                    Vec::new()
+                }
+            }
+        });
+        Ok(self.register(out, None))
+    }
+
+    /// Splits `pair(k, v)` elements; a malformed one records a fault and is dropped.
+    fn keyed(&mut self, s: &Stream<BZ>) -> Stream<ZSet<(Bytes, Bytes)>> {
+        let fault = Rc::clone(&self.fault);
+        self.circuit.flat_map(s, move |b: &Bytes| match unpair(b) {
+            Some((k, v)) => vec![(k.to_vec(), v.to_vec())],
+            None => {
+                note(
+                    &fault,
+                    Code::BadValue,
+                    "a keyed stream carried an element that is not a pair",
+                );
+                Vec::new()
+            }
+        })
+    }
+
+    /// Joins two streams of `pair(k, v)`; the output is `pair(k, pair(a, b))`.
+    pub fn join(&mut self, a: u32, b: u32) -> Res<u32> {
+        self.alive()?;
+        let (a, b) = (self.stream(a)?, self.stream(b)?);
+        let (ka, kb) = (self.keyed(&a), self.keyed(&b));
+        let fault = Rc::clone(&self.fault);
+        let j = self
+            .circuit
+            .join_indexed(&ka, &kb, move |k: &Bytes, x: &Bytes, y: &Bytes| {
+                pair(x, y).and_then(|xy| pair(k, &xy)).unwrap_or_else(|_| {
+                    note(&fault, Code::BadValue, "a key is 4 GiB or longer");
+                    Vec::new()
+                })
+            });
+        Ok(self.register(j, None))
+    }
+
+    /// Groups a stream of `pair(k, v)` (v is 8 bytes, little-endian `i64`, except for
+    /// `Count`); the output is `pair(k, 8 bytes)`. A non-empty group always has a row.
+    pub fn aggregate(&mut self, s: u32, kind: Agg) -> Res<u32> {
+        self.alive()?;
+        let s = self.stream(s)?;
+        let keyed = self.keyed(&s);
+        let fault = Rc::clone(&self.fault);
+        let folded = self
+            .circuit
+            .aggregate(&keyed, move |_, vals: &[(Bytes, i64)]| {
+                if kind == Agg::Count {
+                    let n = vals.iter().try_fold(0i64, |a, (_, w)| a.checked_add(*w));
+                    return Some(checked(n));
+                }
+                let mut nums = Vec::with_capacity(vals.len());
+                for (v, w) in vals {
+                    match <[u8; 8]>::try_from(v.as_slice()) {
+                        Ok(b) => nums.push((i64::from_le_bytes(b), *w)),
+                        Err(_) => {
+                            note(&fault, Code::BadValue, "an aggregated value is not 8 bytes");
+                            return None;
+                        }
+                    }
+                }
+                match kind {
+                    Agg::Sum => {
+                        let t = nums.iter().try_fold(0i64, |a, (v, w)| {
+                            v.checked_mul(*w).and_then(|p| a.checked_add(p))
+                        });
+                        Some(checked(t))
+                    }
+                    Agg::Min => nums.iter().filter(|(_, w)| *w > 0).map(|(v, _)| *v).min(),
+                    Agg::Max | Agg::Count => {
+                        nums.iter().filter(|(_, w)| *w > 0).map(|(v, _)| *v).max()
+                    }
+                }
+            });
+        let fault = Rc::clone(&self.fault);
+        let out = self
+            .circuit
+            .flat_map(&folded, move |(k, r): &(Bytes, i64)| {
+                match pair(k, &r.to_le_bytes()) {
+                    Ok(p) => vec![p],
+                    Err(_) => {
+                        note(&fault, Code::BadValue, "a key is 4 GiB or longer");
+                        Vec::new()
+                    }
+                }
+            });
+        Ok(self.register(out, None))
+    }
+
+    fn input_of(&self, id: u32) -> Res<Input<Bytes>> {
+        self.inputs
+            .get(id as usize)
+            .and_then(|i| i.clone())
+            .ok_or_else(|| Fail::new(Code::BadHandle, "that stream is not an input"))
+    }
+
+    pub fn push(&mut self, input: u32, z: &BZ) -> Res<()> {
+        self.alive()?;
+        self.input_of(input)?.push(z.clone());
+        Ok(())
+    }
+
+    /// Runs `f` on the circuit; a panic or a recorded fault poisons the engine.
+    fn guarded<T>(&mut self, f: impl FnOnce(&mut Circuit) -> T) -> Res<T> {
+        self.alive()?;
+        self.fault.set(None);
+        match catch_unwind(AssertUnwindSafe(|| f(&mut self.circuit))) {
+            Err(payload) => {
+                self.poisoned = true;
+                Err(classify(payload))
+            }
+            Ok(v) => match self.fault.take() {
+                Some((code, msg)) => {
+                    self.poisoned = true;
+                    Err(Fail::new(code, msg))
+                }
+                None => Ok(v),
+            },
+        }
+    }
+
+    pub fn step(&mut self) -> Res<()> {
+        self.guarded(|c| c.step())
+    }
+
+    pub fn take(&mut self, sink: u32) -> Res<BZ> {
+        self.alive()?;
+        self.sinks
+            .get((sink & !SINK_BIT) as usize)
+            .filter(|_| sink & SINK_BIT != 0)
+            .map(|s| s.take())
+            .ok_or_else(|| Fail::new(Code::BadHandle, "no such output"))
+    }
+
+    pub fn start_rebuild(&mut self, input: u32, snapshot: BZ, chunk: usize) -> Res<Rebuild<Bytes>> {
+        self.alive()?;
+        Ok(Rebuild::new(self.input_of(input)?, snapshot, chunk))
+    }
+
+    /// `true` when the snapshot is consumed. With no `expired`, runs to completion.
+    pub fn rebuild_step(
+        &mut self,
+        job: &mut Rebuild<Bytes>,
+        expired: Option<(ExpiredFn, *mut c_void)>,
+    ) -> Res<bool> {
+        self.alive()?;
+        if job.interrupted(&self.circuit) {
+            return Err(Fail::new(
+                Code::Interrupted,
+                "the rebuild was interrupted by a circuit step before it finished",
+            ));
+        }
+        let fault = Rc::clone(&self.fault);
+        let progress = self.guarded(|c| {
+            job.step(c, || match expired {
+                None => false,
+                Some((f, ctx)) => match f(ctx) {
+                    0 => false,
+                    n if n > 0 => true,
+                    _ => {
+                        note(&fault, Code::Callback, "a callback reported failure");
+                        true
+                    }
+                },
+            })
+        })?;
+        Ok(progress == Progress::Done)
+    }
+}

crates/zset-ffi/src/lib.rs

@@ -0,0 +1,612 @@
+//! The C ABI over `zset`: byte keys, `i64` weights, opaque handles, `i32` codes.
+//! Ownership and errors: `context/code/zset-ffi.md` and `include/zset_ffi.h`.
+
+pub mod engine;
+
+use engine::{
+    Agg, BZ, Bytes, Code, Engine, ExpiredFn, Fail, FlatMapFn, KeepFn, classify, decode, encode,
+};
+use std::cell::{Cell, RefCell, UnsafeCell};
+use std::ffi::{CString, c_char, c_void};
+use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::sync::atomic::{AtomicU64, Ordering};
+use zset::{Rebuild, ZSet};
+
+pub const ZS_ABI_VERSION: u32 = 1;
+
+/// An opaque Z-set of byte-string keys.
+pub struct ZsZSet(BZ);
+
+/// An opaque circuit. `in_step` is set while the engine may be running callbacks; it is
+/// still set if a caller's non-local exit skipped our frames, and then the circuit is dead.
+pub struct ZsCircuit {
+    engine: UnsafeCell<Engine>,
+    in_step: Cell<bool>,
+    id: u64,
+}
+
+static NEXT_CIRCUIT_ID: AtomicU64 = AtomicU64::new(1);
+
+/// An opaque resumable snapshot load, bound to the circuit that made it.
+pub struct ZsRebuild {
+    job: Rebuild<Bytes>,
+    owner: u64,
+    busy: Cell<bool>,
+}
+
+thread_local! {
+    static LAST_ERROR: RefCell<CString> = RefCell::new(CString::default());
+}
+
+fn remember(msg: &str) {
+    let clean: String = msg.chars().filter(|c| *c != '\0').collect();
+    LAST_ERROR.with(|e| *e.borrow_mut() = CString::new(clean).unwrap_or_default());
+}
+
+/// Runs an entry point body: no panic crosses the ABI, and every failure is a code.
+fn guard(f: impl FnOnce() -> Result<(), Fail>) -> i32 {
+    let fail = match catch_unwind(AssertUnwindSafe(f)) {
+        Ok(Ok(())) => return 0,
+        Ok(Err(fail)) => fail,
+        Err(payload) => classify(payload),
+    };
+    remember(&fail.msg);
+    fail.code as i32
+}
+
+fn null() -> Fail {
+    Fail::new(Code::NullArgument, "a required argument was null")
+}
+
+/// # Safety
+/// `p` is null or points to a live `T` that nothing else is using.
+unsafe fn as_mut<'a, T>(p: *mut T) -> Result<&'a mut T, Fail> {
+    // SAFETY: the caller's contract.
+    unsafe { p.as_mut() }.ok_or_else(null)
+}
+
+/// # Safety
+/// `p` is null or points to a live `T` that nothing is mutating.
+unsafe fn as_ref<'a, T>(p: *const T) -> Result<&'a T, Fail> {
+    // SAFETY: the caller's contract.
+    unsafe { p.as_ref() }.ok_or_else(null)
+}
+
+/// # Safety
+/// `p` is null (with `n` zero) or has `n` readable bytes.
+unsafe fn bytes<'a>(p: *const u8, n: usize) -> Result<&'a [u8], Fail> {
+    match (p.is_null(), n) {
+        (_, 0) => Ok(&[]),
+        (true, _) => Err(null()),
+        // SAFETY: the caller's contract.
+        (false, _) => Ok(unsafe { std::slice::from_raw_parts(p, n) }),
+    }
+}
+
+/// # Safety
+/// `out` is null or valid for a write of `T`.
+unsafe fn with_out<T>(out: *mut T, f: impl FnOnce() -> Result<T, Fail>) -> Result<(), Fail> {
+    // SAFETY: the caller's contract.
+    let out = unsafe { as_mut(out)? };
+    *out = f()?;
+    Ok(())
+}
+
+/// # Safety
+/// `c` is null or a live circuit handle. Callbacks may call back in: this only reads `in_step`.
+unsafe fn circuit<'a>(c: *const ZsCircuit) -> Result<&'a ZsCircuit, Fail> {
+    // SAFETY: the caller's contract; only a shared reference is made.
+    let c = unsafe { as_ref(c)? };
+    if c.in_step.get() {
+        return Err(Fail::new(
+            Code::Poisoned,
+            "the circuit is mid-step: a callback re-entered it, or left it without returning",
+        ));
+    }
+    Ok(c)
+}
+
+impl ZsCircuit {
+    /// Only `circuit` hands out `&ZsCircuit` while no step is running, and the handle is used
+    /// from one thread, so nothing else holds the engine.
+    #[allow(clippy::mut_from_ref)]
+    fn engine(&self) -> &mut Engine {
+        // SAFETY: see above.
+        unsafe { &mut *self.engine.get() }
+    }
+
+    /// Runs `f` with `in_step` set; a frame skipped by a caller's non-local exit leaves it set.
+    fn stepping<T>(&self, f: impl FnOnce(&mut Engine) -> T) -> T {
+        self.in_step.set(true);
+        let v = f(self.engine());
+        self.in_step.set(false);
+        v
+    }
+}
+
+/// # Safety
+/// `p` is null, or came from this library's constructor and is freed once.
+unsafe fn release<T>(p: *mut T) {
+    if !p.is_null() {
+        // SAFETY: the caller's contract.
+        drop(unsafe { Box::from_raw(p) });
+    }
+}
+
+fn boxed<T>(v: T) -> Result<*mut T, Fail> {
+    Ok(Box::into_raw(Box::new(v)))
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn zs_abi_version() -> u32 {
+    ZS_ABI_VERSION
+}
+
+/// The message of the last failure on this thread; never null, valid until the next failure.
+#[unsafe(no_mangle)]
+pub extern "C" fn zs_last_error() -> *const c_char {
+    LAST_ERROR.with(|e| e.borrow().as_ptr())
+}
+
+/// # Safety
+/// `out` is null or valid for a write.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_new(out: *mut *mut ZsZSet) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || boxed(ZsZSet(ZSet::new()))) })
+}
+
+/// # Safety
+/// `z` is null or a live Z-set handle, freed once.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_free(z: *mut ZsZSet) {
+    // SAFETY: the caller's contract.
+    unsafe { release(z) }
+}
+
+/// # Safety
+/// `z` is a live handle or null; `out` is null or valid for a write.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_clone(z: *const ZsZSet, out: *mut *mut ZsZSet) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || boxed(ZsZSet(as_ref(z)?.0.clone()))) })
+}
+
+/// # Safety
+/// `z` is a live handle or null; `key` has `len` readable bytes.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_insert(
+    z: *mut ZsZSet,
+    key: *const u8,
+    len: usize,
+    weight: i64,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        let (z, key) = unsafe { (as_mut(z)?, bytes(key, len)?) };
+        z.0.insert(key.to_vec(), weight);
+        Ok(())
+    })
+}
+
+/// # Safety
+/// `z` is a live handle or null; `key` has `len` readable bytes; `out` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_weight(
+    z: *const ZsZSet,
+    key: *const u8,
+    len: usize,
+    out: *mut i64,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe { with_out(out, || Ok(as_ref(z)?.0.weight(&bytes(key, len)?.to_vec()))) }
+    })
+}
+
+/// # Safety
+/// `z` is a live handle or null; `out` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_len(z: *const ZsZSet, out: *mut usize) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || Ok(as_ref(z)?.0.len())) })
+}
+
+/// # Safety
+/// Both are live handles or null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_add(a: *mut ZsZSet, b: *const ZsZSet) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract; the shared borrows end before the mutable one.
+        let sum = unsafe { &as_ref(a)?.0 + &as_ref(b)?.0 };
+        unsafe { as_mut(a)? }.0 = sum;
+        Ok(())
+    })
+}
+
+/// # Safety
+/// Both are live handles or null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_sub(a: *mut ZsZSet, b: *const ZsZSet) -> i32 {
+    guard(|| {
+        // SAFETY: as `zs_zset_add`.
+        let diff = unsafe { &as_ref(a)?.0 - &as_ref(b)?.0 };
+        unsafe { as_mut(a)? }.0 = diff;
+        Ok(())
+    })
+}
+
+/// # Safety
+/// `a` is a live handle or null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_neg(a: *mut ZsZSet) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        let a = unsafe { as_mut(a)? };
+        a.0 = -&a.0;
+        Ok(())
+    })
+}
+
+/// # Safety
+/// `a` is a live handle or null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_scale(a: *mut ZsZSet, n: i64) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        let a = unsafe { as_mut(a)? };
+        a.0 = a.0.scale(n);
+        Ok(())
+    })
+}
+
+/// # Safety
+/// `z` is a live handle or null; `out` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_distinct(z: *const ZsZSet, out: *mut *mut ZsZSet) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || boxed(ZsZSet(as_ref(z)?.0.distinct()))) })
+}
+
+/// # Safety
+/// `z` is a live handle or null; `buf` is null or has `cap` writable bytes; `needed` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_encode(
+    z: *const ZsZSet,
+    buf: *mut u8,
+    cap: usize,
+    needed: *mut usize,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        let (z, needed) = unsafe { (as_ref(z)?, as_mut(needed)?) };
+        let enc = encode(&z.0)?;
+        *needed = enc.len();
+        if enc.len() > cap {
+            return Err(Fail::new(
+                Code::BufferTooSmall,
+                "the buffer is smaller than *needed",
+            ));
+        }
+        if !enc.is_empty() {
+            if buf.is_null() {
+                return Err(null());
+            }
+            // SAFETY: `buf` is non-null and has `cap >= enc.len()` writable bytes.
+            unsafe { std::ptr::copy_nonoverlapping(enc.as_ptr(), buf, enc.len()) };
+        }
+        Ok(())
+    })
+}
+
+/// # Safety
+/// `z` is a live handle or null; `buf` has `len` readable bytes.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_zset_decode_add(z: *mut ZsZSet, buf: *const u8, len: usize) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        let (zr, buf) = unsafe { (as_ref(z)?, bytes(buf, len)?) };
+        let sum = &zr.0 + &decode(buf)?;
+        unsafe { as_mut(z)? }.0 = sum;
+        Ok(())
+    })
+}
+
+/// # Safety
+/// `out` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_new(out: *mut *mut ZsCircuit) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe {
+            with_out(out, || {
+                boxed(ZsCircuit {
+                    engine: UnsafeCell::new(Engine::new()),
+                    in_step: Cell::new(false),
+                    id: NEXT_CIRCUIT_ID.fetch_add(1, Ordering::Relaxed),
+                })
+            })
+        }
+    })
+}
+
+/// # Safety
+/// `c` is null, or a live circuit freed once and not mid-step.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_free(c: *mut ZsCircuit) {
+    // SAFETY: the caller's contract; a mid-step circuit is leaked, never freed under its frames.
+    if !c.is_null() && unsafe { &*c }.in_step.get() {
+        return;
+    }
+    unsafe { release(c) }
+}
+
+/// # Safety
+/// `c` is a live circuit or null; `out` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_input(c: *mut ZsCircuit, out: *mut u32) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || circuit(c)?.engine().input()) })
+}
+
+/// # Safety
+/// As `zs_circuit_input`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_output(c: *mut ZsCircuit, stream: u32, out: *mut u32) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || circuit(c)?.engine().output(stream)) })
+}
+
+macro_rules! unary_op {
+    ($name:ident, $method:ident) => {
+        /// # Safety
+        /// As `zs_circuit_input`.
+        #[unsafe(no_mangle)]
+        pub unsafe extern "C" fn $name(c: *mut ZsCircuit, stream: u32, out: *mut u32) -> i32 {
+            // SAFETY: the caller's contract.
+            guard(|| unsafe { with_out(out, || circuit(c)?.engine().$method(stream)) })
+        }
+    };
+}
+
+unary_op!(zs_circuit_delay, delay);
+unary_op!(zs_circuit_integrate, integrate);
+unary_op!(zs_circuit_differentiate, differentiate);
+unary_op!(zs_circuit_distinct, distinct);
+
+/// # Safety
+/// As `zs_circuit_input`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_plus(c: *mut ZsCircuit, a: u32, b: u32, out: *mut u32) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || circuit(c)?.engine().plus(a, b)) })
+}
+
+/// # Safety
+/// As `zs_circuit_input`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_filter(
+    c: *mut ZsCircuit,
+    stream: u32,
+    keep: Option<KeepFn>,
+    ctx: *mut c_void,
+    out: *mut u32,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe {
+            with_out(out, || {
+                circuit(c)?
+                    .engine()
+                    .filter(stream, keep.ok_or_else(null)?, ctx)
+            })
+        }
+    })
+}
+
+/// # Safety
+/// As `zs_circuit_input`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_flat_map(
+    c: *mut ZsCircuit,
+    stream: u32,
+    f: Option<FlatMapFn>,
+    ctx: *mut c_void,
+    out: *mut u32,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe {
+            with_out(out, || {
+                circuit(c)?
+                    .engine()
+                    .flat_map(stream, f.ok_or_else(null)?, ctx)
+            })
+        }
+    })
+}
+
+/// # Safety
+/// As `zs_circuit_input`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_index_by(
+    c: *mut ZsCircuit,
+    stream: u32,
+    f: Option<FlatMapFn>,
+    ctx: *mut c_void,
+    out: *mut u32,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe {
+            with_out(out, || {
+                circuit(c)?
+                    .engine()
+                    .index_by(stream, f.ok_or_else(null)?, ctx)
+            })
+        }
+    })
+}
+
+/// # Safety
+/// As `zs_circuit_input`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_join(c: *mut ZsCircuit, a: u32, b: u32, out: *mut u32) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { with_out(out, || circuit(c)?.engine().join(a, b)) })
+}
+
+/// # Safety
+/// As `zs_circuit_input`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_aggregate(
+    c: *mut ZsCircuit,
+    stream: u32,
+    kind: i32,
+    out: *mut u32,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe {
+            with_out(out, || {
+                circuit(c)?
+                    .engine()
+                    .aggregate(stream, Agg::from_code(kind)?)
+            })
+        }
+    })
+}
+
+/// # Safety
+/// `c` and `z` are live handles or null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_push(c: *mut ZsCircuit, input: u32, z: *const ZsZSet) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        let (c, z) = unsafe { (circuit(c)?, as_ref(z)?) };
+        c.engine().push(input, &z.0)
+    })
+}
+
+/// # Safety
+/// `c` is a live circuit or null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_step(c: *mut ZsCircuit) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        let c = unsafe { circuit(c)? };
+        c.stepping(|e| e.step())
+    })
+}
+
+/// # Safety
+/// `c` is a live circuit or null; `out` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_take(
+    c: *mut ZsCircuit,
+    sink: u32,
+    out: *mut *mut ZsZSet,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe { with_out(out, || boxed(ZsZSet(circuit(c)?.engine().take(sink)?))) }
+    })
+}
+
+/// # Safety
+/// `c` and `snapshot` are live handles or null; `out` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_rebuild_new(
+    c: *mut ZsCircuit,
+    input: u32,
+    snapshot: *const ZsZSet,
+    chunk: usize,
+    out: *mut *mut ZsRebuild,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe {
+            with_out(out, || {
+                let (cr, snap) = (circuit(c)?, as_ref(snapshot)?);
+                let job = cr.engine().start_rebuild(input, snap.0.clone(), chunk)?;
+                boxed(ZsRebuild {
+                    job,
+                    owner: cr.id,
+                    busy: Cell::new(false),
+                })
+            })
+        }
+    })
+}
+
+/// # Safety
+/// `job` and `c` are live handles or null; `done` is null or writable.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_rebuild_step(
+    job: *mut ZsRebuild,
+    c: *mut ZsCircuit,
+    expired: Option<ExpiredFn>,
+    ctx: *mut c_void,
+    done: *mut i32,
+) -> i32 {
+    guard(|| {
+        // SAFETY: the caller's contract.
+        unsafe {
+            with_out(done, || {
+                let cr = circuit(c)?;
+                if job.is_null() {
+                    return Err(null());
+                }
+                // Field places only, never a `&mut ZsRebuild`: a callback reads `busy` through
+                // `zs_rebuild_free` while `job.job` is mutably borrowed here.
+                if (*job).owner != cr.id {
+                    return Err(Fail::new(
+                        Code::BadHandle,
+                        "the rebuild belongs to another circuit",
+                    ));
+                }
+                (*job).busy.set(true);
+                let finished =
+                    cr.stepping(|e| e.rebuild_step(&mut (*job).job, expired.map(|f| (f, ctx))));
+                (*job).busy.set(false);
+                let finished = finished?;
+                Ok(i32::from(finished))
+            })
+        }
+    })
+}
+
+/// # Safety
+/// `job` is null, or a live job freed once.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_rebuild_free(job: *mut ZsRebuild) {
+    // SAFETY: the caller's contract; a job being stepped is leaked, not freed under its frame.
+    if !job.is_null() && unsafe { (*job).busy.get() } {
+        return;
+    }
+    unsafe { release(job) }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn a_circuit_left_mid_step_by_a_skipped_frame_answers_poisoned_to_everything() {
+        let mut c = std::ptr::null_mut();
+        assert_eq!(unsafe { zs_circuit_new(&mut c) }, 0);
+        let mut id = 0;
+        assert_eq!(unsafe { zs_circuit_input(c, &mut id) }, 0);
+        // What a non-local exit out of a callback leaves behind: the flag never cleared.
+        unsafe { &*c }.in_step.set(true);
+        assert_eq!(unsafe { zs_circuit_step(c) }, Code::Poisoned as i32);
+        id = 77;
+        assert_eq!(
+            unsafe { zs_circuit_input(c, &mut id) },
+            Code::Poisoned as i32
+        );
+        assert_eq!(id, 77, "nothing was written");
+        unsafe { &*c }.in_step.set(false);
+        unsafe { zs_circuit_free(c) };
+    }
+}

crates/zset-ffi/tests/abi.rs

@@ -0,0 +1,967 @@
+//! The ABI exercised through the `extern "C"` functions themselves, as a C caller would.
+//! Evidence: `context/test-evidence/zset-ffi.md`.
+
+use std::ffi::{CStr, c_void};
+use std::ptr::{null, null_mut};
+use zset_ffi::engine::Code;
+use zset_ffi::*;
+
+const OK: i32 = 0;
+
+fn pair(k: &[u8], v: &[u8]) -> Vec<u8> {
+    zset_ffi::engine::pair(k, v).unwrap()
+}
+
+fn zset(entries: &[(&[u8], i64)]) -> *mut ZsZSet {
+    let mut z = null_mut();
+    assert_eq!(unsafe { zs_zset_new(&mut z) }, OK);
+    for (k, w) in entries {
+        assert_eq!(unsafe { zs_zset_insert(z, k.as_ptr(), k.len(), *w) }, OK);
+    }
+    z
+}
+
+fn weight(z: *const ZsZSet, k: &[u8]) -> i64 {
+    let mut w = 0;
+    assert_eq!(
+        unsafe { zs_zset_weight(z, k.as_ptr(), k.len(), &mut w) },
+        OK
+    );
+    w
+}
+
+fn len(z: *const ZsZSet) -> usize {
+    let mut n = 0;
+    assert_eq!(unsafe { zs_zset_len(z, &mut n) }, OK);
+    n
+}
+
+fn encoded(z: *const ZsZSet) -> Vec<u8> {
+    let mut needed = 0;
+    assert_eq!(
+        unsafe { zs_zset_encode(z, null_mut(), 0, &mut needed) },
+        if needed == 0 {
+            OK
+        } else {
+            Code::BufferTooSmall as i32
+        }
+    );
+    let mut buf = vec![0u8; needed];
+    assert_eq!(
+        unsafe { zs_zset_encode(z, buf.as_mut_ptr(), buf.len(), &mut needed) },
+        OK
+    );
+    buf
+}
+
+fn last_error() -> String {
+    unsafe { CStr::from_ptr(zs_last_error()) }
+        .to_string_lossy()
+        .into_owned()
+}
+
+struct C(*mut ZsCircuit);
+
+impl C {
+    fn new() -> C {
+        let mut c = null_mut();
+        assert_eq!(unsafe { zs_circuit_new(&mut c) }, OK);
+        C(c)
+    }
+    fn input(&self) -> u32 {
+        let mut s = 0;
+        assert_eq!(unsafe { zs_circuit_input(self.0, &mut s) }, OK);
+        s
+    }
+    fn sink(&self, s: u32) -> u32 {
+        let mut o = 0;
+        assert_eq!(unsafe { zs_circuit_output(self.0, s, &mut o) }, OK);
+        o
+    }
+    fn push(&self, input: u32, entries: &[(&[u8], i64)]) {
+        let z = zset(entries);
+        assert_eq!(unsafe { zs_circuit_push(self.0, input, z) }, OK);
+        unsafe { zs_zset_free(z) };
+    }
+    /// Steps and takes the sink as sorted `(key, weight)` pairs.
+    fn run(&self, sink: u32) -> Vec<(Vec<u8>, i64)> {
+        assert_eq!(unsafe { zs_circuit_step(self.0) }, OK);
+        self.take(sink)
+    }
+    fn take(&self, sink: u32) -> Vec<(Vec<u8>, i64)> {
+        let mut z = null_mut();
+        assert_eq!(unsafe { zs_circuit_take(self.0, sink, &mut z) }, OK);
+        let enc = encoded(z);
+        unsafe { zs_zset_free(z) };
+        decode_pairs(&enc)
+    }
+}
+
+impl Drop for C {
+    fn drop(&mut self) {
+        unsafe { zs_circuit_free(self.0) }
+    }
+}
+
+fn decode_pairs(mut b: &[u8]) -> Vec<(Vec<u8>, i64)> {
+    let mut out = Vec::new();
+    while !b.is_empty() {
+        let n = u32::from_le_bytes(b[..4].try_into().unwrap()) as usize;
+        let w = i64::from_le_bytes(b[4 + n..12 + n].try_into().unwrap());
+        out.push((b[4..4 + n].to_vec(), w));
+        b = &b[12 + n..];
+    }
+    out
+}
+
+fn one(entries: &[(&[u8], i64)]) -> Vec<(Vec<u8>, i64)> {
+    entries.iter().map(|(k, w)| (k.to_vec(), *w)).collect()
+}
+
+fn le(n: i64) -> [u8; 8] {
+    n.to_le_bytes()
+}
+
+#[test]
+fn a_zset_adds_queries_and_frees() {
+    let (a, b) = (
+        zset(&[(b"x", 2), (b"y", -1)]),
+        zset(&[(b"x", -2), (b"z", 5)]),
+    );
+    assert_eq!(unsafe { zs_zset_add(a, b) }, OK);
+    assert_eq!(
+        (weight(a, b"x"), weight(a, b"y"), weight(a, b"z"), len(a)),
+        (0, -1, 5, 2)
+    );
+    assert_eq!(unsafe { zs_zset_neg(a) }, OK);
+    assert_eq!(unsafe { zs_zset_scale(a, 3) }, OK);
+    assert_eq!((weight(a, b"y"), weight(a, b"z")), (3, -15));
+    assert_eq!(unsafe { zs_zset_sub(a, a) }, OK);
+    assert_eq!(len(a), 0);
+    for z in [a, b] {
+        unsafe { zs_zset_free(z) };
+    }
+    unsafe { zs_zset_free(null_mut()) };
+}
+
+#[test]
+fn keys_are_byte_strings_including_nul_and_empty() {
+    let z = zset(&[(b"a\0b", 1), (b"", 2), (b"a", 3)]);
+    assert_eq!(
+        (weight(z, b"a\0b"), weight(z, b""), weight(z, b"a"), len(z)),
+        (1, 2, 3, 3)
+    );
+    assert_eq!(unsafe { zs_zset_insert(z, null(), 0, 4) }, OK);
+    assert_eq!(weight(z, b""), 6);
+    unsafe { zs_zset_free(z) };
+}
+
+#[test]
+fn distinct_clone_and_encoding_round_trip() {
+    let z = zset(&[(b"k1", 5), (b"k2", -2), (b"k3", 1)]);
+    let mut d = null_mut();
+    assert_eq!(unsafe { zs_zset_distinct(z, &mut d) }, OK);
+    assert_eq!(
+        (weight(d, b"k1"), weight(d, b"k2"), weight(d, b"k3")),
+        (1, 0, 1)
+    );
+    let mut copy = null_mut();
+    assert_eq!(unsafe { zs_zset_clone(z, &mut copy) }, OK);
+    let enc = encoded(z);
+    let mut back = null_mut();
+    assert_eq!(unsafe { zs_zset_new(&mut back) }, OK);
+    assert_eq!(
+        unsafe { zs_zset_decode_add(back, enc.as_ptr(), enc.len()) },
+        OK
+    );
+    assert_eq!(encoded(back), encoded(copy));
+    assert_eq!(encoded(back), enc);
+    for h in [z, d, copy, back] {
+        unsafe { zs_zset_free(h) };
+    }
+}
+
+#[test]
+fn a_small_buffer_reports_the_size_it_needs() {
+    let z = zset(&[(b"abc", 1)]);
+    let (mut needed, mut small) = (0, [0u8; 3]);
+    assert_eq!(
+        unsafe { zs_zset_encode(z, small.as_mut_ptr(), small.len(), &mut needed) },
+        Code::BufferTooSmall as i32
+    );
+    assert_eq!(needed, 4 + 3 + 8);
+    assert_eq!(small, [0; 3], "nothing written on failure");
+    unsafe { zs_zset_free(z) };
+}
+
+#[test]
+fn a_bad_encoding_is_refused_and_adds_nothing() {
+    let z = zset(&[(b"a", 1)]);
+    let mut enc = encoded(z);
+    enc.push(7);
+    assert_eq!(
+        unsafe { zs_zset_decode_add(z, enc.as_ptr(), enc.len()) },
+        Code::BadValue as i32
+    );
+    assert_eq!((weight(z, b"a"), len(z)), (1, 1));
+    unsafe { zs_zset_free(z) };
+}
+
+#[test]
+fn overflow_is_an_error_code_and_leaves_the_set_unchanged() {
+    let (a, b) = (
+        zset(&[(b"k", i64::MAX), (b"j", 1)]),
+        zset(&[(b"j", 1), (b"k", 1)]),
+    );
+    assert_eq!(unsafe { zs_zset_add(a, b) }, Code::Overflow as i32);
+    assert_eq!(
+        (weight(a, b"k"), weight(a, b"j")),
+        (i64::MAX, 1),
+        "a is unchanged, even for the key before the overflow"
+    );
+    assert!(last_error().contains("weight overflow"), "{}", last_error());
+    assert_eq!(unsafe { zs_zset_scale(a, 2) }, Code::Overflow as i32);
+    assert_eq!(
+        unsafe { zs_zset_insert(a, b"k".as_ptr(), 1, 1) },
+        Code::Overflow as i32
+    );
+    let m = zset(&[(b"k", i64::MIN)]);
+    assert_eq!(unsafe { zs_zset_neg(m) }, Code::Overflow as i32);
+    assert_eq!(weight(m, b"k"), i64::MIN);
+    assert_eq!(unsafe { zs_zset_sub(a, a) }, OK);
+    for z in [a, b, m] {
+        unsafe { zs_zset_free(z) };
+    }
+}
+
+#[test]
+fn a_null_argument_is_a_code_not_a_crash() {
+    assert_eq!(
+        unsafe { zs_zset_new(null_mut()) },
+        Code::NullArgument as i32
+    );
+    assert_eq!(
+        unsafe { zs_zset_len(null(), &mut 0) },
+        Code::NullArgument as i32
+    );
+    let z = zset(&[]);
+    assert_eq!(
+        unsafe { zs_zset_insert(z, null(), 3, 1) },
+        Code::NullArgument as i32
+    );
+    assert_eq!(unsafe { zs_zset_add(z, null()) }, Code::NullArgument as i32);
+    assert_eq!(
+        unsafe { zs_circuit_filter(null_mut(), 0, None, null_mut(), &mut 0) },
+        Code::NullArgument as i32
+    );
+    unsafe { zs_zset_free(z) };
+    assert_ne!(zs_abi_version(), 0);
+}
+
+#[test]
+fn a_circuit_runs_distinct_and_integrate() {
+    let c = C::new();
+    let i = c.input();
+    let (mut d, mut int) = (0, 0);
+    assert_eq!(unsafe { zs_circuit_distinct(c.0, i, &mut d) }, OK);
+    assert_eq!(unsafe { zs_circuit_integrate(c.0, d, &mut int) }, OK);
+    let (sd, sint) = (c.sink(d), c.sink(int));
+    c.push(i, &[(b"a", 3), (b"b", -1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    assert_eq!(c.take(sd), one(&[(b"a", 1)]));
+    assert_eq!(c.take(sint), one(&[(b"a", 1)]));
+    c.push(i, &[(b"a", -3), (b"b", 1)]);
+    assert_eq!(c.run(sd), one(&[(b"a", -1)]));
+}
+
+extern "C" fn upper(
+    _: *mut c_void,
+    k: *const u8,
+    n: usize,
+    emit: engine::Emit,
+    e: *mut c_void,
+) -> i32 {
+    let up: Vec<u8> = unsafe { std::slice::from_raw_parts(k, n) }.to_ascii_uppercase();
+    emit(e, up.as_ptr(), up.len());
+    emit(e, b"!".as_ptr(), 1);
+    0
+}
+
+extern "C" fn even_len(_: *mut c_void, _: *const u8, n: usize) -> i32 {
+    i32::from(n.is_multiple_of(2))
+}
+
+extern "C" fn first_byte(
+    _: *mut c_void,
+    k: *const u8,
+    n: usize,
+    emit: engine::Emit,
+    e: *mut c_void,
+) -> i32 {
+    emit(e, k, n.min(1));
+    0
+}
+
+#[test]
+fn callbacks_filter_and_flat_map() {
+    let c = C::new();
+    let i = c.input();
+    let (mut f, mut m) = (0, 0);
+    assert_eq!(
+        unsafe { zs_circuit_filter(c.0, i, Some(even_len), null_mut(), &mut f) },
+        OK
+    );
+    assert_eq!(
+        unsafe { zs_circuit_flat_map(c.0, f, Some(upper), null_mut(), &mut m) },
+        OK
+    );
+    let s = c.sink(m);
+    c.push(i, &[(b"ab", 2), (b"abc", 5), (b"cd", -1)]);
+    assert_eq!(c.run(s), one(&[(b"!", 1), (b"AB", 2), (b"CD", -1)]));
+}
+
+#[test]
+fn index_by_join_and_aggregate_follow_the_pair_encoding() {
+    let c = C::new();
+    let (orders, prices) = (c.input(), c.input());
+    let (mut by_item, mut joined) = (0, 0);
+    assert_eq!(
+        unsafe { zs_circuit_index_by(c.0, orders, Some(first_byte), null_mut(), &mut by_item) },
+        OK
+    );
+    assert_eq!(
+        unsafe { zs_circuit_join(c.0, by_item, prices, &mut joined) },
+        OK
+    );
+    let sink = c.sink(joined);
+    c.push(orders, &[(b"apple", 1)]);
+    c.push(prices, &[(&pair(b"a", b"3"), 1)]);
+    let want = pair(b"a", &pair(b"apple", b"3"));
+    assert_eq!(c.run(sink), one(&[(&want, 1)]));
+    c.push(prices, &[(&pair(b"a", b"3"), -1)]);
+    assert_eq!(c.run(sink), one(&[(&want, -1)]));
+}
+
+#[test]
+fn aggregates_count_sum_min_and_max() {
+    let c = C::new();
+    let i = c.input();
+    let outs: Vec<u32> = (0..4)
+        .map(|kind| {
+            let mut o = 0;
+            assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, kind, &mut o) }, OK);
+            c.sink(o)
+        })
+        .collect();
+    c.push(
+        i,
+        &[
+            (&pair(b"g", &le(5)), 1),
+            (&pair(b"g", &le(2)), 2),
+            (&pair(b"h", &le(9)), 1),
+        ],
+    );
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    let val = |k: &[u8], n: i64| (pair(k, &le(n)), 1);
+    let want = [
+        [val(b"g", 3), val(b"h", 1)],
+        [val(b"g", 9), val(b"h", 9)],
+        [val(b"g", 2), val(b"h", 9)],
+        [val(b"g", 5), val(b"h", 9)],
+    ];
+    for (sink, want) in outs.iter().zip(want) {
+        let got = c.take(*sink);
+        let want: Vec<(Vec<u8>, i64)> = want.into_iter().collect();
+        let mut sorted = want.clone();
+        sorted.sort();
+        assert_eq!(got, sorted, "kind of sink {sink}");
+    }
+}
+
+#[test]
+fn a_weight_overflow_in_a_step_is_a_code_and_poisons_the_circuit() {
+    let c = C::new();
+    let i = c.input();
+    let (mut int, mut again) = (0, 0);
+    assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
+    c.push(i, &[(b"k", i64::MAX)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    c.push(i, &[(b"k", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
+    assert!(last_error().contains("weight overflow"));
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
+    assert_eq!(
+        unsafe { zs_circuit_distinct(c.0, i, &mut again) },
+        Code::Poisoned as i32
+    );
+}
+
+#[test]
+fn a_malformed_pair_and_an_unknown_id_are_codes() {
+    let c = C::new();
+    let i = c.input();
+    let mut j = 0;
+    assert_eq!(
+        unsafe { zs_circuit_join(c.0, i, 99, &mut j) },
+        Code::BadHandle as i32
+    );
+    assert_eq!(unsafe { zs_circuit_join(c.0, i, i, &mut j) }, OK);
+    c.push(i, &[(b"\xff\xff\xff\xff", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::BadValue as i32);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
+    let mut a = 0;
+    let c2 = C::new();
+    assert_eq!(
+        unsafe { zs_circuit_aggregate(c2.0, 0, 9, &mut a) },
+        Code::BadValue as i32
+    );
+    let mut z = null_mut();
+    assert_eq!(
+        unsafe { zs_circuit_take(c2.0, 5, &mut z) },
+        Code::BadHandle as i32
+    );
+    let empty = zset(&[]);
+    assert_eq!(
+        unsafe { zs_circuit_push(c2.0, 0, empty) },
+        Code::BadHandle as i32
+    );
+    unsafe { zs_zset_free(empty) };
+}
+
+extern "C" fn out_of_time(ctx: *mut c_void) -> i32 {
+    let calls = unsafe { &mut *(ctx as *mut u32) };
+    *calls += 1;
+    1
+}
+
+#[test]
+fn a_rebuild_resumes_across_calls_and_matches_one_step() {
+    let keys: Vec<Vec<u8>> = (0u8..9).map(|n| vec![b'k', n]).collect();
+    let entries: Vec<(&[u8], i64)> = keys.iter().map(|k| (k.as_slice(), 1)).collect();
+    let snapshot = zset(&entries);
+
+    let whole = C::new();
+    let wi = whole.input();
+    let mut wd = 0;
+    assert_eq!(unsafe { zs_circuit_distinct(whole.0, wi, &mut wd) }, OK);
+    let ws = whole.sink(wd);
+    assert_eq!(unsafe { zs_circuit_push(whole.0, wi, snapshot) }, OK);
+    let want = whole.run(ws);
+
+    let c = C::new();
+    let i = c.input();
+    let mut d = 0;
+    assert_eq!(unsafe { zs_circuit_distinct(c.0, i, &mut d) }, OK);
+    let s = c.sink(d);
+    let mut job = null_mut();
+    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 2, &mut job) }, OK);
+    let (mut calls, mut got, mut done) = (0u32, Vec::new(), 0);
+    for _ in 0..20 {
+        assert_eq!(
+            unsafe {
+                zs_rebuild_step(
+                    job,
+                    c.0,
+                    Some(out_of_time),
+                    (&mut calls as *mut u32).cast(),
+                    &mut done,
+                )
+            },
+            OK
+        );
+        got.extend(c.take(s));
+        if done == 1 {
+            break;
+        }
+    }
+    assert_eq!(done, 1, "the job finished within its bound");
+    assert_eq!(
+        calls, 4,
+        "ceil(9 / 2) = 5 chunks; the last is Done before any check"
+    );
+    got.sort();
+    assert_eq!(got, want);
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+#[test]
+fn a_rebuild_without_a_deadline_runs_to_completion() {
+    let snapshot = zset(&[(b"a", 1), (b"b", 1), (b"c", 1)]);
+    let c = C::new();
+    let i = c.input();
+    let s = c.sink(i);
+    let mut job = null_mut();
+    let mut done = 0;
+    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
+        OK
+    );
+    assert_eq!(done, 1);
+    assert_eq!(c.take(s).len(), 3);
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+#[test]
+fn min_and_max_read_only_values_of_positive_weight() {
+    let c = C::new();
+    let i = c.input();
+    let (mut lo, mut hi) = (0, 0);
+    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 2, &mut lo) }, OK);
+    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 3, &mut hi) }, OK);
+    let (slo, shi) = (c.sink(lo), c.sink(hi));
+    c.push(
+        i,
+        &[
+            (&pair(b"g", &le(5)), 1),
+            (&pair(b"g", &le(1)), -1),
+            (&pair(b"g", &le(9)), -1),
+        ],
+    );
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    assert_eq!(c.take(slo), one(&[(&pair(b"g", &le(5)), 1)]));
+    assert_eq!(c.take(shi), one(&[(&pair(b"g", &le(5)), 1)]));
+}
+
+#[test]
+fn an_index_by_callback_that_emits_nothing_is_a_bad_value() {
+    extern "C" fn nothing(
+        _: *mut c_void,
+        _: *const u8,
+        _: usize,
+        _: engine::Emit,
+        _: *mut c_void,
+    ) -> i32 {
+        0
+    }
+    let c = C::new();
+    let i = c.input();
+    let mut o = 0;
+    assert_eq!(
+        unsafe { zs_circuit_index_by(c.0, i, Some(nothing), null_mut(), &mut o) },
+        OK
+    );
+    c.push(i, &[(b"x", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::BadValue as i32);
+    assert!(last_error().contains("exactly one key"), "{}", last_error());
+}
+
+#[test]
+fn a_chunked_rebuild_through_delay_is_one_tick() {
+    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
+    let c = C::new();
+    let i = c.input();
+    let mut d = 0;
+    assert_eq!(unsafe { zs_circuit_delay(c.0, i, &mut d) }, OK);
+    let s = c.sink(d);
+    c.push(i, &[(b"p", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    c.take(s);
+    let mut job = null_mut();
+    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
+    let (mut calls, mut done) = (0u32, 0);
+    let ctx = (&mut calls as *mut u32).cast();
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, Some(out_of_time), ctx, &mut done) },
+        OK
+    );
+    assert_eq!(done, 0);
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, Some(out_of_time), ctx, &mut done) },
+        OK
+    );
+    assert_eq!(
+        done, 1,
+        "the last chunk is Done though the deadline expired"
+    );
+    assert_eq!(c.take(s), one(&[(b"p", 1)]));
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+#[test]
+fn a_set_added_to_or_subtracted_from_itself_is_defined() {
+    let a = zset(&[(b"x", 2), (b"y", -3)]);
+    assert_eq!(unsafe { zs_zset_add(a, a) }, OK);
+    assert_eq!((weight(a, b"x"), weight(a, b"y")), (4, -6));
+    assert_eq!(unsafe { zs_zset_sub(a, a) }, OK);
+    assert_eq!(len(a), 0);
+    let m = zset(&[(b"k", i64::MAX)]);
+    assert_eq!(unsafe { zs_zset_add(m, m) }, Code::Overflow as i32);
+    assert_eq!(
+        weight(m, b"k"),
+        i64::MAX,
+        "the failed self-add changed nothing"
+    );
+    unsafe { zs_zset_free(a) };
+    unsafe { zs_zset_free(m) };
+}
+
+#[test]
+fn a_null_buffer_is_refused_before_it_is_written() {
+    let z = zset(&[(b"abc", 1)]);
+    let mut needed = 0;
+    assert_eq!(
+        unsafe { zs_zset_encode(z, null_mut(), 100, &mut needed) },
+        Code::NullArgument as i32
+    );
+    assert_eq!(
+        unsafe { zs_zset_encode(z, null_mut(), 0, &mut needed) },
+        Code::BufferTooSmall as i32,
+        "a null buffer still probes the size"
+    );
+    assert_eq!(needed, 15);
+    let empty = zset(&[]);
+    assert_eq!(
+        unsafe { zs_zset_encode(empty, null_mut(), 0, &mut needed) },
+        OK
+    );
+    unsafe { zs_zset_free(z) };
+    unsafe { zs_zset_free(empty) };
+}
+
+#[test]
+fn a_null_out_pointer_is_refused_before_any_state_changes() {
+    let c = C::new();
+    let i = c.input();
+    assert_eq!(
+        unsafe { zs_circuit_input(c.0, null_mut()) },
+        Code::NullArgument as i32
+    );
+    assert_eq!(c.input(), i + 1, "the refused call added no input");
+    let s = c.sink(i);
+    c.push(i, &[(b"k", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    assert_eq!(
+        unsafe { zs_circuit_take(c.0, s, null_mut()) },
+        Code::NullArgument as i32
+    );
+    assert_eq!(
+        c.take(s),
+        one(&[(b"k", 1)]),
+        "the refused take left the sink alone"
+    );
+    assert_eq!(
+        unsafe { zs_zset_new(null_mut()) },
+        Code::NullArgument as i32
+    );
+    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
+    let mut job = null_mut();
+    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), null_mut()) },
+        Code::NullArgument as i32
+    );
+    let mut done = 0;
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
+        OK
+    );
+    assert_eq!(c.take(s).len(), 2, "the refused step consumed nothing");
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+#[test]
+fn a_rebuild_is_stepped_only_on_its_own_circuit() {
+    let (c1, c2) = (C::new(), C::new());
+    let (i1, i2) = (c1.input(), c2.input());
+    let (s1, s2) = (c1.sink(i1), c2.sink(i2));
+    let snapshot = zset(&[(b"a", 1)]);
+    let mut job = null_mut();
+    assert_eq!(
+        unsafe { zs_rebuild_new(c1.0, i1, snapshot, 1, &mut job) },
+        OK
+    );
+    let mut done = 0;
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c2.0, None, null_mut(), &mut done) },
+        Code::BadHandle as i32
+    );
+    assert_eq!((c2.take(s2).len(), c1.take(s1).len()), (0, 0));
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c1.0, None, null_mut(), &mut done) },
+        OK
+    );
+    assert_eq!(c1.take(s1), one(&[(b"a", 1)]));
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+#[test]
+fn stream_ids_and_sink_ids_are_not_interchangeable() {
+    let c = C::new();
+    let i = c.input();
+    let s = c.sink(i);
+    let mut z = null_mut();
+    assert_eq!(
+        unsafe { zs_circuit_take(c.0, i, &mut z) },
+        Code::BadHandle as i32
+    );
+    let mut o = 0;
+    assert_eq!(
+        unsafe { zs_circuit_delay(c.0, s, &mut o) },
+        Code::BadHandle as i32
+    );
+    assert_eq!(
+        unsafe { zs_circuit_output(c.0, s, &mut o) },
+        Code::BadHandle as i32
+    );
+    let empty = zset(&[]);
+    assert_eq!(
+        unsafe { zs_circuit_push(c.0, s, empty) },
+        Code::BadHandle as i32
+    );
+    unsafe { zs_zset_free(empty) };
+}
+
+extern "C" fn fails(_: *mut c_void, _: *const u8, _: usize) -> i32 {
+    -1
+}
+
+extern "C" fn flat_fails(
+    _: *mut c_void,
+    _: *const u8,
+    _: usize,
+    _: engine::Emit,
+    _: *mut c_void,
+) -> i32 {
+    7
+}
+
+extern "C" fn expire_fails(_: *mut c_void) -> i32 {
+    -1
+}
+
+#[test]
+fn a_callback_that_reports_failure_fails_the_step_and_kills_the_circuit() {
+    let c = C::new();
+    let i = c.input();
+    let (mut f, mut m) = (0, 0);
+    assert_eq!(
+        unsafe { zs_circuit_filter(c.0, i, Some(fails), null_mut(), &mut f) },
+        OK
+    );
+    c.push(i, &[(b"k", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
+    let d = C::new();
+    let j = d.input();
+    assert_eq!(
+        unsafe { zs_circuit_flat_map(d.0, j, Some(flat_fails), null_mut(), &mut m) },
+        OK
+    );
+    d.push(j, &[(b"k", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(d.0) }, Code::Callback as i32);
+    let e = C::new();
+    let k = e.input();
+    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
+    let mut job = null_mut();
+    assert_eq!(unsafe { zs_rebuild_new(e.0, k, snapshot, 1, &mut job) }, OK);
+    let mut done = 0;
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, e.0, Some(expire_fails), null_mut(), &mut done) },
+        Code::Callback as i32
+    );
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+extern "C" fn reenters(ctx: *mut c_void, _: *const u8, _: usize) -> i32 {
+    let (c, seen) = unsafe { &mut *(ctx as *mut (*mut ZsCircuit, i32)) };
+    let mut id = 0;
+    *seen = unsafe { zs_circuit_input(*c, &mut id) };
+    1
+}
+
+#[test]
+fn a_callback_that_calls_back_into_its_circuit_gets_poisoned_and_changes_nothing() {
+    let c = C::new();
+    let i = c.input();
+    let mut state: (*mut ZsCircuit, i32) = (c.0, 0);
+    let mut f = 0;
+    let ctx = (&mut state as *mut (*mut ZsCircuit, i32)).cast();
+    assert_eq!(
+        unsafe { zs_circuit_filter(c.0, i, Some(reenters), ctx, &mut f) },
+        OK
+    );
+    c.push(i, &[(b"k", 1)]);
+    assert_eq!(
+        unsafe { zs_circuit_step(c.0) },
+        OK,
+        "the outer step completes"
+    );
+    assert_eq!(state.1, Code::Poisoned as i32);
+    assert_eq!(c.input(), f + 1, "the re-entrant call added no input");
+}
+
+#[test]
+fn a_count_or_sum_that_cancels_to_zero_keeps_its_row() {
+    let c = C::new();
+    let i = c.input();
+    let (mut n, mut t) = (0, 0);
+    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 0, &mut n) }, OK);
+    assert_eq!(unsafe { zs_circuit_aggregate(c.0, i, 1, &mut t) }, OK);
+    let (sn, st) = (c.sink(n), c.sink(t));
+    c.push(
+        i,
+        &[
+            (&pair(b"g", &le(5)), 1),
+            (&pair(b"g", &le(-5)), 1),
+            (&pair(b"h", &le(1)), -1),
+            (&pair(b"h", &le(2)), 1),
+        ],
+    );
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    assert_eq!(
+        c.take(st),
+        one(&[(&pair(b"g", &le(0)), 1), (&pair(b"h", &le(1)), 1)])
+    );
+    assert_eq!(
+        c.take(sn),
+        one(&[(&pair(b"g", &le(2)), 1), (&pair(b"h", &le(0)), 1)])
+    );
+}
+
+#[cfg(target_pointer_width = "64")]
+#[test]
+fn a_length_of_4_gib_or_more_is_refused_not_truncated() {
+    assert_eq!(engine::len32(u32::MAX as usize), Ok(u32::MAX));
+    assert_eq!(
+        engine::len32(1usize << 32).unwrap_err().code,
+        Code::BadValue
+    );
+}
+
+#[test]
+fn an_index_by_callback_that_fails_reports_the_callback_code_not_a_later_fault() {
+    extern "C" fn fails_after_emitting(
+        _: *mut c_void,
+        _: *const u8,
+        _: usize,
+        _: engine::Emit,
+        _: *mut c_void,
+    ) -> i32 {
+        3
+    }
+    let c = C::new();
+    let i = c.input();
+    let mut o = 0;
+    assert_eq!(
+        unsafe { zs_circuit_index_by(c.0, i, Some(fails_after_emitting), null_mut(), &mut o) },
+        OK
+    );
+    c.push(i, &[(b"x", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
+}
+
+struct Reenter {
+    c: *mut ZsCircuit,
+    job: *mut ZsRebuild,
+    seen: Vec<i32>,
+}
+
+extern "C" fn expired_reenters(ctx: *mut c_void) -> i32 {
+    let r = unsafe { &mut *(ctx as *mut Reenter) };
+    let mut done = 0;
+    r.seen
+        .push(unsafe { zs_rebuild_step(r.job, r.c, None, null_mut(), &mut done) });
+    unsafe { zs_rebuild_free(r.job) };
+    unsafe { zs_circuit_free(r.c) };
+    1
+}
+
+#[test]
+fn a_callback_cannot_step_or_free_the_job_or_circuit_it_runs_under() {
+    let snapshot = zset(&[(b"a", 1), (b"b", 1), (b"c", 1)]);
+    let c = C::new();
+    let i = c.input();
+    let s = c.sink(i);
+    let mut job = null_mut();
+    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
+    let mut r = Reenter {
+        c: c.0,
+        job,
+        seen: Vec::new(),
+    };
+    let mut done = 0;
+    let ctx = (&mut r as *mut Reenter).cast();
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, Some(expired_reenters), ctx, &mut done) },
+        OK
+    );
+    assert_eq!(
+        r.seen,
+        vec![Code::Poisoned as i32],
+        "stepping from inside is refused"
+    );
+    assert_eq!(done, 0);
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
+        OK
+    );
+    assert_eq!(
+        (done, c.take(s).len()),
+        (1, 3),
+        "the job and circuit survived the frees"
+    );
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+#[test]
+fn a_circuit_freed_and_another_made_does_not_inherit_its_jobs() {
+    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
+    let c1 = C::new();
+    let i1 = c1.input();
+    let mut job = null_mut();
+    assert_eq!(
+        unsafe { zs_rebuild_new(c1.0, i1, snapshot, 1, &mut job) },
+        OK
+    );
+    drop(c1);
+    for _ in 0..64 {
+        let c2 = C::new();
+        let _ = c2.input();
+        let mut done = 0;
+        assert_eq!(
+            unsafe { zs_rebuild_step(job, c2.0, None, null_mut(), &mut done) },
+            Code::BadHandle as i32,
+            "a new circuit, even at the old address, is not the job's"
+        );
+    }
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}
+
+#[test]
+fn a_host_step_between_rebuild_calls_is_interrupted_not_a_poison() {
+    let snapshot = zset(&[(b"a", 1), (b"b", 1)]);
+    let c = C::new();
+    let i = c.input();
+    let s = c.sink(i);
+    let mut job = null_mut();
+    assert_eq!(unsafe { zs_rebuild_new(c.0, i, snapshot, 1, &mut job) }, OK);
+    let (mut calls, mut done) = (0u32, 0);
+    let ctx = (&mut calls as *mut u32).cast();
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, Some(out_of_time), ctx, &mut done) },
+        OK
+    );
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    assert_eq!(
+        unsafe { zs_rebuild_step(job, c.0, None, null_mut(), &mut done) },
+        Code::Interrupted as i32
+    );
+    assert!(last_error().contains("interrupted"), "{}", last_error());
+    assert_eq!(
+        unsafe { zs_circuit_step(c.0) },
+        OK,
+        "the circuit is still usable"
+    );
+    let _ = c.take(s);
+    unsafe { zs_rebuild_free(job) };
+    unsafe { zs_zset_free(snapshot) };
+}

crates/zset-ffi/tests/dylib.rs

@@ -0,0 +1,63 @@
+//! The built shared library, loaded the way a CFFI caller loads it.
+//! Evidence: `context/test-evidence/zset-ffi.md`.
+
+use libloading::{Library, Symbol};
+use std::path::PathBuf;
+use std::ptr::null_mut;
+
+type Opaque = std::ffi::c_void;
+
+/// Cargo puts the cdylib in `target/<profile>/deps` beside this test binary's own directory.
+fn library() -> PathBuf {
+    let exe = std::env::current_exe().unwrap();
+    let deps = exe.parent().unwrap();
+    let ext = if cfg!(target_os = "macos") {
+        "dylib"
+    } else {
+        "so"
+    };
+    let name = format!("{}zset_ffi.{ext}", if cfg!(windows) { "" } else { "lib" });
+    [deps.join(&name), deps.parent().unwrap().join(&name)]
+        .into_iter()
+        .find(|p| p.exists())
+        .unwrap_or_else(|| panic!("{name} was not built next to {}", deps.display()))
+}
+
+#[test]
+fn a_loaded_library_adds_queries_frees_and_reports_overflow() {
+    // SAFETY: loads this workspace's own library; the symbols below have the header's types.
+    unsafe {
+        let lib = Library::new(library()).unwrap();
+        let version: Symbol<extern "C" fn() -> u32> = lib.get(b"zs_abi_version").unwrap();
+        let new: Symbol<extern "C" fn(*mut *mut Opaque) -> i32> = lib.get(b"zs_zset_new").unwrap();
+        let free: Symbol<extern "C" fn(*mut Opaque)> = lib.get(b"zs_zset_free").unwrap();
+        let insert: Symbol<extern "C" fn(*mut Opaque, *const u8, usize, i64) -> i32> =
+            lib.get(b"zs_zset_insert").unwrap();
+        let weight: Symbol<extern "C" fn(*const Opaque, *const u8, usize, *mut i64) -> i32> =
+            lib.get(b"zs_zset_weight").unwrap();
+        let last: Symbol<extern "C" fn() -> *const std::ffi::c_char> =
+            lib.get(b"zs_last_error").unwrap();
+        assert_eq!(version(), 1);
+        let mut z = null_mut();
+        assert_eq!(new(&mut z), 0);
+        assert_eq!(insert(z, b"k".as_ptr(), 1, 40), 0);
+        assert_eq!(insert(z, b"k".as_ptr(), 1, 2), 0);
+        let mut w = 0;
+        assert_eq!(weight(z, b"k".as_ptr(), 1, &mut w), 0);
+        assert_eq!(w, 42);
+        assert_eq!(
+            insert(z, b"k".as_ptr(), 1, i64::MAX),
+            2,
+            "overflow is ZS_ERR_OVERFLOW"
+        );
+        assert!(
+            std::ffi::CStr::from_ptr(last())
+                .to_str()
+                .unwrap()
+                .contains("overflow")
+        );
+        assert_eq!(weight(z, b"k".as_ptr(), 1, &mut w), 0);
+        assert_eq!(w, 42, "the failed insert changed nothing");
+        free(z);
+    }
+}

crates/zset-ffi/tests/header.rs

@@ -0,0 +1,278 @@
+//! The checked-in C header matches the exported functions, their parameter and return
+//! types, and the constants. Evidence: `context/test-evidence/zset-ffi.md`.
+
+use std::collections::BTreeMap;
+use zset_ffi::ZS_ABI_VERSION;
+use zset_ffi::engine::Code;
+
+const SRC: &str = include_str!("../src/lib.rs");
+const HEADER: &str = include_str!("../include/zset_ffi.h");
+
+fn squash(s: &str) -> String {
+    s.split_whitespace().collect()
+}
+
+/// Splits at the commas that are not inside brackets.
+fn split_params(params: &str) -> Vec<String> {
+    let (mut depth, mut cur, mut out) = (0i32, String::new(), Vec::new());
+    for c in params.chars() {
+        match c {
+            '(' | '<' => depth += 1,
+            ')' | '>' => depth -= 1,
+            ',' if depth == 0 => {
+                out.push(std::mem::take(&mut cur));
+                continue;
+            }
+            _ => {}
+        }
+        cur.push(c);
+    }
+    if !cur.trim().is_empty() {
+        out.push(cur);
+    }
+    out
+}
+
+/// The C type a Rust ABI type must be declared as. Anything unlisted fails the test, so a
+/// new type needs a line here and a conscious choice.
+fn c_type(rust: &str) -> String {
+    let t = squash(rust);
+    let c = match t.as_str() {
+        "*mut*mutZsZSet" => "ZsZSet **",
+        "*mutZsZSet" => "ZsZSet *",
+        "*constZsZSet" => "const ZsZSet *",
+        "*mutZsCircuit" => "ZsCircuit *",
+        "*mut*mutZsCircuit" => "ZsCircuit **",
+        "*mutZsRebuild" => "ZsRebuild *",
+        "*mut*mutZsRebuild" => "ZsRebuild **",
+        "*constu8" => "const uint8_t *",
+        "*mutu8" => "uint8_t *",
+        "*mutc_void" => "void *",
+        "*mutu32" => "uint32_t *",
+        "*muti64" => "int64_t *",
+        "*mutusize" => "size_t *",
+        "*muti32" => "int32_t *",
+        "*constc_char" => "const char *",
+        "usize" => "size_t",
+        "i64" => "int64_t",
+        "u32" => "uint32_t",
+        "i32" => "int32_t",
+        "Option<KeepFn>" => "ZsKeepFn",
+        "Option<FlatMapFn>" => "ZsFlatMapFn",
+        "Option<ExpiredFn>" => "ZsExpiredFn",
+        "Emit" => "ZsEmit",
+        other => panic!("no C type is declared for the Rust type `{other}`"),
+    };
+    squash(c)
+}
+
+/// `name -> (return type, parameter types)` from the Rust source.
+fn exported() -> BTreeMap<String, (String, Vec<String>)> {
+    let mut out = BTreeMap::new();
+    let fns = SRC
+        .split("extern \"C\" fn ")
+        .skip(1)
+        .filter(|p| !p.starts_with('$'));
+    for part in fns {
+        let name = part.split('(').next().unwrap().to_string();
+        let rest = &part[name.len() + 1..];
+        let mut depth = 1;
+        let end = rest
+            .char_indices()
+            .find(|(_, c)| {
+                depth += i32::from(*c == '(') - i32::from(*c == ')');
+                depth == 0
+            })
+            .unwrap()
+            .0;
+        let params: Vec<String> = split_params(&rest[..end])
+            .iter()
+            .map(|p| c_type(p.split_once(':').unwrap().1))
+            .collect();
+        let ret = rest[end + 1..].split('{').next().unwrap().trim();
+        let ret = match ret.strip_prefix("->") {
+            Some(r) => c_type(r),
+            None => "void".to_string(),
+        };
+        out.insert(name, (ret, params));
+    }
+    // The `unary_op!` expansions share the macro body's signature.
+    let body = &SRC[SRC.find("macro_rules! unary_op").unwrap()..];
+    let sig = body.split("extern \"C\" fn $name(").nth(1).unwrap();
+    let (params, tail) = sig.split_once(')').unwrap();
+    let unary_params: Vec<String> = split_params(params)
+        .iter()
+        .map(|p| c_type(p.split_once(':').unwrap().1))
+        .collect();
+    let unary_ret = c_type(
+        tail.split('{')
+            .next()
+            .unwrap()
+            .trim()
+            .trim_start_matches("->"),
+    );
+    for line in SRC.lines().filter(|l| l.starts_with("unary_op!(")) {
+        let name = line["unary_op!(".len()..].split(',').next().unwrap();
+        out.insert(name.to_string(), (unary_ret.clone(), unary_params.clone()));
+    }
+    out
+}
+
+fn strip_comments(text: &str) -> String {
+    let mut flat = text.replace('\n', " ");
+    while let Some(open) = flat.find("/*") {
+        let close = flat[open..].find("*/").expect("an unterminated comment") + open + 2;
+        flat.replace_range(open..close, " ");
+    }
+    flat
+}
+
+/// A header parameter without its name: `const ZsZSet *z` is `const ZsZSet *`.
+fn param_type(p: &str) -> String {
+    let p = p.trim();
+    let name_start = p
+        .rfind(|c: char| !(c.is_alphanumeric() || c == '_'))
+        .map_or(0, |i| i + 1);
+    if name_start == 0 {
+        return squash(p);
+    }
+    squash(&p[..name_start])
+}
+
+fn declared() -> BTreeMap<String, (String, Vec<String>)> {
+    let mut out = BTreeMap::new();
+    for decl in strip_comments(HEADER).split(';') {
+        let Some(open) = decl.find('(') else { continue };
+        let before = &decl[..open];
+        let name_at = before
+            .rfind(|c: char| !(c.is_alphanumeric() || c == '_'))
+            .map_or(0, |i| i + 1);
+        let name = &before[name_at..];
+        if !name.starts_with("zs_") || before.contains("typedef") || before.contains("(*") {
+            continue;
+        }
+        let ret = squash(&before[..name_at]);
+        let params = decl[open + 1..].rsplit_once(')').unwrap().0;
+        let params = if params.trim() == "void" {
+            Vec::new()
+        } else {
+            split_params(params).iter().map(|p| param_type(p)).collect()
+        };
+        out.insert(name.to_string(), (ret, params));
+    }
+    out
+}
+
+fn defines() -> BTreeMap<String, i64> {
+    HEADER
+        .lines()
+        .filter_map(|l| l.strip_prefix("#define "))
+        .filter_map(|l| {
+            let (k, v) = l.split_once(' ')?;
+            Some((k.to_string(), v.trim().parse().ok()?))
+        })
+        .collect()
+}
+
+#[test]
+fn the_header_declares_exactly_the_exported_functions_with_their_types() {
+    let (src, hdr) = (exported(), declared());
+    assert!(
+        src.len() > 30,
+        "the scanner saw {} functions, it is not reading the source",
+        src.len()
+    );
+    assert_eq!(src, hdr);
+}
+
+#[test]
+fn the_header_constants_are_the_rust_constants() {
+    let d = defines();
+    for code in Code::ALL {
+        assert_eq!(
+            d[code.header_name()],
+            *code as i64,
+            "{}",
+            code.header_name()
+        );
+    }
+    let err_lines = d.keys().filter(|k| k.starts_with("ZS_ERR_")).count();
+    assert_eq!(
+        err_lines,
+        Code::ALL.len(),
+        "a header code without a Rust variant"
+    );
+    assert_eq!(d["ZS_ABI_VERSION"], i64::from(ZS_ABI_VERSION));
+    assert_eq!(d["ZS_OK"], 0);
+    assert_eq!(
+        [
+            d["ZS_AGG_COUNT"],
+            d["ZS_AGG_SUM"],
+            d["ZS_AGG_MIN"],
+            d["ZS_AGG_MAX"]
+        ],
+        [0, 1, 2, 3]
+    );
+}
+
+/// `typedef R (*Name)(params)` from the header, and `pub type X = extern "C" fn(..) -> R` from
+/// `engine.rs`, as `name -> (return, params)`.
+type Sigs = BTreeMap<String, (String, Vec<String>)>;
+
+fn callback_types() -> (Sigs, Sigs) {
+    let mut hdr = BTreeMap::new();
+    for decl in strip_comments(HEADER).split(';') {
+        let Some(at) = decl.find("(*") else { continue };
+        if !decl.contains("typedef") {
+            continue;
+        }
+        let ret = squash(
+            decl.split("typedef")
+                .nth(1)
+                .unwrap()
+                .split('(')
+                .next()
+                .unwrap(),
+        );
+        let rest = &decl[at + 2..];
+        let (name, after) = rest.split_once(')').unwrap();
+        let params = after
+            .trim()
+            .trim_start_matches('(')
+            .rsplit_once(')')
+            .unwrap()
+            .0;
+        let params = split_params(params).iter().map(|p| param_type(p)).collect();
+        hdr.insert(name.trim().to_string(), (ret, params));
+    }
+    let engine = include_str!("../src/engine.rs");
+    let mut rust = BTreeMap::new();
+    for line in engine.lines().filter(|l| l.contains("= extern \"C\" fn(")) {
+        let name = line
+            .trim_start_matches("pub type ")
+            .split(' ')
+            .next()
+            .unwrap();
+        let sig = line.split("fn(").nth(1).unwrap();
+        let (params, tail) = sig.rsplit_once(')').unwrap();
+        let ret = match tail.trim().trim_end_matches(';').trim().strip_prefix("->") {
+            Some(r) => c_type(r),
+            None => "void".to_string(),
+        };
+        let params = split_params(params).iter().map(|p| c_type(p)).collect();
+        rust.insert(format!("Zs{name}"), (ret, params));
+    }
+    (hdr, rust)
+}
+
+#[test]
+fn the_header_callback_typedefs_match_the_rust_callback_types() {
+    let (hdr, rust) = callback_types();
+    assert_eq!(
+        rust.len(),
+        4,
+        "the scanner saw {} callback types",
+        rust.len()
+    );
+    assert_eq!(rust, hdr);
+}

crates/zset-ffi/tests/smoke.lisp

@@ -0,0 +1,69 @@
+;;; A CFFI smoke test: load libzset_ffi, add weights, query, hit the overflow code, free.
+;;; Run: sbcl --non-interactive --load ~/quicklisp/setup.lisp \
+;;;        --eval '(defvar cl-user::*zset-lib* "target/release/libzset_ffi.dylib")' \
+;;;        --load crates/zset-ffi/tests/smoke.lisp
+;;; Not a CI gate: sbcl and cffi are not in the dev shell. See context/code/zset-ffi.md.
+(ql:quickload :cffi :silent t)
+(defpackage :zset-smoke (:use :cl :cffi))
+(in-package :zset-smoke)
+
+(load-foreign-library cl-user::*zset-lib*)
+
+(defcfun "zs_abi_version" :uint32)
+(defcfun "zs_last_error" :string)
+(defcfun "zs_zset_new" :int32 (out :pointer))
+(defcfun "zs_zset_free" :void (z :pointer))
+(defcfun "zs_zset_insert" :int32 (z :pointer) (key :pointer) (len :size) (weight :int64))
+(defcfun "zs_zset_weight" :int32 (z :pointer) (key :pointer) (len :size) (out :pointer))
+(defcfun "zs_zset_len" :int32 (z :pointer) (out :pointer))
+
+(defconstant +overflow+ 2)
+
+(defun check (what got want)
+  (unless (equal got want) (error "~a: got ~s, wanted ~s" what got want))
+  (format t "ok ~a~%" what))
+
+(defun key-call (fn z key &rest more)
+  "Calls FN with Z, a foreign copy of the string KEY's UTF-8 bytes, its length, MORE."
+  (with-foreign-string ((ptr len) key :null-terminated-p nil)
+    (apply fn z ptr len more)))
+
+(with-foreign-object (out :pointer)
+  (check "new" (zs-zset-new out) 0)
+  (let ((z (mem-ref out :pointer)))
+    (check "insert" (key-call #'zs-zset-insert z "k" 40) 0)
+    (check "insert again" (key-call #'zs-zset-insert z "k" 2) 0)
+    (with-foreign-object (w :int64)
+      (check "weight" (progn (key-call #'zs-zset-weight z "k" w) (mem-ref w :int64)) 42)
+      (check "overflow code" (key-call #'zs-zset-insert z "k" (1- (expt 2 63))) +overflow+)
+      (check "error text" (and (search "overflow" (zs-last-error)) t) t)
+      (check "unchanged" (progn (key-call #'zs-zset-weight z "k" w) (mem-ref w :int64)) 42))
+    (zs-zset-free z)))
+
+;;; A callback: the body is wrapped in handler-case and returns the failure value, because a
+;;; non-local exit through the library would leave the circuit dead (ZS_ERR_POISONED).
+(defcfun "zs_circuit_new" :int32 (out :pointer))
+(defcfun "zs_circuit_free" :void (c :pointer))
+(defcfun "zs_circuit_input" :int32 (c :pointer) (out :pointer))
+(defcfun "zs_circuit_filter" :int32 (c :pointer) (stream :uint32) (keep :pointer) (ctx :pointer) (out :pointer))
+(defcfun "zs_circuit_push" :int32 (c :pointer) (input :uint32) (z :pointer))
+(defcfun "zs_circuit_step" :int32 (c :pointer))
+
+(defcallback keep-or-fail :int32 ((ctx :pointer) (key :pointer) (len :size))
+  (declare (ignore ctx key len))
+  (handler-case (error "a Lisp error inside a callback")
+    (error () -1)))
+
+(with-foreign-objects ((cout :pointer) (id :uint32) (zout :pointer))
+  (check "circuit" (zs-circuit-new cout) 0)
+  (let ((c (mem-ref cout :pointer)))
+    (check "input" (zs-circuit-input c id) 0)
+    (check "filter" (zs-circuit-filter c (mem-ref id :uint32) (callback keep-or-fail) (null-pointer) id) 0)
+    (check "zset" (zs-zset-new zout) 0)
+    (key-call #'zs-zset-insert (mem-ref zout :pointer) "k" 1)
+    (check "push" (zs-circuit-push c 0 (mem-ref zout :pointer)) 0)
+    (check "step reports the callback's failure" (zs-circuit-step c) 8)
+    (check "the circuit is dead afterwards" (zs-circuit-step c) 7)
+    (zs-zset-free (mem-ref zout :pointer))
+    (zs-circuit-free c)))
+(format t "zset smoke: all ok~%")