feat(zset-ffi): zs_circuit_begin, commit and rollback expose the circuit's transactions (improvise-bw4p.5)

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

crates/zset-ffi/include/zset_ffi.h

@@ -47,6 +47,7 @@ extern "C" {
 #define ZS_ERR_POISONED 7
 #define ZS_ERR_CALLBACK 8
 #define ZS_ERR_INTERRUPTED 9
+#define ZS_ERR_TRANSACTION 10
 
 #define ZS_AGG_COUNT 0
 #define ZS_AGG_SUM 1
@@ -131,8 +132,27 @@ int32_t zs_circuit_aggregate(ZsCircuit *c, uint32_t stream, int32_t kind, uint32
 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. */
+ * uses it returns ZS_ERR_POISONED, except the free, and zs_circuit_rollback of a transaction
+ * the failed step ran in. */
 int32_t zs_circuit_step(ZsCircuit *c);
+
+/* Transactions: undo a failed command. zs_circuit_begin opens one; zs_circuit_commit keeps
+ * what its steps did; zs_circuit_rollback puts the circuit's state back to what it was at
+ * begin, and revives a circuit that a failed step in the transaction poisoned. The cost of a
+ * transaction is the keys its steps touch (plus a delay's held input), not the circuit's size.
+ *  - At most one is open. begin on an open one, and commit or rollback on none, is
+ *    ZS_ERR_TRANSACTION. begin on a poisoned circuit is ZS_ERR_POISONED; so is commit after a
+ *    failed step (roll back instead). A poison from outside a transaction is permanent.
+ *  - Pushes made after begin are dropped by a rollback; what was pending at begin is pending
+ *    again. zs_circuit_take is ZS_ERR_TRANSACTION while one is open: take after commit.
+ *  - Operators built inside a transaction stay, and their state goes back to empty.
+ *  - What is outside the circuit is not undone: a callback's own side effects, and a
+ *    ZsRebuild stepped in the transaction (it reports ZS_ERR_INTERRUPTED afterwards; start
+ *    another job). */
+int32_t zs_circuit_begin(ZsCircuit *c);
+int32_t zs_circuit_commit(ZsCircuit *c);
+int32_t zs_circuit_rollback(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

crates/zset-ffi/src/engine.rs

@@ -40,6 +40,7 @@ codes! {
     Poisoned = 7 => "ZS_ERR_POISONED",
     Callback = 8 => "ZS_ERR_CALLBACK",
     Interrupted = 9 => "ZS_ERR_INTERRUPTED",
+    Transaction = 10 => "ZS_ERR_TRANSACTION",
 }
 
 #[derive(Clone, Debug, PartialEq, Eq)]
@@ -209,6 +210,8 @@ pub struct Engine {
     sinks: Vec<Output<Bytes>>,
     fault: Fault,
     poisoned: bool,
+    /// A transaction is open (`begin`, and not yet `commit` or `rollback`).
+    txn: bool,
 }
 
 impl Default for Engine {
@@ -226,6 +229,7 @@ impl Engine {
             sinks: Vec::new(),
             fault: Rc::default(),
             poisoned: false,
+            txn: false,
         }
     }
 
@@ -474,8 +478,54 @@ impl Engine {
         self.guarded(|c| c.step())
     }
 
+    fn no_transaction() -> Fail {
+        Fail::new(Code::Transaction, "no transaction is open")
+    }
+
+    /// Opens a transaction: the circuit logs what its steps change.
+    pub fn begin(&mut self) -> Res<()> {
+        self.alive()?;
+        if self.txn {
+            return Err(Fail::new(
+                Code::Transaction,
+                "a transaction is already open",
+            ));
+        }
+        self.circuit.begin();
+        self.txn = true;
+        Ok(())
+    }
+
+    /// Keeps the transaction's steps. A transaction whose step failed cannot be committed.
+    pub fn commit(&mut self) -> Res<()> {
+        if !self.txn {
+            return Err(Self::no_transaction());
+        }
+        self.alive()?;
+        self.circuit.commit();
+        self.txn = false;
+        Ok(())
+    }
+
+    /// Undoes the transaction's steps and clears the poison a failed step left in it.
+    pub fn rollback(&mut self) -> Res<()> {
+        if !self.txn {
+            return Err(Self::no_transaction());
+        }
+        self.circuit.rollback();
+        self.poisoned = false;
+        self.txn = false;
+        Ok(())
+    }
+
     pub fn take(&mut self, sink: u32) -> Res<BZ> {
         self.alive()?;
+        if self.txn {
+            return Err(Fail::new(
+                Code::Transaction,
+                "a transaction is open: commit or roll back before taking output",
+            ));
+        }
         self.sinks
             .get((sink & !SINK_BIT) as usize)
             .filter(|_| sink & SINK_BIT != 0)

crates/zset-ffi/src/lib.rs

@@ -499,6 +499,30 @@ pub unsafe extern "C" fn zs_circuit_step(c: *mut ZsCircuit) -> i32 {
     })
 }
 
+/// # Safety
+/// `c` is a live circuit or null.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_begin(c: *mut ZsCircuit) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { circuit(c)? }.engine().begin())
+}
+
+/// # Safety
+/// As `zs_circuit_begin`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_commit(c: *mut ZsCircuit) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { circuit(c)? }.engine().commit())
+}
+
+/// # Safety
+/// As `zs_circuit_begin`.
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn zs_circuit_rollback(c: *mut ZsCircuit) -> i32 {
+    // SAFETY: the caller's contract.
+    guard(|| unsafe { circuit(c)? }.engine().rollback())
+}
+
 /// # Safety
 /// `c` is a live circuit or null; `out` is null or writable.
 #[unsafe(no_mangle)]

