Add gossip replication between nodes

Nodes run anti-entropy over a TCP link: each advertises its version
vector, the peer replies with the log entries it lacks, and ingest is
idempotent so re-delivery and reconnect after a split both converge.
The engine is shared behind a mutex; config gains [gossip] and [[peer]].

Cargo.toml

@@ -5,7 +5,7 @@ edition = "2021"
 description = "Federated IRC services daemon (protocol-agnostic core, InspIRCd link first)"
 
 [dependencies]
-tokio = { version = "1", features = ["net", "io-util", "rt-multi-thread", "macros", "time"] }
+tokio = { version = "1", features = ["net", "io-util", "rt-multi-thread", "macros", "time", "sync"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 toml = "0.8"

src/config.rs

@@ -4,6 +4,24 @@ use serde::Deserialize;
 pub struct Config {
     pub uplink: Uplink,
     pub server: Server,
+    // Node-to-node replication. Absent = single node, no gossip.
+    #[serde(default)]
+    pub gossip: Option<Gossip>,
+    #[serde(default)]
+    pub peer: Vec<Peer>,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Gossip {
+    // Address to accept peer connections on. Absent = dial-only node.
+    pub bind: Option<String>,
+    // Shared secret both nodes must present.
+    pub secret: String,
+}
+
+#[derive(Debug, Deserialize, Clone)]
+pub struct Peer {
+    pub addr: String,
 }
 
 #[derive(Debug, Deserialize)]

src/engine/db.rs

@@ -65,26 +65,31 @@ pub struct EventLog {
     origin: String,
     lamport: u64,                   // logical clock, ticked on every event
     versions: HashMap<String, u64>, // per-origin highest seq applied (version vector)
+    entries: Vec<LogEntry>,         // full log, kept so peers can pull what they lack
 }
 
 impl EventLog {
     fn open(path: PathBuf, origin: String) -> (Self, Vec<Event>) {
-        let mut events = Vec::new();
-        let mut lamport = 0;
-        let mut versions: HashMap<String, u64> = HashMap::new();
-        if let Ok(data) = std::fs::read_to_string(&path) {
+        let mut log = Self { path, origin, lamport: 0, versions: HashMap::new(), entries: Vec::new() };
+        if let Ok(data) = std::fs::read_to_string(&log.path) {
             for line in data.lines().filter(|l| !l.trim().is_empty()) {
                 match serde_json::from_str::<LogEntry>(line) {
                     Ok(entry) => {
-                        lamport = lamport.max(entry.lamport);
-                        versions.entry(entry.origin.clone()).and_modify(|s| *s = (*s).max(entry.seq)).or_insert(entry.seq);
-                        events.push(entry.event);
+                        log.absorb(&entry);
+                        log.entries.push(entry);
                     }
                     Err(e) => tracing::warn!(%e, "skipping malformed event log line"),
                 }
             }
         }
-        (Self { path, origin, lamport, versions }, events)
+        let events = log.entries.iter().map(|e| e.event.clone()).collect();
+        (log, events)
+    }
+
+    // Roll the clock and version vector forward over an entry.
+    fn absorb(&mut self, entry: &LogEntry) {
+        self.lamport = self.lamport.max(entry.lamport);
+        self.versions.entry(entry.origin.clone()).and_modify(|s| *s = (*s).max(entry.seq)).or_insert(entry.seq);
     }
 
     // Seq the next locally-authored event will carry (0-based, per our origin).
@@ -99,13 +104,13 @@ impl EventLog {
         let entry = LogEntry { origin: self.origin.clone(), seq: self.next_seq(), lamport: self.lamport, event };
         self.persist(&entry)?;
         self.versions.insert(entry.origin.clone(), entry.seq);
+        self.entries.push(entry);
         Ok(())
     }
 
     // Ingest an entry authored by another node — the gossip seam. Returns the
     // event to fold into state, or None if already applied (idempotent, so
     // re-delivery converges). Assumes per-origin in-order delivery.
-    #[allow(dead_code)]
     fn ingest(&mut self, entry: LogEntry) -> std::io::Result<Option<Event>> {
         if self.versions.get(&entry.origin).is_some_and(|&s| entry.seq <= s) {
             return Ok(None); // already have it
@@ -113,7 +118,23 @@ impl EventLog {
         self.persist(&entry)?;
         self.lamport = self.lamport.max(entry.lamport) + 1; // Lamport receive rule
         self.versions.insert(entry.origin.clone(), entry.seq);
-        Ok(Some(entry.event))
+        let event = entry.event.clone();
+        self.entries.push(entry);
+        Ok(Some(event))
+    }
+
+    // Our version vector: highest seq applied per origin.
+    fn version_vector(&self) -> HashMap<String, u64> {
+        self.versions.clone()
+    }
+
+    // Entries a peer is missing, given the version vector it advertised.
+    fn missing_for(&self, peer: &HashMap<String, u64>) -> Vec<LogEntry> {
+        self.entries
+            .iter()
+            .filter(|e| peer.get(&e.origin).map_or(true, |&s| e.seq > s))
+            .cloned()
+            .collect()
     }
 
     fn persist(&self, entry: &LogEntry) -> std::io::Result<()> {
@@ -180,7 +201,6 @@ impl Db {
 
     /// Fold an entry authored by another node into the store — the services-side
     /// of the gossip seam. Idempotent (re-delivered entries are dropped).
-    #[allow(dead_code)]
     pub fn ingest(&mut self, entry: LogEntry) -> std::io::Result<()> {
         if let Some(event) = self.log.ingest(entry)? {
             apply(&mut self.accounts, event);
@@ -188,6 +208,16 @@ impl Db {
         Ok(())
     }
 
+    /// Our version vector, advertised to peers so they can send what we lack.
+    pub fn version_vector(&self) -> HashMap<String, u64> {
+        self.log.version_vector()
+    }
+
+    /// The log entries a peer is missing, given the version vector it sent.
+    pub fn missing_for(&self, peer: &HashMap<String, u64>) -> Vec<LogEntry> {
+        self.log.missing_for(peer)
+    }
+
     pub fn exists(&self, name: &str) -> bool {
         self.accounts.contains_key(&key(name))
     }

src/engine/mod.rs

@@ -9,7 +9,7 @@ use std::time::{Duration, Instant};
 use base64::{engine::general_purpose::STANDARD, Engine as _};
 
 use crate::proto::{NetAction, NetEvent, RegReply};
-use db::{Db, RegError};
+use db::{Db, LogEntry, RegError};
 use scram::Verifier;
 use service::{Sender, Service, ServiceCtx};
 use state::Network;
@@ -91,6 +91,29 @@ impl Engine {
         self.db.scram_iterations
     }
 
+    // Gossip pass-throughs to the account store, used by the replication layer.
+    pub fn gossip_digest(&self) -> HashMap<String, u64> {
+        self.db.version_vector()
+    }
+
+    pub fn gossip_missing(&self, peer: &HashMap<String, u64>) -> Vec<LogEntry> {
+        self.db.missing_for(peer)
+    }
+
+    pub fn gossip_ingest(&mut self, entry: LogEntry) -> std::io::Result<()> {
+        self.db.ingest(entry)
+    }
+
+    #[cfg(test)]
+    pub(crate) fn test_register(&mut self, name: &str) {
+        self.db.register(name, "pw", None).unwrap();
+    }
+
+    #[cfg(test)]
+    pub(crate) fn test_has_account(&self, name: &str) -> bool {
+        self.db.exists(name)
+    }
+
     // Insert or refresh a client's in-progress SASL session, stamped now.
     fn stash_sasl(&mut self, client: String, session: SaslSession) {
         self.sasl_sessions.insert(client, TimedSession { touched: Instant::now(), session });

src/gossip.rs

@@ -0,0 +1,212 @@
+// Node-to-node account replication. Peers run anti-entropy: each side advertises
+// its version vector and the other replies with the log entries it is missing.
+// Ingest is idempotent, so re-delivery and reconnect after a split both converge.
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader};
+use tokio::net::{TcpListener, TcpStream};
+use tokio::sync::{mpsc, Mutex};
+
+use crate::config::{Gossip, Peer};
+use crate::engine::db::LogEntry;
+use crate::engine::Engine;
+
+type Shared = Arc<Mutex<Engine>>;
+
+const TICK: Duration = Duration::from_secs(3); // how often we re-advertise our vector
+const REDIAL: Duration = Duration::from_secs(5); // reconnect backoff for dialed peers
+
+// Wire protocol: one JSON object per line.
+#[derive(Serialize, Deserialize)]
+#[serde(tag = "t", rename_all = "lowercase")]
+enum Msg {
+    Hello { secret: String, origin: String },
+    Digest { versions: HashMap<String, u64> },
+    Entry { entry: LogEntry },
+}
+
+// Start the listener (if bound) and a dialer per configured peer.
+pub async fn run(engine: Shared, cfg: Gossip, peers: Vec<Peer>, origin: String) {
+    if let Some(bind) = cfg.bind.clone() {
+        tokio::spawn(listen(bind, engine.clone(), cfg.secret.clone(), origin.clone()));
+    }
+    for peer in peers {
+        tokio::spawn(dial(peer.addr, engine.clone(), cfg.secret.clone(), origin.clone()));
+    }
+}
+
+async fn listen(bind: String, engine: Shared, secret: String, origin: String) {
+    let listener = match TcpListener::bind(&bind).await {
+        Ok(l) => l,
+        Err(e) => return tracing::error!(%e, %bind, "gossip bind failed"),
+    };
+    tracing::info!(%bind, "gossip listening");
+    loop {
+        if let Ok((stream, addr)) = listener.accept().await {
+            tracing::info!(%addr, "gossip peer accepted");
+            let (engine, secret, origin) = (engine.clone(), secret.clone(), origin.clone());
+            tokio::spawn(async move {
+                if let Err(e) = session(stream, engine, secret, origin).await {
+                    tracing::debug!(%e, "gossip session ended");
+                }
+            });
+        }
+    }
+}
+
+async fn dial(addr: String, engine: Shared, secret: String, origin: String) {
+    loop {
+        match TcpStream::connect(&addr).await {
+            Ok(stream) => {
+                tracing::info!(%addr, "gossip dialed peer");
+                if let Err(e) = session(stream, engine.clone(), secret.clone(), origin.clone()).await {
+                    tracing::debug!(%e, %addr, "gossip session ended");
+                }
+            }
+            Err(e) => tracing::debug!(%e, %addr, "gossip dial failed"),
+        }
+        tokio::time::sleep(REDIAL).await;
+    }
+}
+
+// One peer connection: authenticate, then run anti-entropy until it drops.
+async fn session<S>(stream: S, engine: Shared, secret: String, origin: String) -> anyhow::Result<()>
+where
+    S: AsyncRead + AsyncWrite + Send + 'static,
+{
+    let (read, mut write) = tokio::io::split(stream);
+    let mut reader = BufReader::new(read).lines();
+
+    // A single writer task owns the socket; the reader and the ticker feed it.
+    let (tx, mut rx) = mpsc::channel::<String>(1024);
+    let writer = tokio::spawn(async move {
+        while let Some(line) = rx.recv().await {
+            if write.write_all(line.as_bytes()).await.is_err() || write.write_all(b"\n").await.is_err() {
+                break;
+            }
+        }
+    });
+
+    // Handshake: send our hello, require a matching secret back.
+    let _ = send(&tx, &Msg::Hello { secret: secret.clone(), origin }).await;
+    match reader.next_line().await? {
+        Some(line) => match serde_json::from_str::<Msg>(&line) {
+            Ok(Msg::Hello { secret: s, .. }) if s == secret => {}
+            _ => {
+                writer.abort();
+                anyhow::bail!("bad gossip handshake");
+            }
+        },
+        None => {
+            writer.abort();
+            anyhow::bail!("peer closed before handshake");
+        }
+    }
+
+    // Re-advertise our version vector on a timer so new local events propagate.
+    let ticker = {
+        let (tx, engine) = (tx.clone(), engine.clone());
+        tokio::spawn(async move {
+            loop {
+                let versions = engine.lock().await.gossip_digest();
+                if send(&tx, &Msg::Digest { versions }).await.is_err() {
+                    break;
+                }
+                tokio::time::sleep(TICK).await;
+            }
+        })
+    };
+
+    // Answer a peer's digest with what it lacks; apply the entries it sends us.
+    let result = loop {
+        match reader.next_line().await {
+            Ok(Some(line)) => match serde_json::from_str::<Msg>(&line) {
+                Ok(Msg::Digest { versions }) => {
+                    let missing = engine.lock().await.gossip_missing(&versions);
+                    for entry in missing {
+                        let _ = send(&tx, &Msg::Entry { entry }).await;
+                    }
+                }
+                Ok(Msg::Entry { entry }) => {
+                    if let Err(e) = engine.lock().await.gossip_ingest(entry) {
+                        tracing::warn!(%e, "gossip ingest failed");
+                    }
+                }
+                _ => {}
+            },
+            Ok(None) => break Ok(()),
+            Err(e) => break Err(e.into()),
+        }
+    };
+
+    ticker.abort();
+    writer.abort();
+    result
+}
+
+async fn send(tx: &mpsc::Sender<String>, msg: &Msg) -> Result<(), ()> {
+    let line = serde_json::to_string(msg).map_err(|_| ())?;
+    tx.send(line).await.map_err(|_| ())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::engine::db::Db;
+    use crate::services::nickserv::NickServ;
+
+    fn engine(origin: &str, tag: &str) -> Shared {
+        let path = std::env::temp_dir().join(format!("fedserv-gossip-{tag}.jsonl"));
+        let _ = std::fs::remove_file(&path);
+        let mut db = Db::open(&path, origin);
+        db.scram_iterations = 4096;
+        let ns = NickServ { uid: format!("{origin}AAAAAA"), guest_nick: "Guest".into(), guest_seq: 0 };
+        Arc::new(Mutex::new(Engine::new(vec![Box::new(ns)], db)))
+    }
+
+    // An account registered on node A reaches node B over a gossip link.
+    #[tokio::test]
+    async fn two_nodes_converge_over_a_link() {
+        let a = engine("A", "conv-a");
+        let b = engine("B", "conv-b");
+        a.lock().await.test_register("alice");
+
+        let (ca, cb) = tokio::io::duplex(64 * 1024);
+        let sa = tokio::spawn(session(ca, a.clone(), "s3cret".into(), "A".into()));
+        let sb = tokio::spawn(session(cb, b.clone(), "s3cret".into(), "B".into()));
+
+        let mut converged = false;
+        for _ in 0..100 {
+            if b.lock().await.test_has_account("alice") {
+                converged = true;
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+        sa.abort();
+        sb.abort();
+        assert!(converged, "B should receive A's account over gossip");
+    }
+
+    // A wrong secret is rejected before any state is exchanged.
+    #[tokio::test]
+    async fn bad_secret_is_rejected() {
+        let a = engine("A", "auth-a");
+        let b = engine("B", "auth-b");
+        a.lock().await.test_register("alice");
+
+        let (ca, cb) = tokio::io::duplex(64 * 1024);
+        let sa = tokio::spawn(session(ca, a.clone(), "right".into(), "A".into()));
+        let sb = tokio::spawn(session(cb, b.clone(), "wrong".into(), "B".into()));
+
+        let _ = tokio::time::timeout(Duration::from_secs(1), async {
+            let _ = sa.await;
+            let _ = sb.await;
+        })
+        .await;
+        assert!(!b.lock().await.test_has_account("alice"), "no state should cross a bad handshake");
+    }
+}

src/link.rs

@@ -1,13 +1,18 @@
+use std::sync::Arc;
+
 use anyhow::Result;
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::net::TcpStream;
+use tokio::sync::Mutex;
 
 use crate::engine::db::Db;
 use crate::engine::Engine;
 use crate::proto::{NetAction, Protocol};
 
 // One uplink session: connect, handshake + burst, then translate lines forever.
-pub async fn run(mut proto: Box<dyn Protocol>, mut engine: Engine, addr: &str) -> Result<()> {
+// The engine is shared with the gossip layer, so it is locked per operation and
+// never held across the registration key-stretching await.
+pub async fn run(mut proto: Box<dyn Protocol>, engine: Arc<Mutex<Engine>>, addr: &str) -> Result<()> {
     let stream = TcpStream::connect(addr).await?;
     let (read, mut write) = stream.into_split();
     let mut lines = BufReader::new(read).lines();
@@ -15,7 +20,8 @@ pub async fn run(mut proto: Box<dyn Protocol>, mut engine: Engine, addr: &str) -
     for line in proto.handshake() {
         send(&mut write, &line).await?;
     }
-    for action in engine.startup_actions() {
+    let startup = engine.lock().await.startup_actions();
+    for action in startup {
         for line in proto.serialize(&action) {
             send(&mut write, &line).await?;
         }
@@ -24,21 +30,23 @@ pub async fn run(mut proto: Box<dyn Protocol>, mut engine: Engine, addr: &str) -
     while let Some(line) = lines.next_line().await? {
         tracing::debug!(dir = "<<", %line);
         for event in proto.parse(&line) {
-            for action in engine.handle(event) {
+            let actions = engine.lock().await.handle(event);
+            for action in actions {
                 let outs = match action {
                     // Registration: derive the password off the reactor so the
                     // ~1s of key stretching can't stall the link, after a cheap
                     // gate that rejects taken names and rate-limits floods.
                     NetAction::DeferRegister { account, password, email, reply } => {
-                        match engine.pre_register_check(&account, &reply) {
+                        let pre = engine.lock().await.pre_register_check(&account, &reply);
+                        match pre {
                             Some(rejection) => rejection,
                             None => {
-                                let iterations = engine.scram_iterations();
+                                let iterations = engine.lock().await.scram_iterations();
                                 let creds = tokio::task::spawn_blocking(move || {
                                     Db::derive_credentials(&password, iterations)
                                 })
                                 .await?;
-                                engine.complete_register(&account, creds, email, reply)
+                                engine.lock().await.complete_register(&account, creds, email, reply)
                             }
                         }
                     }

src/main.rs

@@ -1,11 +1,14 @@
 mod config;
 mod engine;
+mod gossip;
 mod link;
 mod proto;
 mod services;
 
 use anyhow::Result;
+use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::sync::Mutex;
 
 use engine::Engine;
 use proto::inspircd::InspIrcd;
@@ -39,7 +42,12 @@ async fn main() -> Result<()> {
     })];
     let mut db = engine::db::Db::open("fedserv.db.jsonl", &cfg.server.sid);
     db.scram_iterations = cfg.server.scram_iterations;
-    let engine = Engine::new(services, db);
+    let engine = Arc::new(Mutex::new(Engine::new(services, db)));
+
+    if let Some(gossip) = cfg.gossip.clone() {
+        tracing::info!(peers = cfg.peer.len(), "starting gossip");
+        tokio::spawn(gossip::run(engine.clone(), gossip, cfg.peer.clone(), cfg.server.sid.clone()));
+    }
 
     let addr = format!("{}:{}", cfg.uplink.host, cfg.uplink.port);
     tracing::info!(server = %cfg.server.name, %addr, "linking to uplink");