Push new log entries to peers on commit

A broadcast channel forwards each freshly committed entry straight to
connected peers, so a registration propagates in milliseconds. The
periodic digest drops to a 10s anti-entropy backstop.

src/engine/db.rs

@@ -7,6 +7,7 @@ use argon2::password_hash::rand_core::OsRng;
 use argon2::password_hash::SaltString;
 use argon2::{Argon2, PasswordHash, PasswordHasher, PasswordVerifier};
 use serde::{Deserialize, Serialize};
+use tokio::sync::broadcast;
 
 use super::scram::{self, Hash};
 
@@ -66,11 +67,12 @@ pub struct EventLog {
     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
+    outbound: Option<broadcast::Sender<LogEntry>>, // push newly committed entries to peers
 }
 
 impl EventLog {
     fn open(path: PathBuf, origin: String) -> (Self, Vec<Event>) {
-        let mut log = Self { path, origin, lamport: 0, versions: HashMap::new(), entries: Vec::new() };
+        let mut log = Self { path, origin, lamport: 0, versions: HashMap::new(), entries: Vec::new(), outbound: None };
         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) {
@@ -104,6 +106,7 @@ 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.notify(&entry);
         self.entries.push(entry);
         Ok(())
     }
@@ -119,10 +122,23 @@ impl EventLog {
         self.lamport = self.lamport.max(entry.lamport) + 1; // Lamport receive rule
         self.versions.insert(entry.origin.clone(), entry.seq);
         let event = entry.event.clone();
+        self.notify(&entry);
         self.entries.push(entry);
         Ok(Some(event))
     }
 
+    // Push a freshly committed entry to connected peers, if any are wired up.
+    // Best effort: a lagging subscriber just relies on the periodic digest.
+    fn notify(&self, entry: &LogEntry) {
+        if let Some(tx) = &self.outbound {
+            let _ = tx.send(entry.clone());
+        }
+    }
+
+    fn set_outbound(&mut self, tx: broadcast::Sender<LogEntry>) {
+        self.outbound = Some(tx);
+    }
+
     // Our version vector: highest seq applied per origin.
     fn version_vector(&self) -> HashMap<String, u64> {
         self.versions.clone()
@@ -208,6 +224,11 @@ impl Db {
         Ok(())
     }
 
+    /// Wire the log to a broadcast channel; each new entry is pushed to peers.
+    pub fn set_outbound(&mut self, tx: broadcast::Sender<LogEntry>) {
+        self.log.set_outbound(tx);
+    }
+
     /// 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()

src/gossip.rs

@@ -8,15 +8,16 @@ 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 tokio::sync::{broadcast, mpsc, Mutex};
 
 use crate::config::{Gossip, Peer};
 use crate::engine::db::LogEntry;
 use crate::engine::Engine;
 
 type Shared = Arc<Mutex<Engine>>;
+type Outbound = broadcast::Sender<LogEntry>;
 
-const TICK: Duration = Duration::from_secs(3); // how often we re-advertise our vector
+const TICK: Duration = Duration::from_secs(10); // anti-entropy backstop; pushes handle the common case
 const REDIAL: Duration = Duration::from_secs(5); // reconnect backoff for dialed peers
 
 // Wire protocol: one JSON object per line.
@@ -29,16 +30,16 @@ enum Msg {
 }
 
 // Start the listener (if bound) and a dialer per configured peer.
-pub async fn run(engine: Shared, cfg: Gossip, peers: Vec<Peer>, origin: String) {
+pub async fn run(engine: Shared, cfg: Gossip, peers: Vec<Peer>, origin: String, outbound: Outbound) {
     if let Some(bind) = cfg.bind.clone() {
-        tokio::spawn(listen(bind, engine.clone(), cfg.secret.clone(), origin.clone()));
+        tokio::spawn(listen(bind, engine.clone(), cfg.secret.clone(), origin.clone(), outbound.clone()));
     }
     for peer in peers {
-        tokio::spawn(dial(peer.addr, engine.clone(), cfg.secret.clone(), origin.clone()));
+        tokio::spawn(dial(peer.addr, engine.clone(), cfg.secret.clone(), origin.clone(), outbound.clone()));
     }
 }
 
-async fn listen(bind: String, engine: Shared, secret: String, origin: String) {
+async fn listen(bind: String, engine: Shared, secret: String, origin: String, outbound: Outbound) {
     let listener = match TcpListener::bind(&bind).await {
         Ok(l) => l,
         Err(e) => return tracing::error!(%e, %bind, "gossip bind failed"),
@@ -47,9 +48,9 @@ async fn listen(bind: String, engine: Shared, secret: String, origin: String) {
     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());
+            let (engine, secret, origin, outbound) = (engine.clone(), secret.clone(), origin.clone(), outbound.clone());
             tokio::spawn(async move {
-                if let Err(e) = session(stream, engine, secret, origin).await {
+                if let Err(e) = session(stream, engine, secret, origin, outbound).await {
                     tracing::debug!(%e, "gossip session ended");
                 }
             });
@@ -57,12 +58,12 @@ async fn listen(bind: String, engine: Shared, secret: String, origin: String) {
     }
 }
 
-async fn dial(addr: String, engine: Shared, secret: String, origin: String) {
+async fn dial(addr: String, engine: Shared, secret: String, origin: String, outbound: Outbound) {
     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 {
+                if let Err(e) = session(stream, engine.clone(), secret.clone(), origin.clone(), outbound.clone()).await {
                     tracing::debug!(%e, %addr, "gossip session ended");
                 }
             }
@@ -73,7 +74,7 @@ async fn dial(addr: String, engine: Shared, secret: String, origin: String) {
 }
 
 // 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<()>
+async fn session<S>(stream: S, engine: Shared, secret: String, origin: String, outbound: Outbound) -> anyhow::Result<()>
 where
     S: AsyncRead + AsyncWrite + Send + 'static,
 {
@@ -106,7 +107,9 @@ where
         }
     }
 
-    // Re-advertise our version vector on a timer so new local events propagate.
+    // Re-advertise our version vector on a timer. The first tick fires an
+    // immediate digest for initial sync; later ones are a backstop for anything
+    // a push dropped.
     let ticker = {
         let (tx, engine) = (tx.clone(), engine.clone());
         tokio::spawn(async move {
@@ -120,6 +123,26 @@ where
         })
     };
 
+    // Push freshly committed entries straight to the peer, so a new registration
+    // propagates in milliseconds instead of waiting for the next digest. Idempotent
+    // ingest on the far side absorbs the echo of an entry the peer just sent us.
+    let forwarder = {
+        let (tx, mut rx) = (tx.clone(), outbound.subscribe());
+        tokio::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    Ok(entry) => {
+                        if send(&tx, &Msg::Entry { entry }).await.is_err() {
+                            break;
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Lagged(_)) => continue, // digest backstop catches up
+                    Err(broadcast::error::RecvError::Closed) => break,
+                }
+            }
+        })
+    };
+
     // Answer a peer's digest with what it lacks; apply the entries it sends us.
     let result = loop {
         match reader.next_line().await {
@@ -143,6 +166,7 @@ where
     };
 
     ticker.abort();
+    forwarder.abort();
     writer.abort();
     result
 }
@@ -158,25 +182,28 @@ mod tests {
     use crate::engine::db::Db;
     use crate::services::nickserv::NickServ;
 
-    fn engine(origin: &str, tag: &str) -> Shared {
+    fn engine(origin: &str, tag: &str) -> (Shared, Outbound) {
         let path = std::env::temp_dir().join(format!("fedserv-gossip-{tag}.jsonl"));
         let _ = std::fs::remove_file(&path);
+        let (tx, _) = broadcast::channel(1024);
         let mut db = Db::open(&path, origin);
         db.scram_iterations = 4096;
+        db.set_outbound(tx.clone());
         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)))
+        (Arc::new(Mutex::new(Engine::new(vec![Box::new(ns)], db))), tx)
     }
 
-    // An account registered on node A reaches node B over a gossip link.
+    // An account that exists before the link comes up reaches the peer via the
+    // initial digest exchange (the pull path).
     #[tokio::test]
     async fn two_nodes_converge_over_a_link() {
-        let a = engine("A", "conv-a");
+        let (a, atx) = engine("A", "conv-a");
-        let b = engine("B", "conv-b");
+        let (b, btx) = 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 sa = tokio::spawn(session(ca, a.clone(), "s3cret".into(), "A".into(), atx));
-        let sb = tokio::spawn(session(cb, b.clone(), "s3cret".into(), "B".into()));
+        let sb = tokio::spawn(session(cb, b.clone(), "s3cret".into(), "B".into(), btx));
 
         let mut converged = false;
         for _ in 0..100 {
@@ -191,16 +218,42 @@ mod tests {
         assert!(converged, "B should receive A's account over gossip");
     }
 
+    // A write made after the link is up must arrive within the poll interval,
+    // which only the push path can do (the digest backstop is 10s).
+    #[tokio::test]
+    async fn push_propagates_new_writes() {
+        let (a, atx) = engine("A", "push-a");
+        let (b, btx) = engine("B", "push-b");
+
+        let (ca, cb) = tokio::io::duplex(64 * 1024);
+        let sa = tokio::spawn(session(ca, a.clone(), "s3cret".into(), "A".into(), atx));
+        let sb = tokio::spawn(session(cb, b.clone(), "s3cret".into(), "B".into(), btx));
+        tokio::time::sleep(Duration::from_millis(200)).await; // let both subscribe
+
+        a.lock().await.test_register("late");
+        let mut got = false;
+        for _ in 0..40 {
+            if b.lock().await.test_has_account("late") {
+                got = true;
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+        sa.abort();
+        sb.abort();
+        assert!(got, "a post-connect write should push to B well under the 10s digest");
+    }
+
     // 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 (a, atx) = engine("A", "auth-a");
-        let b = engine("B", "auth-b");
+        let (b, btx) = 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 sa = tokio::spawn(session(ca, a.clone(), "right".into(), "A".into(), atx));
-        let sb = tokio::spawn(session(cb, b.clone(), "wrong".into(), "B".into()));
+        let sb = tokio::spawn(session(cb, b.clone(), "wrong".into(), "B".into(), btx));
 
         let _ = tokio::time::timeout(Duration::from_secs(1), async {
             let _ = sa.await;

src/main.rs

@@ -40,13 +40,15 @@ async fn main() -> Result<()> {
         guest_nick: cfg.server.guest_nick.clone(),
         guest_seq: (ts % 100_000) as u32,
     })];
+    let (gossip_tx, _) = tokio::sync::broadcast::channel::<engine::db::LogEntry>(1024);
     let mut db = engine::db::Db::open("fedserv.db.jsonl", &cfg.server.sid);
     db.scram_iterations = cfg.server.scram_iterations;
+    db.set_outbound(gossip_tx.clone());
     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()));
+        tokio::spawn(gossip::run(engine.clone(), gossip, cfg.peer.clone(), cfg.server.sid.clone(), gossip_tx));
     }
 
     let addr = format!("{}:{}", cfg.uplink.host, cfg.uplink.port);