diff --git a/src/engine/db.rs b/src/engine/db.rs index d8a2296..f2147d4 100644 --- a/src/engine/db.rs +++ b/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, // per-origin highest seq applied (version vector) entries: Vec, // full log, kept so peers can pull what they lack + outbound: Option>, // push newly committed entries to peers } impl EventLog { fn open(path: PathBuf, origin: String) -> (Self, Vec) { - 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::(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) { + self.outbound = Some(tx); + } + // Our version vector: highest seq applied per origin. fn version_vector(&self) -> HashMap { 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) { + self.log.set_outbound(tx); + } + /// Our version vector, advertised to peers so they can send what we lack. pub fn version_vector(&self) -> HashMap { self.log.version_vector() diff --git a/src/gossip.rs b/src/gossip.rs index a59a8bb..589b15d 100644 --- a/src/gossip.rs +++ b/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>; +type Outbound = broadcast::Sender; -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, origin: String) { +pub async fn run(engine: Shared, cfg: Gossip, peers: Vec, 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(stream: S, engine: Shared, secret: String, origin: String) -> anyhow::Result<()> +async fn session(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 b = engine("B", "conv-b"); + let (a, atx) = engine("A", "conv-a"); + 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 sb = tokio::spawn(session(cb, b.clone(), "s3cret".into(), "B".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(), 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 b = engine("B", "auth-b"); + let (a, atx) = engine("A", "auth-a"); + 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 sb = tokio::spawn(session(cb, b.clone(), "wrong".into(), "B".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(), btx)); let _ = tokio::time::timeout(Duration::from_secs(1), async { let _ = sa.await; diff --git a/src/main.rs b/src/main.rs index 07e83f0..1f710c5 100644 --- a/src/main.rs +++ b/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::(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);