From dd7904963dca62e91ffe1a137741511974adcefd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fabr=C3=ADcio=20Bracht?= Date: Wed, 14 Oct 2026 14:07:31 -0700 Subject: [PATCH] refuse unencodable or unpersisted writes and count live sessions --- Cargo.lock | 1 + crates/stitch-p2p/Cargo.toml | 1 + crates/stitch-p2p/README.md | 17 +- crates/stitch-p2p/src/lib.rs | 4 +- crates/stitch-p2p/src/membership.rs | 4 +- crates/stitch-p2p/src/node.rs | 65 +++++- crates/stitch-p2p/src/persistence.rs | 40 ++-- crates/stitch-p2p/src/replog.rs | 36 ++-- crates/stitch-p2p/src/session.rs | 31 ++- crates/stitch-p2p/src/store.rs | 35 ++-- crates/stitch-p2p/src/sync_state.rs | 207 ++++++++++++++++---- crates/stitch-p2p/src/wire.rs | 13 +- crates/stitch-p2p/tests/discovery_broker.rs | 6 +- crates/stitch-p2p/tests/membership_sync.rs | 6 +- crates/stitch-p2p/tests/node_mesh.rs | 12 +- crates/stitch-p2p/tests/persistence.rs | 41 +++- crates/stitch-p2p/tests/quic_loopback.rs | 6 +- crates/stitch-p2p/tests/transitive.rs | 21 +- crates/stitch-tasks/src/bin/demo.rs | 18 +- crates/stitch-tasks/src/bin/soak.rs | 8 +- crates/stitch-tasks/src/harness.rs | 31 +-- crates/stitch-tasks/src/lib.rs | 33 ++-- crates/stitch-tasks/tests/soak.rs | 3 +- 23 files changed, 475 insertions(+), 164 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 8060f7e..7df82a8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2400,6 +2400,7 @@ dependencies = [ "tempfile", "thiserror", "tokio", + "tracing", ] [[package]] diff --git a/crates/stitch-p2p/Cargo.toml b/crates/stitch-p2p/Cargo.toml index fc790d7..f833f1e 100644 --- a/crates/stitch-p2p/Cargo.toml +++ b/crates/stitch-p2p/Cargo.toml @@ -17,6 +17,7 @@ mqp2p = { version = "0.3.0", path = "../mqp2p", optional = true } serde_json = { version = "1.0.150", optional = true } thiserror = "2.0.18" tokio = { version = "1.52.3", features = ["sync", "io-util", "macros", "rt", "rt-multi-thread", "time", "net"] } +tracing = "0.1.44" [dev-dependencies] tempfile = "3.27.0" diff --git a/crates/stitch-p2p/README.md b/crates/stitch-p2p/README.md index 32df11b..01a9603 100644 --- a/crates/stitch-p2p/README.md +++ b/crates/stitch-p2p/README.md @@ -70,7 +70,8 @@ Implemented and tested (the full verified core, transport excluded): no broker, no STUN). - `node` — `SyncNode`, the fan-out layer. One shared `SyncState` across all of a device's sessions; `register_session` hands a session its outbound channel, - `local_write` applies and fans out to every live session. **Transitive + `local_write` applies and fans out to every live session, and + `session_count` counts the sessions whose receiver is still alive. **Transitive forwarding falls out of the shared state**: a write pulled from one peer is served to the others on their next pull — no re-broadcast, no echo suppression. `tests/node_mesh.rs` proves a line `A—B—C` converges through the @@ -99,21 +100,27 @@ Implemented and tested (the full verified core, transport excluded): | Feature | Default | Pulls in | Gates | |---|---|---|---| -| (none) | — | `tokio`, `thiserror` | the verified core: `hlc`, `lww`, `wire`, `replog`, `sync_state`, `protocol`, `session`, `node` | +| (none) | — | `tokio`, `thiserror`, `tracing` | the verified core: `hlc`, `lww`, `wire`, `replog`, `sync_state`, `protocol`, `session`, `node` | | `store` | on | `serde_json` | `stitch_p2p::store::Store` (JSON document facade) | | `discovery` | on | `mqp2p` (→ quinn, mqtt5) | `stitch_p2p::discovery::Swarm` (peer discovery + NAT + QUIC) | | `persistence` | on | `fjall` | `stitch_p2p::persistence::FjallLog` + `Store::open(path)` (durable replication log) | | `membership` | on | `ed25519-dalek` | `stitch_p2p::membership::Identity` — Ed25519-signed writes; `Store::with_identity` | `default-features = false` builds the formally-verified engine with just -`tokio` + `thiserror` — no networking, no JSON, no transitive QUIC/MQTT stack. +`tokio` + `thiserror` + `tracing` — no networking, no JSON, no transitive QUIC/MQTT stack. Add `store` for the document API, `discovery` for the mqp2p transport. - `persistence` — `FjallLog`, a durable replication log. `SyncState` gains a - `FramePersister` hook (core trait, no deps); every appended frame is persisted - and `Store::open(path)` rebuilds state by replaying the log — recovering the + `FramePersister` hook (core trait, no deps); every frame is persisted before + it enters the log, and `Store::open(path)` rebuilds state by replaying the log — recovering the HLC so post-restart writes still win. `tests/persistence.rs` proves state and tombstones survive a reopen. + A local write the persister cannot store is refused (`WriteError::Persist`) + and takes no sequence number, a received frame it cannot store is not + recorded (`RecordOutcome::Unpersisted`) and is pulled again, and a log that + cannot be read fails `Store::open` rather than replaying around the hole. A + write whose entity or id exceeds 255 bytes is refused up front + (`WriteError::Wire`), so every frame in the log encodes. - `membership` — Ed25519 `Identity` (peer id == public key) signing every local write; `SyncState` verifies inbound frames and rejects bad signatures diff --git a/crates/stitch-p2p/src/lib.rs b/crates/stitch-p2p/src/lib.rs index 1a5c7aa..3224253 100644 --- a/crates/stitch-p2p/src/lib.rs +++ b/crates/stitch-p2p/src/lib.rs @@ -40,8 +40,8 @@ pub use lww::{Applier, MergeOutcome, Op, StampedWrite}; pub use node::SyncNode; pub use protocol::{MAX_MESSAGE_LEN, ProtocolError, SyncMessage}; pub use replog::{Cursors, RecordOutcome, ReplLog}; -pub use sync_state::{MutationEvent, SyncState, WriteOrigin}; -pub use wire::{HEADER_LEN, WIRE_VERSION, WireError, WriteFrame}; +pub use sync_state::{MutationEvent, SyncState, WriteError, WriteOrigin}; +pub use wire::{HEADER_LEN, WIRE_VERSION, WireError, WriteFrame, check_lengths}; pub use sync_state::{FrameAuth, FramePersister}; diff --git a/crates/stitch-p2p/src/membership.rs b/crates/stitch-p2p/src/membership.rs index dc81e3b..8a15cc2 100644 --- a/crates/stitch-p2p/src/membership.rs +++ b/crates/stitch-p2p/src/membership.rs @@ -300,7 +300,9 @@ mod tests { let id = Arc::new(Identity::from_seed([5u8; 32])); let mut state = SyncState::new(id.peer_id()); state.set_auth(id.clone()); - let frame = state.local_write(1, Op::Insert, "task", "t1", vec![9]); + let frame = state + .local_write(1, Op::Insert, "task", "t1", vec![9]) + .expect("write"); assert!(frame.signature.is_some()); assert!(verify_frame(&frame)); } diff --git a/crates/stitch-p2p/src/node.rs b/crates/stitch-p2p/src/node.rs index 99d1047..0ef6996 100644 --- a/crates/stitch-p2p/src/node.rs +++ b/crates/stitch-p2p/src/node.rs @@ -1,5 +1,5 @@ use crate::lww::Op; -use crate::sync_state::SyncState; +use crate::sync_state::{SyncState, WriteError}; use crate::wire::WriteFrame; use std::sync::{Arc, Mutex as StdMutex}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -48,17 +48,21 @@ impl SyncNode { /// Build a node whose state is rebuilt by replaying the persister's stored /// frames, then continues to persist new writes to it. - #[must_use] + /// + /// # Errors + /// + /// When the persister cannot load its frames. Starting without them would + /// reuse sequence numbers peers already hold. pub fn with_persister( self_id: crate::hlc::PeerId, persister: Arc, - ) -> Self { + ) -> std::io::Result { let mut state = SyncState::new(self_id); - for frame in persister.load() { + for frame in persister.load()? { state.replay(frame); } state.set_persister(persister); - Self::from_state(state) + Ok(Self::from_state(state)) } #[must_use] @@ -76,22 +80,26 @@ impl SyncNode { /// Apply a local mutation and fan it out to every live session. Closed /// session channels are pruned. + /// + /// # Errors + /// + /// See [`SyncState::local_write`]; a refused write is not sent. pub async fn local_write( &self, op: Op, entity: impl Into, id: impl Into, data: Vec, - ) -> WriteFrame { + ) -> Result { let frame = { let mut guard = self.state.lock().await; - guard.local_write(now_millis(), op, entity, id, data) + guard.local_write(now_millis(), op, entity, id, data)? }; self.sessions .lock() .expect("sessions lock") .retain(|tx| tx.send(frame.clone()).is_ok()); - frame + Ok(frame) } #[must_use] @@ -141,8 +149,47 @@ impl SyncNode { self.state.lock().await.reclaim(members) } + /// Live sessions: those whose receiver has not been dropped. #[must_use] pub fn session_count(&self) -> usize { - self.sessions.lock().expect("sessions lock").len() + let mut sessions = self.sessions.lock().expect("sessions lock"); + sessions.retain(|tx| !tx.is_closed()); + sessions.len() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + struct Unreadable; + + impl crate::sync_state::FramePersister for Unreadable { + fn append(&self, _: &WriteFrame) -> std::io::Result<()> { + Ok(()) + } + + fn load(&self) -> std::io::Result> { + Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "corrupt frame", + )) + } + } + + #[test] + fn session_count_skips_dropped_sessions() { + let node = SyncNode::new([1; crate::hlc::PEER_ID_LEN]); + let live = node.register_session(); + drop(node.register_session()); + assert_eq!(node.session_count(), 1); + drop(live); + assert_eq!(node.session_count(), 0); + } + + #[test] + fn a_log_that_cannot_be_loaded_is_an_error() { + let opened = SyncNode::with_persister([1; crate::hlc::PEER_ID_LEN], Arc::new(Unreadable)); + assert!(opened.is_err()); } } diff --git a/crates/stitch-p2p/src/persistence.rs b/crates/stitch-p2p/src/persistence.rs index 264a6e6..d3da037 100644 --- a/crates/stitch-p2p/src/persistence.rs +++ b/crates/stitch-p2p/src/persistence.rs @@ -2,6 +2,7 @@ use crate::hlc::PEER_ID_LEN; use crate::sync_state::FramePersister; use crate::wire::WriteFrame; use fjall::{Database, Keyspace, KeyspaceCreateOptions, PersistMode}; +use std::io; use std::path::Path; const KEY_LEN: usize = PEER_ID_LEN + 8; @@ -20,6 +21,8 @@ pub struct FjallLog { pub enum PersistError { #[error("fjall: {0}")] Fjall(#[from] fjall::Error), + #[error("replication log unreadable: {0}")] + Load(#[from] io::Error), } impl FjallLog { @@ -38,25 +41,26 @@ fn frame_key(frame: &WriteFrame) -> [u8; KEY_LEN] { } impl FramePersister for FjallLog { - fn append(&self, frame: &WriteFrame) { - let Ok(bytes) = frame.encode() else { - return; - }; - if self.frames.insert(&frame_key(frame)[..], bytes).is_ok() { - let _ = self.db.persist(PersistMode::SyncAll); - } + fn append(&self, frame: &WriteFrame) -> io::Result<()> { + let bytes = frame + .encode() + .map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))?; + self.frames + .insert(&frame_key(frame)[..], bytes) + .map_err(io::Error::other)?; + self.db + .persist(PersistMode::SyncAll) + .map_err(io::Error::other) } - fn load(&self) -> Vec { - let mut out = Vec::new(); - for guard in self.frames.iter() { - let Ok((_, value)) = guard.into_inner() else { - continue; - }; - if let Ok(frame) = WriteFrame::decode(&value) { - out.push(frame); - } - } - out + fn load(&self) -> io::Result> { + self.frames + .iter() + .map(|guard| { + let (_, value) = guard.into_inner().map_err(io::Error::other)?; + WriteFrame::decode(&value) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)) + }) + .collect() } } diff --git a/crates/stitch-p2p/src/replog.rs b/crates/stitch-p2p/src/replog.rs index 11a891b..71cf156 100644 --- a/crates/stitch-p2p/src/replog.rs +++ b/crates/stitch-p2p/src/replog.rs @@ -14,6 +14,9 @@ pub enum RecordOutcome { }, /// The frame failed authentication/authorization and was not recorded. Rejected, + /// The frame could not be written to durable storage and was not + /// recorded; the peer offers it again on a later pull. + Unpersisted, } #[derive(Debug)] @@ -56,20 +59,29 @@ impl ReplLog { self.cursor_for(origin) + 1 } - pub fn record(&mut self, frame: WriteFrame) -> RecordOutcome { - let origin = frame.stamp.peer; - let have = self.cursor_for(&origin); - if frame.seq <= have { - return RecordOutcome::Duplicate; - } - if frame.seq != have + 1 { - return RecordOutcome::Gap { + /// What [`ReplLog::record`] would do with `frame`, without recording it. + #[must_use] + pub fn classify(&self, frame: &WriteFrame) -> RecordOutcome { + let have = self.cursor_for(&frame.stamp.peer); + match frame.seq { + seq if seq <= have => RecordOutcome::Duplicate, + seq if seq == have + 1 => RecordOutcome::Appended, + got => RecordOutcome::Gap { expected: have + 1, - got: frame.seq, - }; + got, + }, + } + } + + pub fn record(&mut self, frame: WriteFrame) -> RecordOutcome { + let outcome = self.classify(&frame); + if outcome == RecordOutcome::Appended { + self.observed + .entry(frame.stamp.peer) + .or_default() + .push(frame); } - self.observed.entry(origin).or_default().push(frame); - RecordOutcome::Appended + outcome } #[must_use] diff --git a/crates/stitch-p2p/src/session.rs b/crates/stitch-p2p/src/session.rs index c4b4124..cc02b39 100644 --- a/crates/stitch-p2p/src/session.rs +++ b/crates/stitch-p2p/src/session.rs @@ -1,4 +1,5 @@ use crate::protocol::{ProtocolError, SyncMessage, read_message, write_message}; +use crate::replog::RecordOutcome; use crate::sync_state::SyncState; use crate::wire::WriteFrame; use std::sync::Arc; @@ -93,7 +94,17 @@ async fn apply(state: &Mutex, frames: Vec) { let now = now_millis(); let mut guard = state.lock().await; for frame in frames { - guard.receive(frame, now); + let origin = frame.stamp.peer; + let seq = frame.seq; + match guard.receive(frame, now) { + RecordOutcome::Appended | RecordOutcome::Duplicate | RecordOutcome::Unpersisted => {} + RecordOutcome::Gap { expected, got } => { + tracing::debug!(origin = ?&origin[..4], expected, got, "frame ahead of the log, waiting for the missing ones"); + } + RecordOutcome::Rejected => { + tracing::warn!(origin = ?&origin[..4], seq, "frame failed verification and was dropped"); + } + } } } @@ -125,11 +136,13 @@ mod tests { a_state .lock() .await - .local_write(10, Op::Insert, "task", "t1", b"from-a".to_vec()); + .local_write(10, Op::Insert, "task", "t1", b"from-a".to_vec()) + .expect("write"); b_state .lock() .await - .local_write(11, Op::Insert, "task", "t2", b"from-b".to_vec()); + .local_write(11, Op::Insert, "task", "t2", b"from-b".to_vec()) + .expect("write"); let (a_io, b_io) = tokio::io::duplex(16 * 1024); let (a_r, a_w) = split(a_io); @@ -178,13 +191,11 @@ mod tests { let b = tokio::spawn(run(b_state.clone(), b_r, b_w, b_rx, FAST_PULL)); // Write on A *after* the session is live; push it through A's outbound. - let frame = a_state.lock().await.local_write( - now_millis(), - Op::Insert, - "task", - "late", - b"v".to_vec(), - ); + let frame = a_state + .lock() + .await + .local_write(now_millis(), Op::Insert, "task", "late", b"v".to_vec()) + .expect("write"); a_tx.send(frame).unwrap(); let ok = wait_until(Duration::from_secs(2), || async { diff --git a/crates/stitch-p2p/src/store.rs b/crates/stitch-p2p/src/store.rs index 1dae3ad..eb75d56 100644 --- a/crates/stitch-p2p/src/store.rs +++ b/crates/stitch-p2p/src/store.rs @@ -1,7 +1,7 @@ use crate::hlc::PeerId; use crate::lww::Op; use crate::node::SyncNode; -use crate::sync_state::MutationEvent; +use crate::sync_state::{MutationEvent, WriteError}; use serde_json::{Map, Value}; use tokio::sync::broadcast; @@ -28,6 +28,8 @@ pub struct Store { pub enum StoreError { #[error("serialize: {0}")] Serialize(#[from] serde_json::Error), + #[error(transparent)] + Write(#[from] WriteError), } impl Store { @@ -48,7 +50,7 @@ impl Store { ) -> Result { let log = std::sync::Arc::new(crate::persistence::FjallLog::open(path)?); Ok(Self { - node: SyncNode::with_persister(self_id, log), + node: SyncNode::with_persister(self_id, log)?, owner: None, }) } @@ -106,7 +108,7 @@ impl Store { ) -> Result { let log = std::sync::Arc::new(crate::persistence::FjallLog::open(path)?); let mut state = crate::SyncState::new(identity.peer_id()); - for frame in crate::FramePersister::load(log.as_ref()) { + for frame in crate::FramePersister::load(log.as_ref())? { state.replay(frame); } state.set_persister(log); @@ -126,7 +128,7 @@ impl Store { /// Insert or replace a record. pub async fn create(&self, entity: &str, id: &str, record: Value) -> Result<(), StoreError> { let data = serde_json::to_vec(&record)?; - self.node.local_write(Op::Insert, entity, id, data).await; + self.node.local_write(Op::Insert, entity, id, data).await?; Ok(()) } @@ -158,15 +160,16 @@ impl Store { merged.insert(k, v); } let data = serde_json::to_vec(&Value::Object(merged))?; - self.node.local_write(Op::Update, entity, id, data).await; + self.node.local_write(Op::Update, entity, id, data).await?; Ok(()) } /// Delete a record (writes a tombstone). - pub async fn delete(&self, entity: &str, id: &str) { + pub async fn delete(&self, entity: &str, id: &str) -> Result<(), StoreError> { self.node .local_write(Op::Delete, entity, id, Vec::new()) - .await; + .await?; + Ok(()) } /// All visible records of an entity, filtered to authorized authors when the @@ -201,7 +204,11 @@ impl Store { /// is evaluated). No-op semantics if the caller lacks authority: the record /// is written but ignored by `authorized_members`. #[cfg(feature = "membership")] - pub async fn invite(&self, target: PeerId, role: crate::membership::Role) { + pub async fn invite( + &self, + target: PeerId, + role: crate::membership::Role, + ) -> Result<(), StoreError> { let id = crate::membership::member_record_id(&target); self.node .local_write( @@ -210,12 +217,13 @@ impl Store { &id, vec![role.as_byte()], ) - .await; + .await?; + Ok(()) } /// Revoke a peer's membership. #[cfg(feature = "membership")] - pub async fn revoke(&self, target: PeerId) { + pub async fn revoke(&self, target: PeerId) -> Result<(), StoreError> { let id = crate::membership::member_record_id(&target); self.node .local_write( @@ -224,7 +232,8 @@ impl Store { &id, Vec::new(), ) - .await; + .await?; + Ok(()) } async fn author_authorized(&self, author: PeerId) -> bool { @@ -306,7 +315,7 @@ mod tests { async fn delete_removes_record() { let store = Store::new(peer(1)); store.create("task", "t1", json!({"x": 1})).await.unwrap(); - store.delete("task", "t1").await; + store.delete("task", "t1").await.expect("write"); assert_eq!(store.read("task", "t1").await, None); } @@ -316,7 +325,7 @@ mod tests { store.create("task", "t1", json!({"n": 1})).await.unwrap(); store.create("task", "t2", json!({"n": 2})).await.unwrap(); store.create("note", "n1", json!({"n": 3})).await.unwrap(); - store.delete("task", "t2").await; + store.delete("task", "t2").await.expect("write"); let tasks = store.list("task").await; assert_eq!(tasks.len(), 1); assert_eq!(tasks[0], json!({"n": 1})); diff --git a/crates/stitch-p2p/src/sync_state.rs b/crates/stitch-p2p/src/sync_state.rs index 1e81fd6..f088a5d 100644 --- a/crates/stitch-p2p/src/sync_state.rs +++ b/crates/stitch-p2p/src/sync_state.rs @@ -1,19 +1,30 @@ use crate::hlc::{Hlc, PeerId, Stamp}; use crate::lww::{Applier, MergeOutcome, Op}; use crate::replog::{Cursors, RecordOutcome, ReplLog}; -use crate::wire::{SIGNATURE_LEN, WriteFrame}; +use crate::wire::{SIGNATURE_LEN, WireError, WriteFrame, check_lengths}; use std::sync::Arc; use tokio::sync::broadcast; const EVENT_CHANNEL_CAPACITY: usize = 1024; /// Durable sink for the per-origin replication log. Every frame that enters the -/// log (local or received) is handed to `append`; `load` returns all persisted -/// frames at startup so state can be rebuilt by replay. Implemented by the -/// `persistence` feature; the verified core stays storage-agnostic. +/// log (local or received) is handed to `append` first, and enters the log +/// only if `append` succeeds; `load` returns all persisted frames at startup so +/// state can be rebuilt by replay. Implemented by the `persistence` feature; +/// the verified core stays storage-agnostic. pub trait FramePersister: Send + Sync { - fn append(&self, frame: &WriteFrame); - fn load(&self) -> Vec; + fn append(&self, frame: &WriteFrame) -> std::io::Result<()>; + fn load(&self) -> std::io::Result>; +} + +/// Why a local write was refused. A refused write takes no sequence number +/// and changes no state. +#[derive(Debug, thiserror::Error)] +pub enum WriteError { + #[error("write does not fit a frame: {0}")] + Wire(#[from] WireError), + #[error("write could not be persisted: {0}")] + Persist(#[source] std::io::Error), } /// Signs locally-originated frames and verifies/authorizes received ones. @@ -156,8 +167,15 @@ impl SyncState { } /// Originate a local mutation. Advances the HLC, builds and (if a - /// [`FrameAuth`] is set) signs the frame, records it to our own log, applies - /// it locally, emits a [`MutationEvent`], and returns the frame to broadcast. + /// [`FrameAuth`] is set) signs the frame, persists it, records it to our + /// own log, applies it locally, emits a [`MutationEvent`], and returns the + /// frame to broadcast. + /// + /// # Errors + /// + /// [`WriteError::Wire`] when the entity, id or data do not fit a frame, and + /// [`WriteError::Persist`] when the persister fails. Either way the write + /// takes no sequence number and changes no state. pub fn local_write( &mut self, wall_ms: u64, @@ -165,7 +183,10 @@ impl SyncState { entity: impl Into, id: impl Into, data: Vec, - ) -> WriteFrame { + ) -> Result { + let entity = entity.into(); + let id = id.into(); + check_lengths(&entity, &id, &data)?; let hlc = self.clock.tick(wall_ms); let stamp = Stamp::new(hlc, self.self_id); let seq = self.log.next_seq(&self.self_id); @@ -173,28 +194,28 @@ impl SyncState { stamp, seq, op, - entity: entity.into(), - id: id.into(), + entity, + id, data, signature: None, }; - if let Some(auth) = &self.auth - && let Ok(bytes) = frame.signing_bytes() - { - frame.signature = Some(auth.sign(&bytes)); + if let Some(auth) = &self.auth { + frame.signature = Some(auth.sign(&frame.signing_bytes()?)); } + self.persist(&frame).map_err(WriteError::Persist)?; self.log.record(frame.clone()); - self.persist(&frame); self.applier.merge(frame.clone().into_stamped()); self.emit(&frame, WriteOrigin::Local); - frame + Ok(frame) } /// Apply a frame received from a peer. With a [`FrameAuth`] set, an /// unverified frame is rejected before it touches the log or state. Always /// advances the HLC by observing the remote stamp. Merges only when the /// frame is the next in-order write for its origin; emits a - /// [`MutationEvent`] only when it changes the visible state (won LWW). + /// [`MutationEvent`] only when it changes the visible state (won LWW). A + /// frame the persister fails to store is not recorded + /// ([`RecordOutcome::Unpersisted`]), so the peer offers it again. pub fn receive(&mut self, frame: WriteFrame, wall_ms: u64) -> RecordOutcome { if let Some(auth) = &self.auth && !auth.verify(&frame) @@ -202,19 +223,25 @@ impl SyncState { return RecordOutcome::Rejected; } self.clock.observe(frame.stamp.hlc, wall_ms); - let outcome = self.log.record(frame.clone()); - if outcome == RecordOutcome::Appended { - self.persist(&frame); - if self.applier.merge(frame.clone().into_stamped()) == MergeOutcome::Applied { - self.emit(&frame, WriteOrigin::Remote); - } + let outcome = self.log.classify(&frame); + if outcome != RecordOutcome::Appended { + return outcome; + } + if let Err(error) = self.persist(&frame) { + tracing::error!(seq = frame.seq, %error, "received frame not persisted"); + return RecordOutcome::Unpersisted; + } + self.log.record(frame.clone()); + if self.applier.merge(frame.clone().into_stamped()) == MergeOutcome::Applied { + self.emit(&frame, WriteOrigin::Remote); } outcome } - fn persist(&self, frame: &WriteFrame) { - if let Some(persister) = &self.persister { - persister.append(frame); + fn persist(&self, frame: &WriteFrame) -> std::io::Result<()> { + match &self.persister { + Some(persister) => persister.append(frame), + None => Ok(()), } } @@ -275,6 +302,7 @@ impl SyncState { mod tests { use super::*; use crate::hlc::PEER_ID_LEN; + use crate::wire::WireError; fn peer(n: u8) -> PeerId { let mut id = [0u8; PEER_ID_LEN]; @@ -282,10 +310,108 @@ mod tests { id } + struct Flaky { + failing: std::sync::atomic::AtomicBool, + stored: std::sync::Mutex>, + } + + impl Flaky { + fn new(failing: bool) -> Arc { + Arc::new(Self { + failing: std::sync::atomic::AtomicBool::new(failing), + stored: std::sync::Mutex::new(Vec::new()), + }) + } + + fn set_failing(&self, failing: bool) { + self.failing + .store(failing, std::sync::atomic::Ordering::SeqCst); + } + } + + impl FramePersister for Flaky { + fn append(&self, frame: &WriteFrame) -> std::io::Result<()> { + if self.failing.load(std::sync::atomic::Ordering::SeqCst) { + return Err(std::io::Error::other("disk full")); + } + self.stored.lock().expect("lock").push(frame.clone()); + Ok(()) + } + + fn load(&self) -> std::io::Result> { + Ok(self.stored.lock().expect("lock").clone()) + } + } + + #[test] + fn oversized_fields_are_refused_without_taking_a_sequence_number() { + let mut s = SyncState::new(peer(1)); + let long = "x".repeat(256); + assert!(matches!( + s.local_write(1, Op::Insert, "task", long.clone(), b"a".to_vec()), + Err(WriteError::Wire(WireError::IdTooLong)) + )); + assert!(matches!( + s.local_write(1, Op::Insert, long, "t1", b"a".to_vec()), + Err(WriteError::Wire(WireError::EntityTooLong)) + )); + let next = s + .local_write(2, Op::Insert, "task", "t1", b"a".to_vec()) + .expect("write"); + assert_eq!(next.seq, 1); + assert_eq!( + next.encode().map(|bytes| bytes.len()), + Ok(next.encoded_len()) + ); + } + + #[test] + fn unpersisted_local_write_changes_nothing() { + let disk = Flaky::new(true); + let mut s = SyncState::new(peer(1)); + s.set_persister(disk.clone()); + let mut events = s.subscribe(); + assert!(matches!( + s.local_write(1, Op::Insert, "task", "t1", b"a".to_vec()), + Err(WriteError::Persist(_)) + )); + assert_eq!(s.visible("task", "t1"), None); + assert!(s.cursors().is_empty()); + assert!(events.try_recv().is_err()); + + disk.set_failing(false); + let frame = s + .local_write(2, Op::Insert, "task", "t1", b"a".to_vec()) + .expect("write"); + assert_eq!(frame.seq, 1); + assert_eq!(disk.load().expect("load"), [frame]); + } + + #[test] + fn unpersisted_received_frame_is_not_recorded_and_is_taken_again() { + let mut a = SyncState::new(peer(1)); + let frame = a + .local_write(1, Op::Insert, "task", "t1", b"a".to_vec()) + .expect("write"); + let disk = Flaky::new(true); + let mut b = SyncState::new(peer(2)); + b.set_persister(disk.clone()); + + assert_eq!(b.receive(frame.clone(), 1), RecordOutcome::Unpersisted); + assert_eq!(b.visible("task", "t1"), None); + assert_eq!(b.cursors().get(&peer(1)), None); + + disk.set_failing(false); + assert_eq!(b.receive(frame, 2), RecordOutcome::Appended); + assert_eq!(b.visible("task", "t1"), Some(&b"a"[..])); + } + #[test] fn local_write_is_visible_and_broadcastable() { let mut s = SyncState::new(peer(1)); - let frame = s.local_write(100, Op::Insert, "task", "t1", b"hi".to_vec()); + let frame = s + .local_write(100, Op::Insert, "task", "t1", b"hi".to_vec()) + .expect("write"); assert_eq!(frame.seq, 1); assert_eq!(s.visible("task", "t1"), Some(&b"hi"[..])); } @@ -294,7 +420,9 @@ mod tests { fn receive_applies_remote_write() { let mut a = SyncState::new(peer(1)); let mut b = SyncState::new(peer(2)); - let f = a.local_write(100, Op::Insert, "task", "t1", b"from-a".to_vec()); + let f = a + .local_write(100, Op::Insert, "task", "t1", b"from-a".to_vec()) + .expect("write"); assert_eq!(b.receive(f, 50), RecordOutcome::Appended); assert_eq!(b.visible("task", "t1"), Some(&b"from-a"[..])); } @@ -303,17 +431,23 @@ mod tests { fn receiving_advances_clock_past_remote() { let mut a = SyncState::new(peer(1)); let mut b = SyncState::new(peer(2)); - let f = a.local_write(1000, Op::Insert, "task", "t1", b"x".to_vec()); + let f = a + .local_write(1000, Op::Insert, "task", "t1", b"x".to_vec()) + .expect("write"); b.receive(f, 10); - let later = b.local_write(10, Op::Update, "task", "t1", b"y".to_vec()); + let later = b + .local_write(10, Op::Update, "task", "t1", b"y".to_vec()) + .expect("write"); assert!(later.stamp.hlc.physical >= 1000); } #[test] fn reclaim_waits_for_lagging_member_then_truncates() { let mut a = SyncState::new(peer(1)); - a.local_write(1, Op::Insert, "task", "t1", b"a".to_vec()); - a.local_write(2, Op::Insert, "task", "t2", b"b".to_vec()); + a.local_write(1, Op::Insert, "task", "t1", b"a".to_vec()) + .expect("write"); + a.local_write(2, Op::Insert, "task", "t2", b"b".to_vec()) + .expect("write"); // No cursor heard from peer 2 → low-water-mark 0 → nothing reclaimed. assert_eq!(a.reclaim(&[peer(2)]), 0); @@ -332,7 +466,9 @@ mod tests { assert_eq!(a.visible("task", "t1"), Some(&b"a"[..])); assert_eq!(a.visible("task", "t2"), Some(&b"b"[..])); // The clock/seq continuity survives reclamation. - let next = a.local_write(3, Op::Insert, "task", "t3", b"c".to_vec()); + let next = a + .local_write(3, Op::Insert, "task", "t3", b"c".to_vec()) + .expect("write"); assert_eq!(next.seq, 3); } @@ -340,7 +476,8 @@ mod tests { fn reclaim_truncates_only_to_the_slowest_member() { let mut a = SyncState::new(peer(1)); for i in 1..=3u64 { - a.local_write(i, Op::Insert, "task", format!("t{i}"), vec![i as u8]); + a.local_write(i, Op::Insert, "task", format!("t{i}"), vec![i as u8]) + .expect("write"); } let mut fast = Cursors::new(); fast.insert(peer(1), 3); diff --git a/crates/stitch-p2p/src/wire.rs b/crates/stitch-p2p/src/wire.rs index aaea230..b21daf5 100644 --- a/crates/stitch-p2p/src/wire.rs +++ b/crates/stitch-p2p/src/wire.rs @@ -21,6 +21,8 @@ pub enum WireError { EntityTooLong, #[error("id exceeds 255 bytes")] IdTooLong, + #[error("payload exceeds u32::MAX bytes")] + DataTooLong, #[error("entity or id is not valid UTF-8")] Utf8, #[error("invalid signature presence flag {0}")] @@ -60,7 +62,7 @@ impl WriteFrame { let id_bytes = self.id.as_bytes(); let entity_len = u8::try_from(entity_bytes.len()).map_err(|_| WireError::EntityTooLong)?; let id_len = u8::try_from(id_bytes.len()).map_err(|_| WireError::IdTooLong)?; - let data_len = u32::try_from(self.data.len()).expect("payload exceeds u32"); + let data_len = u32::try_from(self.data.len()).map_err(|_| WireError::DataTooLong)?; let mut out = Vec::with_capacity(HEADER_LEN + entity_bytes.len() + id_bytes.len() + self.data.len()); @@ -174,6 +176,15 @@ impl WriteFrame { } } +/// Check that a write with these fields fits the frame format, so a frame +/// built from them always encodes. +pub fn check_lengths(entity: &str, id: &str, data: &[u8]) -> Result<(), WireError> { + u8::try_from(entity.len()).map_err(|_| WireError::EntityTooLong)?; + u8::try_from(id.len()).map_err(|_| WireError::IdTooLong)?; + u32::try_from(data.len()).map_err(|_| WireError::DataTooLong)?; + Ok(()) +} + fn op_to_u8(op: Op) -> u8 { match op { Op::Insert => OP_INSERT, diff --git a/crates/stitch-p2p/tests/discovery_broker.rs b/crates/stitch-p2p/tests/discovery_broker.rs index d7d8479..6a4a21f 100644 --- a/crates/stitch-p2p/tests/discovery_broker.rs +++ b/crates/stitch-p2p/tests/discovery_broker.rs @@ -132,10 +132,12 @@ async fn discover_and_sync(spawn: impl Fn(Arc, SyncNode) -> Swarm) { node_a .local_write(Op::Insert, "task", "t1", b"from-alice".to_vec()) - .await; + .await + .expect("write"); node_b .local_write(Op::Insert, "task", "t2", b"from-bob".to_vec()) - .await; + .await + .expect("write"); let converged = wait_until(Duration::from_secs(30), || async { node_a.visible("task", "t2").await.is_some() && node_b.visible("task", "t1").await.is_some() diff --git a/crates/stitch-p2p/tests/membership_sync.rs b/crates/stitch-p2p/tests/membership_sync.rs index d968989..c413c5c 100644 --- a/crates/stitch-p2p/tests/membership_sync.rs +++ b/crates/stitch-p2p/tests/membership_sync.rs @@ -59,7 +59,7 @@ async fn member_writes_visible_outsider_filtered() { let outsider = Store::join(outsider_id, owner_pid); // Owner invites the member (not the outsider). - owner.invite(member_pid, Role::Member).await; + owner.invite(member_pid, Role::Member).await.expect("write"); // Each writes a record under its own identity. member @@ -112,7 +112,7 @@ async fn revocation_hides_records() { let owner = Store::with_owner(owner_id); let member = Store::join(member_id, owner_pid); - owner.invite(member_pid, Role::Member).await; + owner.invite(member_pid, Role::Member).await.expect("write"); member.create("task", "m", json!({"x": 1})).await.unwrap(); let (a_io, b_io) = tokio::io::duplex(64 * 1024); @@ -126,7 +126,7 @@ async fn revocation_hides_records() { "member write should be visible before revocation" ); - owner.revoke(member_pid).await; + owner.revoke(member_pid).await.expect("write"); assert!( wait_until(Duration::from_secs(5), || async { diff --git a/crates/stitch-p2p/tests/node_mesh.rs b/crates/stitch-p2p/tests/node_mesh.rs index 3928206..8a9cc13 100644 --- a/crates/stitch-p2p/tests/node_mesh.rs +++ b/crates/stitch-p2p/tests/node_mesh.rs @@ -62,11 +62,14 @@ async fn line_topology_converges_through_hub() { assert_eq!(b.session_count(), 2, "hub B should hold two sessions"); a.local_write(Op::Insert, "task", "t1", b"from-a".to_vec()) - .await; + .await + .expect("write"); b.local_write(Op::Insert, "task", "t2", b"from-b".to_vec()) - .await; + .await + .expect("write"); c.local_write(Op::Insert, "task", "t3", b"from-c".to_vec()) - .await; + .await + .expect("write"); let converged = wait_until(Duration::from_secs(5), || async { a.visible("task", "t3").await.is_some() && c.visible("task", "t1").await.is_some() @@ -101,7 +104,8 @@ async fn write_after_connect_reaches_far_end() { // Write at A only after the mesh is live; it must reach the far end C. a.local_write(Op::Insert, "task", "late", b"v".to_vec()) - .await; + .await + .expect("write"); let reached = wait_until(Duration::from_secs(5), || async { c.visible("task", "late").await.is_some() diff --git a/crates/stitch-p2p/tests/persistence.rs b/crates/stitch-p2p/tests/persistence.rs index aeadbb7..12fc44d 100644 --- a/crates/stitch-p2p/tests/persistence.rs +++ b/crates/stitch-p2p/tests/persistence.rs @@ -4,7 +4,7 @@ //! advancing so post-reopen writes still win. use serde_json::json; -use stitch_p2p::{PEER_ID_LEN, PeerId, Store}; +use stitch_p2p::{PEER_ID_LEN, PeerId, Store, StoreError, WireError, WriteError}; use tempfile::TempDir; fn peer(n: u8) -> PeerId { @@ -28,7 +28,7 @@ async fn state_survives_reopen() { .create("task", "t2", json!({"title": "beta"})) .await .unwrap(); - store.delete("task", "t2").await; + store.delete("task", "t2").await.expect("write"); assert_eq!( store.read("task", "t1").await, Some(json!({"title": "alpha"})) @@ -72,3 +72,40 @@ async fn post_reopen_write_wins_over_prior_state() { let reopened = Store::open(peer(1), &path).unwrap(); assert_eq!(reopened.read("doc", "d1").await, Some(json!({"v": 2}))); } + +#[tokio::test] +async fn refused_write_leaves_no_hole_in_the_log() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("log"); + + { + let store = Store::open(peer(1), &path).unwrap(); + let refused = store + .create("task", &"L".repeat(300), json!({"title": "long"})) + .await; + assert!( + matches!( + refused, + Err(StoreError::Write(WriteError::Wire(WireError::IdTooLong))) + ), + "{refused:?}" + ); + store + .create("task", "t2", json!({"title": "beta"})) + .await + .unwrap(); + } + + let store = Store::open(peer(1), &path).unwrap(); + assert_eq!( + store.read("task", "t2").await, + Some(json!({"title": "beta"})), + "the write after the refused one replays" + ); + let next = store + .node() + .local_write(stitch_p2p::Op::Insert, "task", "t3", b"{}".to_vec()) + .await + .unwrap(); + assert_eq!(next.seq, 2, "sequence continues after the replayed write"); +} diff --git a/crates/stitch-p2p/tests/quic_loopback.rs b/crates/stitch-p2p/tests/quic_loopback.rs index 9da14e3..7be11ae 100644 --- a/crates/stitch-p2p/tests/quic_loopback.rs +++ b/crates/stitch-p2p/tests/quic_loopback.rs @@ -35,11 +35,13 @@ async fn two_peers_converge_over_real_quic() { a_state .lock() .await - .local_write(10, Op::Insert, "task", "t1", b"from-a".to_vec()); + .local_write(10, Op::Insert, "task", "t1", b"from-a".to_vec()) + .expect("write"); b_state .lock() .await - .local_write(11, Op::Insert, "task", "t2", b"from-b".to_vec()); + .local_write(11, Op::Insert, "task", "t2", b"from-b".to_vec()) + .expect("write"); // Acceptor B: accept connection (verifying A's fingerprint), then the bidi // stream A opens once it writes its Hello. diff --git a/crates/stitch-p2p/tests/transitive.rs b/crates/stitch-p2p/tests/transitive.rs index aaf4e70..e7f91d2 100644 --- a/crates/stitch-p2p/tests/transitive.rs +++ b/crates/stitch-p2p/tests/transitive.rs @@ -23,9 +23,12 @@ fn line_topology_converges_transitively() { let mut p2 = SyncState::new(peer(2)); let mut p3 = SyncState::new(peer(3)); - p1.local_write(10, Op::Insert, "task", "t1", b"from1".to_vec()); - p2.local_write(11, Op::Insert, "task", "t2", b"from2".to_vec()); - p3.local_write(12, Op::Insert, "task", "t3", b"from3".to_vec()); + p1.local_write(10, Op::Insert, "task", "t1", b"from1".to_vec()) + .expect("write"); + p2.local_write(11, Op::Insert, "task", "t2", b"from2".to_vec()) + .expect("write"); + p3.local_write(12, Op::Insert, "task", "t3", b"from3".to_vec()) + .expect("write"); // Gossip only along the links 1—2 and 2—3. Two rounds is enough for a // 3-node line: round 1 fills peer 2, round 2 spreads to the ends. @@ -54,8 +57,10 @@ fn concurrent_same_record_edits_converge_across_line() { // p1 and p3 both write "shared" at the same wall time, never having seen // each other. Peer-id tiebreak must pick the same winner everywhere. - p1.local_write(50, Op::Insert, "task", "shared", b"by1".to_vec()); - p3.local_write(50, Op::Insert, "task", "shared", b"by3".to_vec()); + p1.local_write(50, Op::Insert, "task", "shared", b"by1".to_vec()) + .expect("write"); + p3.local_write(50, Op::Insert, "task", "shared", b"by3".to_vec()) + .expect("write"); for round in 0..3 { let wall = 200 + round; @@ -81,7 +86,8 @@ fn delete_propagates_across_line() { let mut p2 = SyncState::new(peer(2)); let mut p3 = SyncState::new(peer(3)); - p1.local_write(10, Op::Insert, "task", "t1", b"alive".to_vec()); + p1.local_write(10, Op::Insert, "task", "t1", b"alive".to_vec()) + .expect("write"); for round in 0..2 { pull(&mut p2, &p1, 100 + round); pull(&mut p3, &p2, 100 + round); @@ -89,7 +95,8 @@ fn delete_propagates_across_line() { assert_eq!(p3.visible("task", "t1"), Some(&b"alive"[..])); // p3 deletes; the tombstone travels back up the line to p1. - p3.local_write(200, Op::Delete, "task", "t1", Vec::new()); + p3.local_write(200, Op::Delete, "task", "t1", Vec::new()) + .expect("write"); for round in 0..2 { pull(&mut p2, &p3, 300 + round); pull(&mut p1, &p2, 300 + round); diff --git a/crates/stitch-tasks/src/bin/demo.rs b/crates/stitch-tasks/src/bin/demo.rs index 8122a2b..db8e09c 100644 --- a/crates/stitch-tasks/src/bin/demo.rs +++ b/crates/stitch-tasks/src/bin/demo.rs @@ -122,7 +122,9 @@ async fn run_peer(args: &[String]) { for member in PEERS { if member != OWNER { let id = Identity::from_seed(seed_from_name(member)).peer_id(); - board.store().invite(id, Role::Member).await; + if let Err(error) = board.store().invite(id, Role::Member).await { + emit(&format!("ERR {name} invite {member}: {error}")); + } } } } @@ -165,11 +167,17 @@ async fn run_peer(args: &[String]) { let cmd = parts.next().unwrap_or(""); let id = parts.next().unwrap_or(""); let rest = parts.next().unwrap_or(""); + let written = match cmd { + "add" => Some(board.add(id, rest).await), + "rename" => Some(board.rename(id, rest).await), + "done" => Some(board.set_done(id, rest == "true").await), + "remove" => Some(board.remove(id).await), + _ => None, + }; + if let Some(Err(error)) = written { + emit(&format!("ERR {name} {cmd} {id}: {error}")); + } match cmd { - "add" => board.add(id, rest).await, - "rename" => board.rename(id, rest).await, - "done" => board.set_done(id, rest == "true").await, - "remove" => board.remove(id).await, "offline" => { if let Some((peer, swarm)) = link.take() { disconnect(peer, swarm).await; diff --git a/crates/stitch-tasks/src/bin/soak.rs b/crates/stitch-tasks/src/bin/soak.rs index 358505a..4aa3098 100644 --- a/crates/stitch-tasks/src/bin/soak.rs +++ b/crates/stitch-tasks/src/bin/soak.rs @@ -27,7 +27,13 @@ async fn main() { cfg.peers, cfg.rounds, cfg.seed ); let start = Instant::now(); - let (_, report) = run_chaos(cfg).await; + let report = match run_chaos(cfg).await { + Ok((_, report)) => report, + Err(error) => { + eprintln!(" write failed: {error}"); + std::process::exit(1); + } + }; let elapsed = start.elapsed(); println!( diff --git a/crates/stitch-tasks/src/harness.rs b/crates/stitch-tasks/src/harness.rs index 0569fe1..695106a 100644 --- a/crates/stitch-tasks/src/harness.rs +++ b/crates/stitch-tasks/src/harness.rs @@ -12,7 +12,7 @@ use rand::{RngExt, SeedableRng}; use std::collections::HashMap; use std::time::Duration; use stitch_p2p::membership::Role; -use stitch_p2p::{Identity, PeerId, Store, session}; +use stitch_p2p::{Identity, PeerId, Store, StoreError, session}; use tokio::io::split; use tokio::task::JoinHandle; @@ -72,18 +72,19 @@ impl Cluster { } /// Owner invites every other peer as a member. - pub async fn invite_all(&self) { + pub async fn invite_all(&self) -> Result<(), StoreError> { for i in 1..self.boards.len() { self.boards[0] .store() .invite(self.ids[i], Role::Member) - .await; + .await?; } + Ok(()) } /// Owner revokes peer `i`. - pub async fn revoke(&self, i: usize) { - self.boards[0].store().revoke(self.ids[i]).await; + pub async fn revoke(&self, i: usize) -> Result<(), StoreError> { + self.boards[0].store().revoke(self.ids[i]).await } /// Establish a bidirectional session between peers `a` and `b` (idempotent). @@ -220,10 +221,10 @@ pub struct ChaosReport { /// Run a randomized, partition-and-membership-churning workload against a fresh /// cluster, then heal, restore membership, and wait for convergence. Returns a /// report; `report.converged` is the invariant a caller asserts. -pub async fn run_chaos(cfg: Chaos) -> (Chaos, ChaosReport) { +pub async fn run_chaos(cfg: Chaos) -> Result<(Chaos, ChaosReport), StoreError> { let mut cluster = Cluster::new(cfg.peers, cfg.pull); cluster.full_mesh(); - cluster.invite_all().await; + cluster.invite_all().await?; let mut rng = StdRng::seed_from_u64(cfg.seed); let ids: Vec = (0..cfg.id_pool).map(|i| format!("t{i}")).collect(); @@ -235,21 +236,23 @@ pub async fn run_chaos(cfg: Chaos) -> (Chaos, ChaosReport) { let id = ids[rng.random_range(0..cfg.id_pool)].clone(); match rng.random_range(0..4u8) { 0 => { - board.add(&id, &format!("t-{}", rng.random::())).await; + board + .add(&id, &format!("t-{}", rng.random::())) + .await?; report.adds += 1; } 1 => { board .rename(&id, &format!("r-{}", rng.random::())) - .await; + .await?; report.renames += 1; } 2 => { - board.set_done(&id, rng.random::()).await; + board.set_done(&id, rng.random::()).await?; report.toggles += 1; } _ => { - board.remove(&id).await; + board.remove(&id).await?; report.removes += 1; } } @@ -269,7 +272,7 @@ pub async fn run_chaos(cfg: Chaos) -> (Chaos, ChaosReport) { } if cfg.peers > 1 && rng.random_range(0..40u8) == 0 { - cluster.revoke(rng.random_range(1..cfg.peers)).await; + cluster.revoke(rng.random_range(1..cfg.peers)).await?; report.revokes += 1; } @@ -277,8 +280,8 @@ pub async fn run_chaos(cfg: Chaos) -> (Chaos, ChaosReport) { } cluster.full_mesh(); - cluster.invite_all().await; + cluster.invite_all().await?; report.converged = cluster.await_convergence(Duration::from_secs(20)).await; report.final_tasks = cluster.board(0).len().await; - (cfg, report) + Ok((cfg, report)) } diff --git a/crates/stitch-tasks/src/lib.rs b/crates/stitch-tasks/src/lib.rs index 5a14dec..3e312cd 100644 --- a/crates/stitch-tasks/src/lib.rs +++ b/crates/stitch-tasks/src/lib.rs @@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize}; use serde_json::json; use std::collections::BTreeMap; -use stitch_p2p::Store; +use stitch_p2p::{Store, StoreError}; pub mod harness; @@ -41,27 +41,26 @@ impl TaskBoard { /// Add a task under `id` (caller supplies the id so the harness can pick /// collision-prone ids deliberately). - pub async fn add(&self, id: &str, title: &str) { - let _ = self - .store + pub async fn add(&self, id: &str, title: &str) -> Result<(), StoreError> { + self.store .create(ENTITY, id, json!({ "title": title, "done": false })) - .await; + .await } - pub async fn rename(&self, id: &str, title: &str) { + pub async fn rename(&self, id: &str, title: &str) -> Result<(), StoreError> { let mut fields = serde_json::Map::new(); fields.insert("title".into(), json!(title)); - let _ = self.store.update(ENTITY, id, fields).await; + self.store.update(ENTITY, id, fields).await } - pub async fn set_done(&self, id: &str, done: bool) { + pub async fn set_done(&self, id: &str, done: bool) -> Result<(), StoreError> { let mut fields = serde_json::Map::new(); fields.insert("done".into(), json!(done)); - let _ = self.store.update(ENTITY, id, fields).await; + self.store.update(ENTITY, id, fields).await } - pub async fn remove(&self, id: &str) { - self.store.delete(ENTITY, id).await; + pub async fn remove(&self, id: &str) -> Result<(), StoreError> { + self.store.delete(ENTITY, id).await } #[must_use] @@ -107,8 +106,8 @@ mod tests { #[tokio::test] async fn add_get_snapshot() { let board = TaskBoard::new(Store::new(peer(1))); - board.add("t1", "write tests").await; - board.add("t2", "ship it").await; + board.add("t1", "write tests").await.expect("write"); + board.add("t2", "ship it").await.expect("write"); assert_eq!( board.get("t1").await, Some(Task { @@ -122,9 +121,9 @@ mod tests { #[tokio::test] async fn rename_toggle_remove() { let board = TaskBoard::new(Store::new(peer(1))); - board.add("t1", "draft").await; - board.rename("t1", "final").await; - board.set_done("t1", true).await; + board.add("t1", "draft").await.expect("write"); + board.rename("t1", "final").await.expect("write"); + board.set_done("t1", true).await.expect("write"); assert_eq!( board.get("t1").await, Some(Task { @@ -132,7 +131,7 @@ mod tests { done: true }) ); - board.remove("t1").await; + board.remove("t1").await.expect("write"); assert_eq!(board.get("t1").await, None); assert!(board.is_empty().await); } diff --git a/crates/stitch-tasks/tests/soak.rs b/crates/stitch-tasks/tests/soak.rs index 467a370..925db31 100644 --- a/crates/stitch-tasks/tests/soak.rs +++ b/crates/stitch-tasks/tests/soak.rs @@ -14,7 +14,8 @@ async fn soak_converges_across_seeds() { seed, ..Chaos::default() }) - .await; + .await + .expect("chaos run writes"); assert!(report.converged, "peers failed to converge (seed {seed})"); assert!( report.final_tasks > 0,