crates/zset-ffi/tests/abi.rs

@@ -965,3 +965,94 @@ fn a_host_step_between_rebuild_calls_is_interrupted_not_a_poison() {
     unsafe { zs_rebuild_free(job) };
     unsafe { zs_zset_free(snapshot) };
 }
+
+/// An integrate over `k` held at `i64::MAX` (taken), and its sink.
+fn at_max() -> (C, u32, u32) {
+    let c = C::new();
+    let i = c.input();
+    let mut int = 0;
+    assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
+    let s = c.sink(int);
+    c.push(i, &[(b"k", i64::MAX)]);
+    assert_eq!(c.run(s), one(&[(b"k", i64::MAX)]));
+    (c, i, s)
+}
+
+#[test]
+fn a_rollback_after_a_failed_step_revives_the_circuit_in_its_old_state() {
+    let (c, i, s) = at_max();
+    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
+    c.push(i, &[(b"a", 1), (b"k", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
+    assert_eq!(
+        unsafe { zs_circuit_commit(c.0) },
+        Code::Poisoned as i32,
+        "a failed step cannot be committed"
+    );
+    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
+    c.push(i, &[(b"k", -1)]);
+    assert_eq!(
+        c.run(s),
+        one(&[(b"k", i64::MAX - 1)]),
+        "[rollback] neither `a` nor the failed +1 is in the integral"
+    );
+}
+
+#[test]
+fn a_committed_transaction_keeps_its_steps_and_a_new_one_may_follow() {
+    let (c, i, s) = at_max();
+    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
+    c.push(i, &[(b"k", -1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, OK);
+    assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
+    assert_eq!(c.take(s), one(&[(b"k", i64::MAX - 1)]));
+    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
+    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
+    c.push(i, &[(b"k", -1)]);
+    assert_eq!(c.run(s), one(&[(b"k", i64::MAX - 2)]));
+}
+
+#[test]
+fn a_failed_callback_step_is_rolled_back_too() {
+    let c = C::new();
+    let i = c.input();
+    let (mut f, mut int) = (0, 0);
+    assert_eq!(unsafe { zs_circuit_integrate(c.0, i, &mut int) }, OK);
+    let s = c.sink(int);
+    assert_eq!(
+        unsafe { zs_circuit_filter(c.0, i, Some(fails), null_mut(), &mut f) },
+        OK
+    );
+    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
+    c.push(i, &[(b"k", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Callback as i32);
+    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, OK);
+    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK, "usable again");
+    assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
+    assert_eq!(
+        c.take(s),
+        vec![],
+        "[callback] the failed step left no output"
+    );
+}
+
+#[test]
+fn transaction_misuse_is_a_code_and_changes_nothing() {
+    let (c, i, s) = at_max();
+    let t = Code::Transaction as i32;
+    assert_eq!(unsafe { zs_circuit_commit(c.0) }, t);
+    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
+    assert_eq!(unsafe { zs_circuit_begin(c.0) }, OK);
+    assert_eq!(unsafe { zs_circuit_begin(c.0) }, t);
+    let mut z = null_mut();
+    assert_eq!(unsafe { zs_circuit_take(c.0, s, &mut z) }, t);
+    assert!(z.is_null(), "nothing written");
+    assert!(last_error().contains("roll back"), "{}", last_error());
+    assert_eq!(unsafe { zs_circuit_commit(c.0) }, OK);
+    // A poison from outside any transaction is not undone by a rollback.
+    c.push(i, &[(b"k", 1)]);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Overflow as i32);
+    assert_eq!(unsafe { zs_circuit_rollback(c.0) }, t);
+    assert_eq!(unsafe { zs_circuit_step(c.0) }, Code::Poisoned as i32);
+}

crates/zset-ffi/tests/smoke.lisp

@@ -48,6 +48,8 @@
 (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))
+(defcfun "zs_circuit_begin" :int32 (c :pointer))
+(defcfun "zs_circuit_rollback" :int32 (c :pointer))
 
 (defcallback keep-or-fail :int32 ((ctx :pointer) (key :pointer) (len :size))
   (declare (ignore ctx key len))
@@ -61,9 +63,13 @@
     (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 "begin a transaction" (zs-circuit-begin c) 0)
     (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)
+    (check "rollback revives it" (zs-circuit-rollback c) 0)
+    (check "a transaction opens again" (zs-circuit-begin c) 0)
+    (check "and a second begin is refused" (zs-circuit-begin c) 10)
     (zs-zset-free (mem-ref zout :pointer))
     (zs-circuit-free c)))
 (format t "zset smoke: all ok~%")