diff --git a/src/engine/db.rs b/src/engine/db.rs index 23edcb6..36b0f8c 100644 --- a/src/engine/db.rs +++ b/src/engine/db.rs @@ -17,6 +17,13 @@ pub struct Account { pub password_hash: String, pub email: Option, pub ts: u64, + // The node that first registered this account (its "home"). With `ts` it + // deterministically resolves a concurrent registration of the same name across + // the federation. Named `home` not `origin` so it can't collide with the log + // envelope's `origin` when this struct is flattened into a LogEntry. Defaulted + // for records written before this existed. + #[serde(default)] + pub home: String, // SCRAM verifiers (`v=1,i=,s=,sk=,sv=`), computed from the password at // registration. Absent on accounts registered before SCRAM support. #[serde(default)] @@ -512,6 +519,7 @@ impl Db { password_hash: creds.password_hash, email, ts: now(), + home: self.log.origin.clone(), scram256: Some(creds.scram256), scram512: Some(creds.scram512), certfps: Vec::new(), @@ -529,6 +537,11 @@ impl Db { self.register_prepared(name, creds, email) } + #[cfg(test)] + pub(crate) fn test_hash(&self, name: &str) -> Option { + self.accounts.get(&key(name)).map(|a| a.password_hash.clone()) + } + /// Check credentials; on success return the account's canonical name (its /// stored casing), else None. pub fn authenticate(&self, name: &str, password: &str) -> Option<&str> { @@ -743,12 +756,29 @@ impl Db { } } +// Whether `held` is the rightful owner over a rival `claim` to the same name: +// the earlier registration wins (lower ts), ties broken by the lower origin. A +// total order over the fields carried in the account, so it survives compaction +// (which re-authors the log envelope but keeps account content) and is identical +// on every node. Strictly less-than, so an equal (idempotent) re-delivery still +// overwrites with identical content. +fn owns_over(held: &Account, claim: &Account) -> bool { + (held.ts, held.home.as_str()) < (claim.ts, claim.home.as_str()) +} + // Fold one event into the store. Shared by log replay (`open`) and gossip // ingest, so both routes reconstruct identical state. fn apply(accounts: &mut HashMap, channels: &mut HashMap, event: Event) { match event { Event::AccountRegistered(a) => { - accounts.insert(key(&a.name), a); + // Resolve a concurrent registration of the same name deterministically: + // keep whichever claim is earlier (lower ts, then lower origin), so every + // node converges on the same owner regardless of gossip delivery order. + let k = key(&a.name); + let keep_existing = accounts.get(&k).is_some_and(|cur| owns_over(cur, &a)); + if !keep_existing { + accounts.insert(k, a); + } } Event::CertAdded { account, fp } => { if let Some(a) = accounts.get_mut(&key(&account)) { @@ -880,6 +910,32 @@ mod tests { assert!(!glob_match("alice!*@*", "bob!~b@h")); } + #[test] + fn account_conflict_resolves_deterministically() { + let alice = |hash: &str, ts: u64, home: &str| Account { + name: "alice".into(), password_hash: hash.into(), email: None, + ts, home: home.into(), scram256: None, scram512: None, certfps: vec![], + }; + let converge = |first: &Account, second: &Account| { + let (mut acc, mut ch) = (HashMap::new(), HashMap::new()); + apply(&mut acc, &mut ch, Event::AccountRegistered(first.clone())); + apply(&mut acc, &mut ch, Event::AccountRegistered(second.clone())); + acc["alice"].password_hash.clone() + }; + // Earlier registration wins, regardless of which claim applies first. + let early = alice("EARLY", 100, "nodeB"); + let late = alice("LATE", 200, "nodeA"); + assert_eq!(converge(&early, &late), "EARLY"); + assert_eq!(converge(&late, &early), "EARLY", "commutative: same winner in either order"); + // Same ts: the lower origin wins, in either order. + let a = alice("A", 100, "nodeA"); + let b = alice("B", 100, "nodeB"); + assert_eq!(converge(&a, &b), "A"); + assert_eq!(converge(&b, &a), "A", "ts tie broken by lower origin, either order"); + // Idempotent: re-delivering the winner keeps it. + assert_eq!(converge(&a, &a), "A"); + } + #[test] fn channel_state_is_node_local_but_persists() { let path = tmp("scope"); @@ -963,7 +1019,7 @@ mod tests { db.register("alice", "pw", None).unwrap(); let bob = Account { name: "bob".into(), password_hash: "x".into(), email: None, - ts: 0, scram256: None, scram512: None, certfps: vec![], + ts: 0, home: "peer".into(), scram256: None, scram512: None, certfps: vec![], }; let entry = LogEntry { origin: "peer".into(), seq: 0, lamport: 1, event: Event::AccountRegistered(bob) }; db.ingest(entry).unwrap(); diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 6c7c153..c4885a7 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -127,6 +127,16 @@ impl Engine { self.db.exists(name) } + #[cfg(test)] + pub(crate) fn test_register_pw(&mut self, name: &str, pw: &str) { + self.db.register(name, pw, None).unwrap(); + } + + #[cfg(test)] + pub(crate) fn test_account_hash(&self, name: &str) -> Option { + self.db.test_hash(name) + } + #[cfg(test)] pub(crate) fn test_register_channel(&mut self, name: &str, founder: &str) { self.db.register_channel(name, founder).unwrap(); diff --git a/src/gossip.rs b/src/gossip.rs index b8b5e8e..96440d1 100644 --- a/src/gossip.rs +++ b/src/gossip.rs @@ -327,6 +327,34 @@ mod tests { assert!(got, "a post-connect write should push to B well under the 10s digest"); } + // Two nodes registering the same name before they link must converge on one + // owner — no split-brain. Both register "alice" with different passwords, then + // the link resolves it deterministically to a single winner on both sides. + #[tokio::test] + async fn concurrent_registration_converges_to_one_owner() { + let (a, atx) = engine("A", "conflict-a"); + let (b, btx) = engine("B", "conflict-b"); + a.lock().await.test_register_pw("alice", "from-a"); + b.lock().await.test_register_pw("alice", "from-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)); + + let mut converged = false; + for _ in 0..100 { + let (ha, hb) = (a.lock().await.test_account_hash("alice"), b.lock().await.test_account_hash("alice")); + if ha.is_some() && ha == hb { + converged = true; + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + sa.abort(); + sb.abort(); + assert!(converged, "both nodes must agree on a single owner of the contested name"); + } + // Accounts federate, but channel registrations stay on the node that made // them: A registers both; B receives the account and never the channel. #[tokio::test]