| … | … | @@ -19,25 +19,39 @@ export const PRESENCE_PROTOCOL = 'atproto-room/signer-presence/v1' |
| 19 | 19 | // Returns a cleanup fn (idempotent). Safe to call on every connection AND to double-attach the same (conn, |
| 20 | 20 | // id): protomux returns null on the duplicate, so the 2nd attach is a harmless no-op (the 1st stays live). |
| 21 | 21 | // A non-presence peer simply never sends, so it's never marked present (and thus never eligible to sign). |
| 22 | export function attachPresence (conn, { selfKey, presence, id = null } = {}) { | |
| 23 | if (!conn || !selfKey || !presence) return () => {} | |
| 22 | // getHead: optional () => our current signed head rev (a TID string) — announced so the peer's election | |
| 23 | // can tell whether WE are caught up (currency). Returns { send(headRev?), close() }: `send` RE-announces | |
| 24 | // our head when it changes (the caller wires this to head advances) so a stale head can't mis-elect a | |
| 25 | // signer; `close` tears the channel down. (A non-object return would break callers that register `send`.) | |
| 26 | export function attachPresence (conn, { selfKey, presence, id = null, getHead = null } = {}) { | |
| 27 | // Returns a DETACH function (call it to tear down — the original contract) with a `.send(headRev)` | |
| 28 | // property for re-announcing our head when it changes. Hybrid so existing callers that use the return | |
| 29 | // as a cleanup fn keep working, while the currency wiring calls `.send`. | |
| 30 | const noop = () => {}; noop.send = () => {} | |
| 31 | if (!conn || !selfKey || !presence) return noop | |
| 24 | 32 | const self = String(selfKey) |
| 25 | 33 | let peerKey = null |
| 26 | 34 | let done = false |
| 27 | 35 | const clear = () => { if (peerKey) presence.markAbsent(peerKey) } |
| 28 | 36 | try { |
| 29 | 37 | const chan = Protomux.from(conn).createChannel({ protocol: PRESENCE_PROTOCOL, id }) |
| 30 | if (!chan) return () => {} | |
| 38 | if (!chan) return noop // duplicate (protocol,id) on this mux → harmless no-op | |
| 31 | 39 | const msg = chan.addMessage({ |
| 32 | 40 | encoding: c.json, |
| 33 | 41 | onmessage: (data) => { |
| 34 | if (data && data.key) { peerKey = String(data.key); presence.markPresent(peerKey) } | |
| 42 | if (!data || !data.key) return | |
| 43 | peerKey = String(data.key) | |
| 44 | presence.markPresent(peerKey) // presence (fires onChange on connect) | |
| 45 | if (data.headRev !== undefined) presence.noteHead(peerKey, data.headRev || null) // currency (fires onChange on a head change) | |
| 35 | 46 | } |
| 36 | 47 | }) |
| 37 | 48 | chan.open() |
| 38 | try { msg.send({ key: self }) } catch {} | |
| 49 | const announce = (headRev) => { try { msg.send({ key: self, headRev: headRev !== undefined ? (headRev || null) : (getHead ? getHead() : null) }) } catch {} } | |
| 50 | announce() // initial announce with our current head | |
| 39 | 51 | const onclose = () => { if (done) return; done = true; clear() } |
| 40 | 52 | conn.on('close', onclose) |
| 41 | return () => { if (done) return; done = true; try { conn.off('close', onclose) } catch {}; clear() } | |
| 42 | } catch { return () => {} } | |
| 53 | const detach = () => { if (done) return; done = true; try { conn.off('close', onclose) } catch {}; clear() } | |
| 54 | detach.send = (headRev) => announce(headRev) // re-announce our head (currency) when it changes | |
| 55 | return detach | |
| 56 | } catch { return noop } | |
| 43 | 57 | } |
| … | … | @@ -33,19 +33,38 @@ export class SignerPresence { |
| 33 | 33 | this.onChange = onChange |
| 34 | 34 | this._present = new Map() // memberKey -> lastSeen: Infinity = connected now; a number = ms it departed |
| 35 | 35 | this._present.set(this.self, Infinity) // we are always present to ourselves |
| 36 | // CURRENCY (anti-freeze): each co-owner's last-announced signed head rev (a monotonic TID string) or | |
| 37 | // null if unknown. A BEHIND co-owner must not win the election — else it holds the lease but the | |
| 38 | // forward-only sign gate blocks it (it's behind) while the caught-up peer defers to it → the chain | |
| 39 | // FREEZES (the stortron split-brain aftermath). currentSigner filters on this vs the network head. | |
| 40 | this._heads = new Map() // memberKey -> headRev (string) | null | |
| 36 | 41 | } |
| 37 | 42 | |
| 38 | 43 | _fire () { if (this.onChange) { try { this.onChange() } catch {} } } |
| 39 | 44 | |
| 40 | 45 | // A co-owner (holds the signing key + is attested) is connected. Idempotent. Fires onChange only on |
| 41 | 46 | // an actual transition (not-present/departed -> present) so the daemon re-evaluates the signer. |
| 42 | markPresent (memberKey) { | |
| 47 | markPresent (memberKey, headRev = undefined) { | |
| 43 | 48 | const k = String(memberKey) |
| 49 | if (headRev !== undefined) this._heads.set(k, headRev || null) // learn the peer's signed-head currency | |
| 44 | 50 | if (this._present.get(k) === Infinity) return // already connected — no change |
| 45 | 51 | this._present.set(k, Infinity) |
| 46 | 52 | this._fire() |
| 47 | 53 | } |
| 48 | 54 | |
| 55 | // Record OUR OWN current signed head rev (from the sign loop) so the election knows we're caught up. | |
| 56 | setSelfHead (rev) { this._heads.set(this.self, rev || null) } | |
| 57 | // Record a co-owner's announced signed head rev (from the presence channel). Fires onChange when the | |
| 58 | // currency changes (a peer catching up may flip who is the elected signer → re-evaluate). | |
| 59 | noteHead (memberKey, rev) { | |
| 60 | const k = String(memberKey) | |
| 61 | const prev = this._heads.get(k) | |
| 62 | if (prev === (rev || null)) return | |
| 63 | this._heads.set(k, rev || null) | |
| 64 | this._fire() | |
| 65 | } | |
| 66 | headOf (memberKey) { const v = this._heads.get(String(memberKey)); return v === undefined ? null : v } | |
| 67 | ||
| 49 | 68 | // A co-owner disconnected — start its grace countdown (do NOT drop it immediately; a blip must not fail |
| 50 | 69 | // the signer over). No-op for a key we never saw. Fires onChange so a departure is noticed event-driven. |
| 51 | 70 | markAbsent (memberKey) { |
| … | … | @@ -61,16 +80,26 @@ export class SignerPresence { |
| 61 | 80 | } |
| 62 | 81 | |
| 63 | 82 | // The elected signer = lowest-keyed effective co-owner (every connected peer computes the same one). |
| 64 | currentSigner (now = this.clock()) { | |
| 83 | // CURRENCY-AWARE (anti-freeze): when netHead is given, a co-owner is only eligible if its known signed | |
| 84 | // head is caught up to netHead (headRev >= netHead; revs are monotonic TID strings, null = unknown = | |
| 85 | // not caught up). This EXCLUDES a behind co-owner so it can't hold the lease while the forward-only | |
| 86 | // gate blocks it → the caught-up peer signs instead. If NONE are known-caught-up (no currency info | |
| 87 | // yet), fall back to the lowest-key effective co-owner — the forward-only sign gate still prevents any | |
| 88 | // behind sign, so the worst case is a transient no-op, never a regression. Deterministic: every peer | |
| 89 | // computes the same signer from the same (presence, announced-heads, netHead) → still exactly one signer. | |
| 90 | currentSigner (now = this.clock(), netHead = null) { | |
| 65 | 91 | const e = this.effective(now) |
| 66 | return e.length ? e[0] : null | |
| 92 | if (!e.length) return null | |
| 93 | if (!netHead) return e[0] | |
| 94 | const caughtUp = e.filter((k) => { const h = this._heads.get(k); return h != null && h >= netHead }) | |
| 95 | return caughtUp.length ? caughtUp[0] : e[0] | |
| 67 | 96 | } |
| 68 | 97 | |
| 69 | 98 | // Do WE hold the signer role right now? (Drop-in for lease.holds() / renewOrClaim() in the sign loop.) |
| 70 | holds (now = this.clock()) { return this.currentSigner(now) === this.self } | |
| 99 | holds (now = this.clock(), netHead = null) { return this.currentSigner(now, netHead) === this.self } | |
| 71 | 100 | |
| 72 | 101 | // Diagnostics snapshot (no writes). |
| 73 | state (now = this.clock()) { | |
| 74 | return { self: this.self, signer: this.currentSigner(now), effective: this.effective(now), holds: this.holds(now) } | |
| 102 | state (now = this.clock(), netHead = null) { | |
| 103 | return { self: this.self, signer: this.currentSigner(now, netHead), effective: this.effective(now), holds: this.holds(now, netHead), heads: Object.fromEntries(this._heads) } | |
| 75 | 104 | } |
| 76 | 105 | } |
| … | … | @@ -146,6 +146,8 @@ export class StateBaseV3 { |
| 146 | 146 | // as a view advance, so a single sb.watch(tick) drives ZERO-TIMER failover: the daemon re-evaluates |
| 147 | 147 | // holdsSigner + signs on writes, replication, AND presence events — no polling timer. |
| 148 | 148 | this._presence = this._selfKey ? new SignerPresence({ self: this._selfKey.toString('hex'), graceMs: Number(opts.signerGraceMs) || 15000, onChange: () => this._fireWatchers() }) : null |
| 149 | this._selfHead = null // our current signed head rev — announced over presence so a BEHIND co-owner is | |
| 150 | this._presenceSenders = new Set() // excluded from the election (anti-freeze); re-announced on head change | |
| 149 | 151 | this._getMembers = typeof opts.getMembers === 'function' ? opts.getMembers : null |
| 150 | 152 | // Injectable crypto for membership-gated admission (keet attestation primitives). |
| 151 | 153 | // attest: sign localKey‖base.key as our keet identity; verifyAttest: IdentityKey.verify. |
| … | … | @@ -409,14 +411,30 @@ export class StateBaseV3 { |
| 409 | 411 | // id: this._roomKey — a per-DID channel id so N DIDs' presence coexist on ONE shared multi-DID daemon |
| 410 | 412 | // connection (protomux keys on (protocol, id); without it every DID collides on a null id and only the |
| 411 | 413 | // first opens). Single-DID and multi-DID both derive the same roomKey per DID, so peers of a DID match. |
| 412 | attachSignerPresence (conn) { return this._presence ? attachPresence(conn, { selfKey: this._selfKey.toString('hex'), presence: this._presence, id: this._roomKey }) : (() => {}) } | |
| 414 | attachSignerPresence (conn) { | |
| 415 | if (!this._presence) return () => {} | |
| 416 | const handle = attachPresence(conn, { selfKey: this._selfKey.toString('hex'), presence: this._presence, id: this._roomKey, getHead: () => this._selfHead }) | |
| 417 | this._presenceSenders.add(handle.send) | |
| 418 | return () => { this._presenceSenders.delete(handle.send); try { handle() } catch {} } // handle() detaches | |
| 419 | } | |
| 420 | ||
| 421 | // Feed our current signed head rev into the presence election (currency) + RE-ANNOUNCE it to peers, so a | |
| 422 | // co-owner BEHIND the network head is excluded from the election while it is gate-blocked from signing — | |
| 423 | // the caught-up peer signs instead of the chain freezing. Called by the sign loop after sign/restore. | |
| 424 | setSignerHead (rev) { | |
| 425 | const r = rev || null | |
| 426 | if (r === this._selfHead) return | |
| 427 | this._selfHead = r | |
| 428 | if (this._presence) this._presence.setSelfHead(r) | |
| 429 | for (const send of this._presenceSenders) { try { send(r) } catch {} } | |
| 430 | } | |
| 413 | 431 | |
| 414 | 432 | // Do WE hold the active-signer role right now? Drop-in for the old lease.holds() in the sign loop: |
| 415 | 433 | // election is a deterministic function of who is CONNECTED (lowest-key effective co-owner), computed in |
| 416 | 434 | // RAM. A device with no co-owners connected is the sole signer (holds=true). No writes, ever. |
| 417 | holdsSigner (now) { return this._presence ? this._presence.holds(now) : true } | |
| 418 | currentSigner (now) { return this._presence ? this._presence.currentSigner(now) : (this._selfKey ? this._selfKey.toString('hex') : null) } | |
| 419 | signerState (now) { return this._presence ? this._presence.state(now) : { self: null, signer: null, effective: [], holds: true } } | |
| 435 | holdsSigner (now, netHead = null) { return this._presence ? this._presence.holds(now, netHead) : true } | |
| 436 | currentSigner (now, netHead = null) { return this._presence ? this._presence.currentSigner(now, netHead) : (this._selfKey ? this._selfKey.toString('hex') : null) } | |
| 437 | signerState (now, netHead = null) { return this._presence ? this._presence.state(now, netHead) : { self: null, signer: null, effective: [], holds: true } } | |
| 420 | 438 | |
| 421 | 439 | // keet's _autobaseApply commit shape: a per-batch transaction; handlers write |
| 422 | 440 | // into it; flush on success / close on error (never insert on the db directly). |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "atproto-room", |
| 3 | "version": "0.2.8", | |
| 3 | "version": "0.2.9", | |
| 4 | 4 | "description": "Attestation-gated, encrypted multi-writer room engine — StateBase (keet-model admission) + Lease + signer + RepoBase + LabelBase/LabelRoom + serve-membrane. Extracted from hyper-proto-peer; consumed by the atproto peer (repo devices) and the co-authored labeler.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "index.js", |
| … | … | @@ -16,3 +16,4 @@ import './multi-did.test.js' |
| 16 | 16 | import './non-indexer-default.test.js' |
| 17 | 17 | import './co-authored-labeler.test.js' |
| 18 | 18 | import './migration-readopt.test.js' |
| 19 | import './signer-presence-currency.test.mjs' |
| … | … | @@ -0,0 +1,64 @@ |
| 1 | // SignerPresence CURRENCY election — the anti-freeze fix for the stortron/tabertron split-brain aftermath. | |
| 2 | // | |
| 3 | // The forward-only sign gate stops a BEHIND signer from publishing a regression — but on its own it can | |
| 4 | // FREEZE the chain: the lowest-key co-owner still wins the presence election and holds the lease, the gate | |
| 5 | // blocks it (it's behind), and the caught-up peer defers to it → nobody signs. The election must therefore | |
| 6 | // exclude a behind co-owner so the caught-up peer signs instead. Deterministic: every peer computes the | |
| 7 | // same signer from (presence, announced heads, netHead), so there is still exactly one signer (no fork). | |
| 8 | import test from 'brittle' | |
| 9 | import { SignerPresence } from '../lib/signer-presence.js' | |
| 10 | ||
| 11 | // Faithful to live: stortron key < tabertron key (lowest key = classic election winner). | |
| 12 | const STOR = '71555d858367c85346d3e19752e82ec889613298e7da1066b9362130e5103583' | |
| 13 | const TAB = '81e1cbbf9169d43e2232fca60e10ff937783790eba06c8cb0a74fff12e285715' | |
| 14 | const BEHIND = '3mqzqosnt6c22' // 85-record head (stortron's stale line) | |
| 15 | const AHEAD = '3mr4ocucu6c22' // 91-record head (tabertron's advanced line) — AHEAD > BEHIND lexicographically | |
| 16 | const NOW = 1000 | |
| 17 | const mk = (self) => new SignerPresence({ self, clock: () => NOW }) | |
| 18 | ||
| 19 | test('currency election: a BEHIND lowest-key co-owner does NOT freeze the signer — the caught-up peer signs', (t) => { | |
| 20 | // From STORTRON's view (self behind at BEHIND; TAB present + caught up at AHEAD): | |
| 21 | const stor = mk(STOR) | |
| 22 | stor.setSelfHead(BEHIND) | |
| 23 | stor.markPresent(TAB, AHEAD) | |
| 24 | t.is(stor.currentSigner(NOW, AHEAD), TAB, 'behind STOR (lowest key) does NOT win — caught-up TAB does') | |
| 25 | t.absent(stor.holds(NOW, AHEAD), 'behind STOR does not hold the signer role (so the gate-skip is not a freeze)') | |
| 26 | ||
| 27 | // From TABERTRON's view (self caught up at AHEAD; STOR present but behind at BEHIND): | |
| 28 | const tab = mk(TAB) | |
| 29 | tab.setSelfHead(AHEAD) | |
| 30 | tab.markPresent(STOR, BEHIND) | |
| 31 | t.is(tab.currentSigner(NOW, AHEAD), TAB, 'TAB computes ITSELF the signer (behind STOR excluded)') | |
| 32 | t.ok(tab.holds(NOW, AHEAD), 'caught-up TAB holds the signer role → it signs → chain advances (unfreeze)') | |
| 33 | // Both peers elected TAB — deterministic, exactly one signer, no fork. | |
| 34 | }) | |
| 35 | ||
| 36 | test('currency election: once the behind co-owner CATCHES UP, the lowest key reclaims the role', (t) => { | |
| 37 | const stor = mk(STOR) | |
| 38 | stor.markPresent(TAB, AHEAD) | |
| 39 | stor.setSelfHead(BEHIND) | |
| 40 | t.is(stor.currentSigner(NOW, AHEAD), TAB, 'while behind, STOR defers to TAB') | |
| 41 | stor.setSelfHead(AHEAD) // STOR replicated the missing records + caught up | |
| 42 | t.is(stor.currentSigner(NOW, AHEAD), STOR, 'caught-up STOR (lowest key) reclaims the signer role — normal steady state') | |
| 43 | }) | |
| 44 | ||
| 45 | test('currency election: no netHead / no currency info → presence-only lowest-key (backward compatible)', (t) => { | |
| 46 | const sp = mk(STOR) | |
| 47 | sp.markPresent(TAB) | |
| 48 | t.is(sp.currentSigner(NOW), STOR, 'no netHead → classic lowest-key present election, unchanged') | |
| 49 | // netHead given but NOBODY is known-caught-up → fall back to lowest-key present; the forward-only sign | |
| 50 | // gate still prevents a behind sign, so the worst case is a transient no-op, never a regression/freeze-forever. | |
| 51 | t.is(sp.currentSigner(NOW, AHEAD), STOR, 'netHead but no currency announced → lowest-key fallback (gate still guards)') | |
| 52 | }) | |
| 53 | ||
| 54 | test('currency election: noteHead (peer announcement) flips the elected signer + fires onChange', (t) => { | |
| 55 | let fired = 0 | |
| 56 | const tab = new SignerPresence({ self: TAB, clock: () => NOW, onChange: () => { fired++ } }) | |
| 57 | tab.setSelfHead(AHEAD) | |
| 58 | tab.markPresent(STOR) // STOR present, currency unknown yet | |
| 59 | t.is(tab.currentSigner(NOW, AHEAD), TAB, 'unknown STOR currency → TAB (self, caught up) signs') | |
| 60 | fired = 0 | |
| 61 | tab.noteHead(STOR, AHEAD) // STOR announces it caught up | |
| 62 | t.ok(fired > 0, 'a currency change fires onChange (re-evaluate the signer)') | |
| 63 | t.is(tab.currentSigner(NOW, AHEAD), STOR, 'now STOR is known caught up + lower key → it becomes the signer') | |
| 64 | }) |