diff --git a/Cargo.toml b/Cargo.toml index df0387e..96527c2 100644 --- a/Cargo.toml +++ b/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" diff --git a/src/config.rs b/src/config.rs index 56e20fa..71933a8 100644 --- a/src/config.rs +++ b/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, + #[serde(default)] + pub peer: Vec, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct Gossip { + // Address to accept peer connections on. Absent = dial-only node. + pub bind: Option, + // Shared secret both nodes must present. + pub secret: String, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct Peer { + pub addr: String, } #[derive(Debug, Deserialize)] diff --git a/src/engine/db.rs b/src/engine/db.rs index f7c3206..d8a2296 100644 --- a/src/engine/db.rs +++ b/src/engine/db.rs @@ -65,26 +65,31 @@ pub struct EventLog { origin: String, 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 } impl EventLog { fn open(path: PathBuf, origin: String) -> (Self, Vec) { - let mut events = Vec::new(); - let mut lamport = 0; - let mut versions: HashMap = 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::(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> { 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 { + self.versions.clone() + } + + // Entries a peer is missing, given the version vector it advertised. + fn missing_for(&self, peer: &HashMap) -> Vec { + 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 { + self.log.version_vector() + } + + /// The log entries a peer is missing, given the version vector it sent. + pub fn missing_for(&self, peer: &HashMap) -> Vec { + self.log.missing_for(peer) + } + pub fn exists(&self, name: &str) -> bool { self.accounts.contains_key(&key(name)) } diff --git a/src/engine/mod.rs b/src/engine/mod.rs index ae4a34d..d9807ac 100644 --- a/src/engine/mod.rs +++ b/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 { + self.db.version_vector() + } + + pub fn gossip_missing(&self, peer: &HashMap) -> Vec { + 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 }); diff --git a/src/gossip.rs b/src/gossip.rs new file mode 100644 index 0000000..a59a8bb --- /dev/null +++ b/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>; + +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 }, + Entry { entry: LogEntry }, +} + +// Start the listener (if bound) and a dialer per configured peer. +pub async fn run(engine: Shared, cfg: Gossip, peers: Vec, 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(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::(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::(&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::(&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, 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"); + } +} diff --git a/src/link.rs b/src/link.rs index a08e15c..696fde2 100644 --- a/src/link.rs +++ b/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, 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, engine: Arc>, 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, 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, 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) } } } diff --git a/src/main.rs b/src/main.rs index 08faa71..07e83f0 100644 --- a/src/main.rs +++ b/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");