refuse unencodable or unpersisted writes and count live sessions

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"

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

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};
 

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));
     }

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<dyn crate::sync_state::FramePersister>,
-    ) -> Self {
+    ) -> std::io::Result<Self> {
         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<String>,
         id: impl Into<String>,
         data: Vec<u8>,
-    ) -> WriteFrame {
+    ) -> Result<WriteFrame, WriteError> {
         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<Vec<WriteFrame>> {
+            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());
     }
 }

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<WriteFrame> {
-        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<Vec<WriteFrame>> {
+        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()
     }
 }

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]

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<SyncState>, frames: Vec<WriteFrame>) {
     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 {