feat(smg): outbound worker mesh sync — single-writer ownership, publish loop, tombstones, recovery

crates/mesh/src/kv.rs

@@ -531,6 +531,20 @@ impl MeshKV {
         self.chunk_assembler.clone()
     }
 
+    /// Reclaim tombstone metadata older than the default grace period
+    /// across every CRDT engine. Returns the number reclaimed.
+    ///
+    /// Deliberately not driven anywhere: time-based collection is unsound.
+    /// A peer absent longer than the grace still holds the deleted key's
+    /// older insert in its op-log and replays it on reconnect; with the
+    /// tombstone metadata gone, the stale insert lands on an empty entry
+    /// and resurrects the key. Collection is only safe at causal stability
+    /// — every live peer acked past the tombstone and absent peers
+    /// declared dead — so the caller must be the dead-node cleanup layer.
+    pub fn gc_tombstones(&self) -> usize {
+        self.store.gc_tombstones()
+    }
+
     /// Fire subscribers whose prefix matches `key`. Used by the gossip
     /// receive path when a chunked value completes (or a single-chunk
     /// entry arrives), so handlers can deliver into adapter-owned

crates/mesh/src/tests/crdt_integration.rs

@@ -189,6 +189,22 @@ async fn remote_tombstone_after_insert_notifies_none() {
     assert_eq!(r_ns.get("worker:a"), None);
 }
 
+#[test]
+fn gc_tombstones_respects_grace_through_mesh_kv() {
+    // The controller drives this periodically; a fresh tombstone must
+    // survive the grace period (engine-level reclamation is covered by
+    // the crdt_kv tests).
+    let mesh = MeshKV::new("node-a".into());
+    let ns = mesh.configure_crdt_prefix("worker:", MergeStrategy::LastWriterWins);
+    ns.put("worker:a", b"v".to_vec());
+    ns.delete("worker:a");
+    assert_eq!(
+        mesh.gc_tombstones(),
+        0,
+        "a fresh tombstone survives the grace period"
+    );
+}
+
 #[test]
 fn caught_up_peer_is_sent_nothing() {
     let sender = MeshKV::new("sender".to_string());

crates/mesh/src/types.rs

@@ -5,7 +5,7 @@
 use serde::{Deserialize, Serialize};
 
 /// Worker state entry synced across mesh nodes. `spec` is an
-/// opaque bincode-serialized `WorkerSpec`; the mesh crate
+/// opaque JSON-serialized `WorkerSpec`; the mesh crate
 /// doesn't interpret it.
 ///
 /// `Eq`/`Hash` are intentionally omitted: `load: f64` can be
@@ -18,9 +18,10 @@ pub struct WorkerState {
     pub health: bool,
     pub load: f64,
     pub version: u64,
-    /// Opaque worker specification (bincode-serialized
-    /// `WorkerSpec` from the gateway). Empty on old nodes that
-    /// don't populate this field.
+    /// Opaque worker specification (JSON-serialized `WorkerSpec`
+    /// from the gateway; JSON because the type's serde-skip
+    /// attributes don't round-trip positional formats). Empty on
+    /// old nodes that don't populate this field.
     #[serde(default)]
     pub spec: Vec<u8>,
 }