Skip to content

refactor(node): feed a dispute ruler only the window it was positioned at #302

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GCdePaula merged 5 commits into from
Oct 20, 2026
+108 −311
Merged

refactor(node): feed a dispute ruler only the window it was positioned at #302

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
10f2fb6
test(node): replay levels wider than a window on the scratch feeder
GCdePaula Oct 19, 2026
ffa2ed5
test(node): pin a rejection fed at a crossed dispute boundary
GCdePaula Oct 19, 2026
436fc1c
refactor(node): feed a dispute ruler only the window it was positione…
GCdePaula Oct 19, 2026
42d571e
refactor(node): delete the dispute write-back's boundary store
GCdePaula Oct 19, 2026
7becfb0
docs(node): say which stores the dispute machine can serve unaided
GCdePaula Oct 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions cartesi-rollups/node/src/engine/dispute.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,12 +244,15 @@ impl<F: RulerFactory> DisputeSource<F> {
// beyond the prefix - a dispute descent through a padding
// window's root stores its fanout there - and counting those
// once bricked reconstruction after the hero's own join. A
// store the runner never processed (the engine harnesses; a
// freshly initialized node) has an empty prefix and the machine
// serves everything - the pre-frontier full-replay behavior.
// A nonzero prefix must match the closed epoch's input count
// exactly; anything else is corruption. The padding value is
// the final boundary hash.
// store the runner never processed has an empty prefix, so the
// machine serves every quartet, which it can only where the
// quartet feeds at most one input: the dispute feeder feeds only
// its positioned window. Production never disputes such a store
// with inputs, since a hero starts only once the runner rolled
// the epoch; the engine harnesses replay wider levels on the
// scratch feeder. A nonzero prefix must match the closed epoch's
// input count exactly; anything else is corruption. The padding
// value is the final boundary hash.
let interior_height = structure.log2_window_span() - log2_run_stride;
let inputs = storage.input_count(epoch)?;
let rows = storage.window_root_count(epoch, log2_run_stride, interior_height, inputs)?;
Expand Down
61 changes: 33 additions & 28 deletions cartesi-rollups/node/src/engine/machine_stf.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,10 +46,9 @@ use std::path::{Path, PathBuf};
/// aspects of the one fused verb. Scratch carries explicit payload
/// vectors and the directory its checkpoints go to (the storage-less
/// harnesses: measure, the differential tests). Store is the dispute
/// path's mode: payloads come from the inputs table and the pre-feed
/// snapshot commits into the boundary store; positioning has usually
/// published it already (Positioner::cross), so the commit adopts it,
/// and the row insert is a cross-regime nondeterminism tripwire.
/// path's mode: it feeds only the window it was positioned at, its
/// payload from the inputs table, and the pre-feed snapshot is that
/// window's stored boundary, the directory positioning loaded.
/// Advance is the mode of the machine runner and of positioning's
/// crossed windows: one window, its payload handed in, and the
/// pre-feed snapshot is the batch's closed checkpoint. It may be
Expand All @@ -65,7 +64,8 @@ enum Feeder {
Store {
storage: Storage,
epoch: u64,
next_input: u64,
window: u64,
boundary: PathBuf,
},
Advance {
window: u64,
Expand Down Expand Up @@ -250,15 +250,22 @@ impl MachineStf {
matches!(self.feeder, Feeder::Advance { reverted: true, .. })
}

/// Upgrades the feeder into the node's storage: this machine sits
/// at `next_input`'s boundary of `epoch`; every window it feeds
/// from here reads its payload from the inputs table and commits
/// the crossed boundary.
pub fn with_write_back(mut self, storage: Storage, epoch: u64, next_input: u64) -> Self {
/// The dispute path's feeder: this machine was loaded from
/// `boundary`, the stored boundary of `epoch` before input `window`.
/// It feeds that window alone, its payload from the inputs table,
/// and a rejection reloads `boundary`.
fn at_stored_boundary(
mut self,
storage: Storage,
epoch: u64,
window: u64,
boundary: PathBuf,
) -> Self {
self.feeder = Feeder::Store {
storage,
epoch,
next_input,
window,
boundary,
};
self
}
Expand Down Expand Up @@ -375,27 +382,23 @@ impl Stf for MachineStf {
Feeder::Store {
storage,
epoch,
next_input,
window: expected,
boundary,
} => {
assert_eq!(
window, *next_input,
"windows feed sequentially from the resume point"
window, *expected,
"the dispute stf feeds only the window it was positioned at"
);
*next_input += 1;
let payload = storage
.input(&InputId {
epoch_number: *epoch,
input_index_in_epoch: window,
})?
.expect("fed windows lie in the ingested contiguous prefix")
.data;
// The write-back: this boundary joins the store
// (stored only where regime 1 has not already), and
// the committed directory is the revert point -
// never removed here, it is the store's.
let dir =
storage.commit_boundary_machine(*epoch, window, &root, &mut self.machine)?;
(dir, payload)
// Positioning verified the directory against its row, and
// nothing ran before this feed.
(boundary.clone(), payload)
}
Feeder::Advance {
window: expected,
Expand Down Expand Up @@ -750,7 +753,7 @@ impl RulerFactory for Positioner {

fn ruler_at(&mut self, position: U256, hashing: Hashing) -> Result<Ruler<MachineStf>> {
let target = self.structure.decompose(position).input;
let (boundary, stf) = loop {
let (boundary, path, stf) = loop {
let (boundary, path) = self
.store
.nearest_boundary_at_or_before(self.epoch, target)?;
Expand All @@ -768,9 +771,10 @@ impl RulerFactory for Positioner {
// free - committed boundaries carry exact sidecars).
// Publication verifies before its rename, so a mismatch is
// corrupt local state, which stops the node as the runner's
// cursor does. Skipping it would heal only sidecar damage at
// an accepted input, at the price of full machine stores,
// since the write-back would re-adopt the torn directory.
// cursor does. Skipping it would mean feeding the windows
// above an earlier boundary, since a crossing re-adopts the
// torn directory by its name, and the dispute feeder feeds
// only the window it is positioned at.
let expected = self
.store
.snapshot_hash(self.epoch, boundary.0)?
Expand All @@ -790,18 +794,19 @@ impl RulerFactory for Positioner {
self.cross(boundary.0, expected, to)?;
continue;
}
break (boundary, stf);
break (boundary, path, stf);
};

// The seam's one boundary-to-position conversion.
let at = boundary.position(&self.structure);
assert!(at <= position, "boundary store answered past the target");

// Left to cross: padding windows past the inputs.
let stf = stf.with_write_back(
let stf = stf.at_stored_boundary(
Storage::new(self.store.state_dir())?,
self.epoch,
boundary.0,
path,
);
let mut ruler = Ruler::new_at(stf, self.structure, self.fed_windows, at);
ruler.advance(position)?;
Expand Down
2 changes: 1 addition & 1 deletion cartesi-rollups/node/src/harness/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -484,7 +484,7 @@ impl World {
) -> Result<Adversary> {
// A private copy of the node's store, as an independent sybil would
// have: the adversary reads the node's rolled material, but its
// event cache, quartet rows and write-backs stay out of the node's
// event cache, quartet rows and crossings stay out of the node's
// store, so the node computes and harvests everything itself. Its
// snapshots link the node's, which are immutable once published;
// linked after the copy, so every copied row finds its directory.
Expand Down
188 changes: 2 additions & 186 deletions cartesi-rollups/node/src/storage/snapshots.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,74 +97,6 @@ pub(super) fn assert_runner_boundary_root(
}

impl Storage {
/// The dispute write-back: commits a raw machine sitting at an
/// input boundary - stored only when the content-addressed
/// directory is absent, then registered. The row's write-once
/// cell makes the whole verb idempotent AND a cross-regime
/// nondeterminism tripwire: a replay that reaches a boundary
/// regime 1 recorded must produce the identical hash or fail
/// loudly. Returns the committed directory (the caller's revert
/// point).
pub fn commit_boundary_machine(
&mut self,
epoch_number: u64,
input_number: u64,
state_hash: &Hash,
machine: &mut Machine,
) -> Result<PathBuf> {
let actual_hash = machine.root_hash()?;
assert_eq!(
actual_hash, *state_hash,
"dispute boundary CAS key disagrees with the machine root"
);
let dest = self.stage_machine_store(state_hash, |staging| machine.store(staging))?;
self.insert_boundary(epoch_number, input_number, state_hash)?;
Ok(dest)
}

/// Serializes into a unique staging directory before durable CAS
/// publication. Staging names are never keyed by hash: the hero's
/// write-back and the runner's batch commit may publish an identical
/// state concurrently, and publish_staged settles the race.
fn stage_machine_store(
&self,
state_hash: &Hash,
store: impl FnOnce(&Path) -> cartesi_machine::error::MachineResult<()>,
) -> anyhow::Result<PathBuf> {
let snapshots = snapshots_path(self.state_dir());
let dest = snapshot_path(self.state_dir(), state_hash);

if !dest.exists() {
static SEQ: AtomicU64 = AtomicU64::new(0);
let staging = snapshots.join(format!(
".part-{}-{}",
std::process::id(),
SEQ.fetch_add(1, Ordering::Relaxed)
));

if let Err(machine_err) = store(&staging) {
// cleanup partial store before returning error.
let fs_status = std::fs::remove_dir_all(&staging);

if let Err(fs_err) = fs_status {
anyhow::bail!(
"Failed to cleanup partial store {fs_err}, caused by {machine_err}"
);
} else {
return Err(machine_err.into());
}
}

return self.publish_staged(staging, state_hash);
}

// A concurrent publisher may have made the directory visible
// before its own parent fsync completed. Adoption establishes
// durability independently before any caller registers a row.
Machine::sync_stored(&dest)?;
Ok(dest)
}

/// The one durable CAS publication boundary. The staged machine
/// is root-verified, synced, and renamed without replacement.
/// Only a destination that won the same CAS race makes a failed
Expand Down Expand Up @@ -242,7 +174,8 @@ impl Storage {
}

/// Startup ritual: removes staging leftovers a crash orphaned -
/// partial stores (.part-*) and working clones (.work-*). Runs at
/// working clones (.work-*), and the partial stores (.part-*) of
/// versions that still stored dispute boundaries. Runs at
/// open, under the directory lock and before seeding, so no clone is
/// held yet; never touches committed directories.
pub fn sweep_stale_staging(&self) -> Result<()> {
Expand Down Expand Up @@ -590,86 +523,6 @@ mod tests {
let _ = s.insert_boundary(7, 3, &[6u8; 32]);
}

/// The dispute write-back verb: stores only when the CAS misses,
/// absorbs identical recommits, and fails loudly on a divergent
/// hash for a written boundary (the cross-regime tripwire).
#[test]
fn commit_boundary_machine_is_idempotent_and_loud() {
let (_handle, mut s) = setup_storage();
let template = s.snapshot_dir(0, 0).unwrap().unwrap();
let mut machine = cartesi_machine::machine::Machine::load(
&template,
&cartesi_machine::config::runtime::RuntimeConfig::quiet_console(),
)
.unwrap();
let hash = machine.root_hash().unwrap();

// The CAS already holds this state (it is the template), so
// the commit only registers the row.
let dest = s
.commit_boundary_machine(7, 3, &hash, &mut machine)
.unwrap();
assert_eq!(dest, template);
assert_eq!(s.snapshot_dir(7, 3).unwrap().unwrap(), dest);

// An identical recommit absorbs.
s.commit_boundary_machine(7, 3, &hash, &mut machine)
.unwrap();

// A boundary whose state the CAS misses gets stored under the
// root derived from the actual machine.
machine
.write_memory(cartesi_machine::constants::ar::RAM_START, &[0xA5])
.unwrap();
let other = machine.root_hash().unwrap();
assert_ne!(other, hash);
let dest2 = s
.commit_boundary_machine(7, 4, &other, &mut machine)
.unwrap();
assert_ne!(dest2, dest);
assert!(dest2.join("config.json").exists());
}

/// A divergent hash for a written boundary is nondeterminism (the
/// cross-regime tripwire): PANIC, never a retryable error.
#[test]
#[should_panic(expected = "disagrees with its stored row")]
fn commit_boundary_machine_divergence_panics() {
let (_handle, mut s) = setup_storage();
let template = s.snapshot_dir(0, 0).unwrap().unwrap();
let mut machine = cartesi_machine::machine::Machine::load(
&template,
&cartesi_machine::config::runtime::RuntimeConfig::quiet_console(),
)
.unwrap();
let hash = machine.root_hash().unwrap();
s.commit_boundary_machine(7, 3, &hash, &mut machine)
.unwrap();

machine
.write_memory(cartesi_machine::constants::ar::RAM_START, &[0x5A])
.unwrap();
let divergent = machine.root_hash().unwrap();
assert_ne!(divergent, hash);
let _ = s.commit_boundary_machine(7, 3, &divergent, &mut machine);
}

#[test]
#[should_panic(expected = "CAS key disagrees")]
fn commit_boundary_machine_refuses_a_supplied_wrong_root() {
let (_handle, mut s) = setup_storage();
let template = s.snapshot_dir(0, 0).unwrap().unwrap();
let mut machine = cartesi_machine::machine::Machine::load(
&template,
&cartesi_machine::config::runtime::RuntimeConfig::quiet_console(),
)
.unwrap();
let mut wrong = machine.root_hash().unwrap();
wrong[0] ^= 0xFF;

let _ = s.commit_boundary_machine(7, 3, &wrong, &mut machine);
}

/// The floor query answers the nearest surviving boundary,
/// skipping rows whose directories vanished.
#[test]
Expand Down Expand Up @@ -701,43 +554,6 @@ mod tests {
assert!(s.nearest_boundary_at_or_before(6, 100).is_err());
}

/// GC never removes a directory a crossing can adopt (the argument is in
/// node-architecture.md); one that vanished anyway is skipped by
/// positioning and republished by the next crossing.
#[test]
fn a_vanished_dispute_boundary_is_republished() {
let (_handle, mut s) = setup_storage();
let template = s.snapshot_dir(0, 0).unwrap().unwrap();
let mut machine = cartesi_machine::machine::Machine::load(
&template,
&cartesi_machine::config::runtime::RuntimeConfig::quiet_console(),
)
.unwrap();
let start = machine.root_hash().unwrap();
s.commit_boundary_machine(7, 0, &start, &mut machine)
.unwrap();
machine
.write_memory(cartesi_machine::constants::ar::RAM_START, &[0xA5])
.unwrap();
let hash = machine.root_hash().unwrap();
let dest = s
.commit_boundary_machine(7, 4, &hash, &mut machine)
.unwrap();

std::fs::remove_dir_all(&dest).unwrap();
let (floor, _) = s.nearest_boundary_at_or_before(7, 4).unwrap();
assert_eq!(floor, InputBoundary(0));

let republished = s
.commit_boundary_machine(7, 4, &hash, &mut machine)
.unwrap();
assert_eq!(republished, dest);
assert!(dest.join("config.json").exists());
assert_eq!(s.snapshot_hash(7, 4).unwrap(), Some(hash));
let (floor, _) = s.nearest_boundary_at_or_before(7, 4).unwrap();
assert_eq!(floor, InputBoundary(4));
}

/// A release orphans exactly the states no remaining boundary names,
/// once each: a state a later epoch shares survives.
#[test]
Expand Down
Loading
Loading