// SwarmDevice — one device of the swarm: the composition that IS atproto-base at the daemon
// level. It wraps the shared StateBaseV3 intent-log with the repo-op layer (RepoBase), the
// single-signer lease, and the shared-key signer. Every attested device writes repo intents into
// the shared autobase; whichever device currently holds the lease materializes the linearized
// view into the ONE canonical signed atproto commit chain, continuing from the last signed head.
//
// Construct the StateBaseV3 elsewhere (with the membership admission opts from lib/membership.js —
// getMembers/attest/verifyAttest/selfKey — and the members-only roomSeed/encryptionKey) and pass
// it in. `privateKey` is the SHARED #atproto signing key (distributed to attested devices).
import { RepoBase, isTombstone } from 'atproto-room/lib/repo-base.js'
import { PointerSubstrate } from 'atproto-store/lib/pointer-substrate.js'
import { Lease } from 'atproto-room/lib/lease.js'
import { materialize, buildCommitWrite } from 'atproto-room/lib/signer.js'
import { getRepoFromCommit } from 'atproto-store/lib/read-view.js'
import { tidNow, nextRev } from 'atproto-store/lib/tid.js'
import { signAssertion } from 'atproto-identity/lib/auth.js'
import b4a from 'b4a'
import { CID } from 'multiformats/cid'
import * as dagCbor from '@ipld/dag-cbor'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
// The firehose lives in the SAME shared StateBaseV3 (option a): each lease-holder sign() writes the
// signed #commit under __commit__/<rev>. It replicates to every device + the keyless membrane reads
// it there — no separate Hypercore, no fork hazard, continuous across lease hand-offs. RepoBase
// skips the reserved "__" namespace so these entries never leak into the signed repo.
const FIREHOSE_PREFIX = '__commit__/'
import { encodeReceipt, receiptEpoch } from 'atproto-room/lib/membership.js'
// ⛔ Frames are keyed __commit__/<rev>/<commitCid> (frameKey), NOT __commit__/<rev>. A partition elects a
// signer on each side; both continue nextRev(head) — the SAME rev over different records — and the old
// key let one overwrite the other, so the losing fork vanished and its consumers never got the other
// side's records (test/partition-fork-heal.test.mjs). commitLine decides the chain the same way on
// every device; frames off it are forks that the next commit heals.
import { commitLine, frameKey, healPlan, CHAIN_MARK } from 'atproto-room/lib/commit-line.js'
// The stand-in old value that turns a restated path into an `update` op. Never a real record's CID.
// indigo verify.go MaxCommitOps — a #commit with more ops is an error in strict mode (the default).
const MAX_COMMIT_OPS = 200
const HEAL_PLACEHOLDER = cidForBytesSync(dagCbor.encode({ $type: 'hyper-proto-peer.healPlaceholder' })).toString()
// StateBaseV3's durable banned-identity set + minimum-epoch receipt (see state-base.js).
// atproto-base-REVOCATION.md §3/§6.
const REMOVED_SET = '__removed__'
const RECEIPT_KEY = '__receipt__'
const PREFS_KEY = '__prefs__' // private bsky preferences (birthdate etc.); reserved → never in the public repo
const DEVICE_PREFIX = '__device__/' // delegated browser sub-key registry (reserved → never in the public repo)
const DEVICE_REVOKED_PREFIX = '__device_revoked__/' // per-sub-key revocation tombstone (reserved → never public)
// The rev from which record deletions are guaranteed to leave a TOMBSTONE. Paths whose create frame
// predates it may have been hard-deleted by an older build, so for those the forward-only gate keeps
// the legacy "absent == deleted" reading; at or after it, absence can only mean "not replicated to me"
// and the gate fails safe. Unset => every path is legacy => behaviour identical to before tombstones.
const TOMBSTONE_EPOCH = '__tombstone_epoch__'
// Serialize a buildCommitWrite result into a JSON-safe firehose entry (binary as hex) — the shape
// StateBaseV3 stores + the membrane reconstructs a subscribeRepos #commit from.
function serializeEntry (write, seq) {
const hex = (u8) => b4a.toString(b4a.from(u8), 'hex')
return {
seq,
// The REAL wall-clock emission time, recorded once by the signer into the shared __commit__ log.
// The membrane used to recompute the frame's `time` from the rev (revTime) so that every device
// projected a byte-identical frame — fine while revs were wall-clock TIDs, but revs are now a
// deterministic prev+1µs chain, so a derived time freezes near the FIRST commit and recedes
// further from reality with every commit. Relays read `time` as liveness metadata and this fleet
// already has to fight non-archival relays dropping accounts that look stale.
// Recording it here keeps BOTH properties: it is real, and it is identical on every device
// because they all replicate and read the same stored fact instead of each deriving it.
// `time` is frame metadata only — never part of the signed commit (atproto-store firehose.js:37).
time: new Date().toISOString(),
rev: write.rev,
since: write.since,
prev: write.signedCommit.prev ? write.signedCommit.prev.toString() : null,
prevData: write.prevData ? write.prevData.toString() : null, // prev MST root (for the #commit frame)
commitCid: write.commitCid.toString(),
// Written by the fork-aware signer: only marked frames can be classified as forks (commit-line.js —
// legacy logs carry restarted chains and falling revs that must stay on the line untouched).
chain: CHAIN_MARK,
// A heal: the fork commits it knew (heals) and the fork-point rev from which it restated every touched
// path (asserts). atproto-room healPlan decides from these whether a heal can have been skipped.
...(write.asserts !== undefined ? { heals: write.heals || [], asserts: write.asserts } : {}),
...(write.partsFrom ? { partsFrom: write.partsFrom } : {}),
// A PARTIAL heal: restated a slice of the window that starts at this fork-point rev.
...(write.restates !== undefined ? { restates: write.restates } : {}),
commit: hex(write.commitBlock.bytes),
ops: write.ops.map((o) => ({ action: o.action, path: o.path, cid: o.cid ? o.cid.toString() : null, prev: o.prev ? o.prev.toString() : null })),
blocks: write.newBlocks.map((b) => ({ cid: b.cid.toString(), bytes: hex(b.bytes) }))
}
}
export class SwarmDevice {
constructor (statebase, { did, privateKey, self, ttlMs = 15000, clock, firehose = false, recordStore = null, maxCommitOps = MAX_COMMIT_OPS } = {}) {
if (!did || !privateKey || !self) throw new Error('SwarmDevice: did, privateKey, self required')
this.sb = statebase
this.did = did
this.privateKey = privateKey
// Records: when `recordStore` (an OverlayRepoStore) is given, use the POINTER design — a tiny
// {$oid,size} pointer per record in the shared StateBase autobase (so the record SET converges
// symmetrically across attested devices, like the default path) while the record BYTES live as
// deduped overlay files (O(current), off the append-only log; the firehose no longer carries them
// once 2b lands). Else the legacy path stores full records in the StateBase. Reserved "__" keys
// (firehose __commit__, admission, prefs) ALWAYS stay on the StateBase directly (this.sb). RepoBase
// sees the same records either way, so sign()/materialize are unchanged. See
// docs/SYNTHESIS-storage-consensus-append-2026-07-19.md (Step 2a).
// ⚠ A repo previously written files-only (OverlayRepoSubstrate) has bytes-on-disk but NO pointers —
// migratePointers() below backfills them at boot so sign() never reads an empty set + wipes (wj62).
this.repo = new RepoBase(recordStore ? new PointerSubstrate(statebase, recordStore) : statebase, { did })
this._recordStore = recordStore
this.lease = new Lease(statebase, { self, ttlMs })
this._clock = clock // optional injectable micros clock for deterministic revs in tests
this.firehose = firehose // when true, sign() persists the signed #commit into __commit__/<rev>
this._maxCommitOps = maxCommitOps // the relay's per-commit op limit; a heal is split to fit it (tests lower it)
this._lastRecords = [] // record set at the last sign — the `before` side of the #commit diff
this.head = null // { rev, commitCid, dataCid, signedCommit } — the last signed commit
}
// Tolerant snapshot reads. The substrate contract is only { set, del, get, list } (RepoBase's
// documented interface, and all test/serve-device.test.js's memKV implements) — getNow/listNow are a
// StateBaseV3 extension for non-blocking reads. RepoBase already falls back this way; reading them
// off this.sb directly does not, which broke get-state on a minimal substrate.
_listNow (prefix = '') { return (typeof this.sb.listNow === 'function') ? this.sb.listNow(prefix) : this.sb.list(prefix) }
_getNow (key) { return (typeof this.sb.getNow === 'function') ? this.sb.getNow(key) : this.sb.get(key) }
// ── writes: any attested device appends repo intents to the shared log ──
putRecord (collection, rkey, record) { return this.repo.putRecord(collection, rkey, record) }
deleteRecord (collection, rkey) { return this.repo.deleteRecord(collection, rkey) }
// ── Step 2a migration: backfill files-only records into autobase pointers ──
// Idempotent, WRITABLE-ONLY. A repo previously written files-only (OverlayRepoSubstrate) has record
// BYTES on disk but no pointers in the shared autobase, so under PointerSubstrate the pointer set
// reads empty — the wj62-wipe hazard. This writes a {$oid,size} pointer per un-pointed record file
// so listRecords() reflects the real set BEFORE the boot path decides adopt/restore/fresh and before
// the first sign(). No-op unless the record substrate is a PointerSubstrate (has backfill) AND this
// device is a writable (admitted) writer — only the creator can append pointers; peers converge them
// by replication. Delegates to PointerSubstrate.backfill (never clobbers a converged pointer/tombstone).
async migratePointers () {
const sub = this.repo.sb
if (!sub || typeof sub.backfill !== 'function') return { skipped: 'not-pointer-substrate' }
if (!(this.sb.base && this.sb.base.writable)) return { skipped: 'not-writable' }
const r = await sub.backfill()
if (r.written) console.log(JSON.stringify({ msg: 'migratePointers: backfilled files-only records into pointers', did: this.did, ...r }))
return r
}
// ── MIGRATION: inline records → pointers (the loser-club shape) ─────────────────────────────────
// A DID written through the LEGACY substrate but now running the pointer design has full RECORDS
// sitting in the autobase where { $oid, size } pointers belong. It reads perfectly here and is
// UNFETCHABLE BY EVERY PEER — _resolve has no oid to fetch by, so it falls back to the local file.
// A co-owner converges the keys and can never turn them into bytes, which is why loser-club (3
// records, all inline) cannot be co-owned at all until this runs.
//
// Distinct from migratePointers/backfill, which covers the opposite gap (bytes on disk, NO entry in
// the autobase) and deliberately skips any key that already has a value — an inline record is such a
// value, so backfill can never convert one.
//
// Content is PRESERVED, not chosen: the pointer addresses the bytes that already resolve (the local
// file), asserted by hash, so the signed repo does not change. WRITABLE-ONLY (it appends pointers;
// peers converge them by replication). Idempotent. Nothing deleted. Operator-triggered, not a boot
// step: it rewrites autobase entries, which is not something to do implicitly on every start.
async migrateInlinePointers () {
const sub = this.repo.sb
if (!sub || typeof sub.migrateInline !== 'function') return { skipped: 'not-pointer-substrate' }
if (!(this.sb.base && this.sb.base.writable)) return { skipped: 'not-writable' }
const r = await sub.migrateInline()
console.log(JSON.stringify({ msg: 'migrateInlinePointers: inline records converted to fetchable pointers', did: this.did, ...r }))
return r
}
// EXPLICIT REPAIR for a store whose overlay INDEX is missing or stale — re-hashes and re-registers
// every record file so peers can fetch it by oid. Distinct from the boot path: OverlayRepoStore.ready()
// only re-seeds the in-memory path maps from an already-healthy index, which is cheap enough to run
// every boot; this walks and re-hashes everything, which is not (it is the cold-start cost that made
// pgh-sidecar take ~190s to bind its socket), so it stays operator-triggered.
//
// Why it is worth exposing: byte resolution GATES SIGNING — the forward-only gate skips while a
// converged pointer's bytes are unresolved — so an impoverished index silently stops this device
// signing at all. Measured on tabertron: 51 of ~131 record paths indexed. The capability shipped in
// atproto-store with NO trigger (no RPC, no CLI), i.e. unreachable; this is that trigger.
async reindexOverlay () {
if (!this._recordStore || typeof this._recordStore.reindexOverlay !== 'function') return { skipped: 'no-record-store' }
const r = await this._recordStore.reindexOverlay()
console.log(JSON.stringify({ msg: 'reindexOverlay: record files re-registered as servable-by-oid', did: this.did, ...r }))
return r
}
// ── PEER-LEVEL BYTE CONVERGENCE — the missing half of the design ────────────────────────────────
// The autobase guarantees POINTER convergence: every co-owner agrees on WHICH records exist. Byte
// convergence had no equivalent. The ONLY path that resolves record bytes is sign()'s
// listRecords({now:true}) — and sign() returns at the ELECTION GATE (holdsSigner, above) before it
// ever gets there. So a co-owner that is not the current signer never fetched a single record byte:
// its byte set grew only from what it authored locally, while its pointer set stayed correct and
// agreeing. Both nodes therefore report a healthy, identical record set while silently diverging —
// "co-owned on paper, single-homed in practice". Measured on wj62: 125 vs 124 posts, each node
// holding one the other lacked, and a record unresolved for 75+ hours that appeared instantly the
// moment anything called a byte-resolving read.
//
// PULL-ONLY: this never appends to the autobase, so it is safe on a NON-WRITABLE replica — which is
// precisely the device that needs it. Idempotent (ensureLocal short-circuits on a local hit) and
// bounded per pass (`more` tells the caller to keep going) so a backlog never blocks signing.
// Re-seeding is automatic — hyper-overlay re-registers what it fetches (overlay.js fetchFile), so a
// peer that catches up becomes a server for the next one.
async reconcileBytes ({ limit = 32 } = {}) {
if (!this._recordStore || typeof this._recordStore.ensureLocal !== 'function') return { skipped: 'no-record-store' }
let raw
try { raw = await this._listNow('') } catch (e) { return { skipped: 'list-failed', error: e.message } }
let checked = 0; let missing = 0; let stale = 0; let fetched = 0; let failed = 0; let more = false
for (const [k, ptr] of Object.entries(raw)) {
const i = k.indexOf('/')
if (i <= 0 || k.startsWith('__')) continue // reserved/control key — never a record
if (isTombstone(ptr)) continue // deleted — there are no bytes to pull
checked++
const collection = k.slice(0, i)
const rkey = k.slice(i + 1)
let have = null
try { have = await this._recordStore.get(collection, rkey) } catch {}
if (have != null) {
// PRESENCE IS NOT AGREEMENT. This used to `continue` on any local file, so a SUPERSEDED copy
// was indistinguishable from the converged one and byte convergence silently skipped exactly
// the records that had diverged — the wj62 profile/avatar case, invisible to this pass by
// construction. Verify the local bytes against the pointer; only a match is convergence.
let ok = true
if (ptr && ptr.$oid && typeof this._recordStore.localOid === 'function') {
const h = await this._recordStore.localOid(collection, rkey)
ok = !h || h.oid === ptr.$oid // only a POSITIVE mismatch is staleness; unhashable => leave alone
}
if (ok) continue
stale++ // fall through: ensureLocal now re-fetches the authoritative bytes over a stale file
} else missing++
if (fetched + failed >= limit) { more = true; continue } // bounded pass; caller continues
const rec = await this._recordStore.ensureLocal(collection, rkey, ptr && ptr.$oid)
if (rec != null) fetched++; else failed++
}
// Quiet in steady state, reports on change — a peer that is simply up to date says nothing, and a
// peer that is stuck says so once rather than on every event.
const sig = checked + ':' + missing + ':' + stale + ':' + fetched + ':' + failed
if ((fetched || missing || stale) && sig !== this._lastReconcileSig) {
console.log(JSON.stringify({ msg: 'reconcileBytes: resolving converged pointers into local bytes', did: this.did, checked, missing, stale, fetched, failed, more }))
}
this._lastReconcileSig = sig
return { checked, missing, stale, fetched, failed, more }
}
// ── READ-ONLY INTEGRITY DIAGNOSTIC: does my local byte set match the converged pointer set? ──────
// The autobase converges POINTERS ({$oid,size} per record); the signed repo is built from the BYTES
// those pointers resolve to. Nothing checked that the two agree, because ensureLocal returns a local
// file hit without ever comparing it to the pointer's $oid (overlay-repo-store.js) — so a node
// holding a SUPERSEDED copy of a record resolves it happily, never re-fetches, and signs it as truth.
//
// That is not hypothetical: on wj62 both co-owners held the SAME 139 converged pointers, both
// resolved all 139, and still produced different signed repos — they differed on exactly one record
// (app.bsky.actor.profile/self, two different avatar blob refs). The signed content therefore FLIPPED
// on every signer change, republishing the profile with a different avatar each time. Diagnosing it
// required fetching two signed CARs and diffing them by hand; this answers the same question on ONE
// node in one call, which is the difference between minutes and days.
//
// reconcileBytes cannot see this class of fault by construction: it fetches only records with NO
// local bytes (`if (have != null) continue`), so a stale-but-present file is invisible to it too.
//
// Pure read: hashes local files, reads the pointer snapshot, writes nothing, fetches nothing. Returns
// { checked, ok, stale, missing, unresolvable, tombstones, entries[] } where entries lists only the
// NON-ok records: state 'stale' (local bytes hash != the pointer's $oid — the silent divergence),
// 'missing' (pointer converged, no local bytes — normal on a lagging peer, reconcileBytes fetches it),
// or 'no-oid' (a pointer carrying no $oid, e.g. a legacy inline record under a pointer substrate).
async pointerDiag ({ prefix = '', limit = 0 } = {}) {
if (!this._recordStore || typeof this._recordStore.localOid !== 'function') return { skipped: 'no-record-store' }
let raw
try { raw = await this._listNow(prefix) } catch (e) { return { skipped: 'list-failed', error: e.message } }
let checked = 0; let ok = 0; let stale = 0; let missing = 0; let unresolvable = 0; let tombstones = 0
const entries = []
for (const [k, ptr] of Object.entries(raw)) {
const i = k.indexOf('/')
if (i <= 0 || k.startsWith('__')) continue // reserved/control key — never a record
if (isTombstone(ptr)) { tombstones++; continue }
checked++
const collection = k.slice(0, i)
const rkey = k.slice(i + 1)
const ptrOid = ptr && ptr.$oid ? String(ptr.$oid) : null
const local = await this._recordStore.localOid(collection, rkey)
let state
if (!ptrOid) { state = 'no-oid'; unresolvable++ } else if (!local) { state = 'missing'; missing++ } else if (local.oid !== ptrOid) { state = 'stale'; stale++ } else { state = 'ok'; ok++ }
if (state === 'ok') continue
if (limit && entries.length >= limit) continue
// pointerFields: the FIELD NAMES of the raw autobase value (never its content). A pointer should
// be exactly ['$oid','size']; anything else is a malformed or wrong-shaped entry, and the field
// names say which — e.g. ['$type','displayName',…] means an INLINE RECORD was written where a
// pointer belongs (a legacy-substrate write against an overlay-backed DID), which reads fine
// locally and is unfetchable by every other node. loser-club sits in exactly that state and the
// shape was not knowable from any surface, so diagnosing it meant guessing. Names only, capped.
entries.push({
key: k,
state,
pointerOid: ptrOid,
pointerSize: (ptr && ptr.size) || null,
pointerFields: (ptr && typeof ptr === 'object') ? Object.keys(ptr).slice(0, 12) : typeof ptr,
localOid: local ? local.oid : null,
localSize: local ? local.size : null
})
}
if (stale) {
console.error(JSON.stringify({
msg: 'pointerDiag: LOCAL BYTES DIVERGE FROM THE CONVERGED POINTER — this node resolves these records to superseded bytes and will sign them as truth; ensureLocal never re-fetches because a local file is present',
did: this.did, stale, checked, keys: entries.filter((e) => e.state === 'stale').map((e) => e.key).slice(0, 10)
}))
}
return { did: this.did, checked, ok, stale, missing, unresolvable, tombstones, entries }
}
// Install the tombstone epoch: from this rev on, every deletion leaves a tombstone, so the
// forward-only gate may treat a missing pointer as "not replicated" rather than "deleted" — the
// fail-safe that stops a lagging co-owner publishing a delete for a live record.
//
// ⚠ Install this only when NO deletion is pending-but-unsigned anywhere in the group: such a delete
// left no tombstone and its create frame is at/after the epoch, so the gate would classify it as
// "behind" forever and wedge the chain. Audit first — .probe-delete-audit.mjs replays every #commit
// op read-only over P2P and shows whether any delete is outstanding. Idempotent + writable-only;
// never lowers an existing epoch.
async ensureTombstoneEpoch () {
if (!(this.sb.base && this.sb.base.writable)) return { skipped: 'not-writable' }
const cur = await this.sb.get(TOMBSTONE_EPOCH).catch(() => null)
if (cur && cur.rev) return { already: cur.rev }
if (!this.head) await this.restore()
const rev = this.head ? this.head.rev : ''
await this.sb.set(TOMBSTONE_EPOCH, { rev })
return { installed: rev }
}
// ── lease: become / stay the active signer ──
claimLease () { return this.lease.claim() }
renewLease () { return this.lease.renew() }
renewOrClaimLease () { return this.lease.renewOrClaim() } // safe per-tick: renew, else claim iff free
holdsLease () { return this.lease.holds() }
// ── revocation (atproto-base-REVOCATION §6) ──
// Ban a device by its keet identity hex: records it in the shared REMOVED_SET, drops it from the
// roster, and removeWriters it (apply-side). Only takes effect if THIS device is a role-1 member —
// the apply guard authorizes by the caller's writer. After this, the banned device can neither
// write nor re-admit. NB: also PLC-rotate the shared #atproto key (root, recovery key) to kill any
// exfiltrated signing-key copy — that closes layer B; this closes layer A.
banDevice (identityHex) { return this.sb.banDevice(String(identityHex)) }
// The durable banned-identity set { identityHex: { ts } } from the shared state.
async removed () { const m = await this.sb.get(REMOVED_SET); return m || {} }
// Bump the minimum attestation epoch: after this, any device whose attestation epoch is below
// `epoch` is rejected on admission — re-attest survivors at >= epoch to keep them. Invalidates all
// stale/revoked proofs at once (atproto-base-REVOCATION §3). Only a role-1 member's bump takes.
bumpEpoch (epoch) { return this.sb.bumpEpoch(b4a.toString(encodeReceipt(epoch), 'hex')) }
// The current minimum attestation epoch (0 = no restriction).
async minEpoch () { const h = await this.sb.get(RECEIPT_KEY); return h ? receiptEpoch(b4a.from(h, 'hex')) : 0 }
// DETERMINISTIC continuation (the atproto-base merge): rev is a pure function of the CURRENT head
// — which sign() has already re-anchored to the network-latest #commit via the forward-only gate —
// so two co-owners continuing the same chain over the same finalized set produce byte-identical
// commits instead of diverging on wall-clock. Genesis (no head yet) still mints a time-plausible
// TID; an injected clock (tests) still wins for reproducibility.
_nextRev () {
if (this._clock) return tidNow(0, this._clock)
if (this.head && this.head.rev) return nextRev(this.head.rev)
return tidNow()
}
// Would sign() pass its election gate right now? For a CO-SIGNER of this DID's key (the labeler's
// LabelRoom): it must sign on exactly the node that signs commits, so it asks the same election with the
// same network-head rev — two elections that can disagree is how two nodes end up signing. Read-only.
async isSigner () {
if (!this.sb.holdsSigner) return true // no election (single writer / unit mock): sign()'s default
let net = null
if (this.firehose) net = commitLine(Object.values(await this.sb.listNow(FIREHOSE_PREFIX)).filter((f) => f && f.rev)).head
return !!this.sb.holdsSigner(undefined, net ? net.rev : null)
}
// Materialize the current linearized view into the NEXT canonical commit — only if we hold the
// lease (or force, for a bootstrap/test). Continue-from-head: prev = last commit CID, rev
// monotonic. Returns the new head, or null if we don't hold the lease or the view is unchanged.
async sign ({ force = false } = {}) {
// Lazy restore FIRST (before the election gate) so the currency-aware election knows whether WE are
// caught up: re-seed head + the diff baseline from the shared firehose so we CONTINUE the chain
// incrementally instead of re-authoring the full record set (which breaks relay/AppView MST continuity).
if (!this.head) await this.restore()
// NETWORK HEAD (highest-rev __commit__ frame) — computed ONCE, used by BOTH the signer election and the
// forward-only gate below. Non-blocking (listNow).
let frames = null
let net = null
let chain = null
if (this.firehose) {
frames = Object.values(await this.sb.listNow(FIREHOSE_PREFIX)).filter((f) => f && f.rev)
// The head of the chain, decided identically on every device (highest rev; a same-rev fork is
// broken by the lower commit CID). The old "highest rev, first seen" picked a different fork on
// different devices.
chain = commitLine(frames)
net = chain.head
}
// Feed our current currency into the presence election + announce it to peers (setSignerHead).
if (this.sb.setSignerHead) this.sb.setSignerHead(this.head ? this.head.rev : null)
// S1 election, CURRENCY-AWARE: gate on the StateBase's PRESENCE election (replaces the old __lease__
// ts-heartbeat), but exclude a co-owner BEHIND the network head (netHead=net.rev) — else the lowest-key
// -but-behind holder is gate-blocked below while the caught-up peer defers → the chain FREEZES (the
// stortron aftermath). A behind self returns here so the caught-up peer signs. A substrate without
// holdsSigner (unit mocks / a single-writer repo) defaults to signing. See signer-presence.js.
if (!force && this.sb.holdsSigner && !this.sb.holdsSigner(undefined, net ? net.rev : null)) return null
// NON-BLOCKING read (Phase-2 #6, extended to the WRITE-materialization path): sign() must not
// await a full base.update() drain — that can hang indefinitely on an un-converged replica and
// FREEZE posting (the liveness reads got getNow/listNow; this path was the last blocking one). The
// autoUpdate snapshot reflects the linearized tip incl. our own admitted writes; a just-arrived
// write not yet in the snapshot is simply materialized on the next sign trigger (eventual).
const records = await this.repo.listRecords(null, { now: true })
if (!this.head && records.length === 0) return null // fresh repo, nothing to sign yet (was implicit under the lease gate)
// ANTI-WIPE GUARD (Step 2a migration safety). A previously-signed repo (head != null) whose record
// set reads EMPTY, while the underlying overlay store still holds record FILES, is the files-only →
// pointer migration reading a not-yet-backfilled pointer set — NOT a genuine "all records deleted".
// Materializing it would advance head to an empty MST root and emit a delete-every-record commit to
// the relay (the wj62 wipe). Skip until migratePointers() populates the pointers; a genuinely empty
// repo has no record files either, so this never blocks a legitimate empty state. `force` (bootstrap/
// tests) bypasses. See docs/SYNTHESIS-storage-consensus-append-2026-07-19.md (Step 2a wipe hazard).
if (!force && this.head && records.length === 0) {
if (this._recordStore) {
const files = await this._recordStore.records()
if (files.length > 0) {
console.error(JSON.stringify({ msg: 'sign: anti-wipe guard — pointer set empty but record files present; skipping until 2a backfill', did: this.did, files: files.length }))
return null
}
}
// SECOND ARM — INLINE entries. The files check above is BLIND to a legacy repo being flipped to the
// pointer substrate: its records sit INLINE in the statebase with no files at all, so the record set
// reads empty AND files.length is 0, and the guard would wave through a delete-every-record commit.
// That is e2jev's exact state (hyper-social#16 P1). The boot-time inline migration (step 2a-0) is
// what repairs it; this is the fail-closed backstop for when that has not run or did not finish —
// an empty read with un-migrated records still in the log is never a genuine "all records deleted".
const raw = await this._listNow('').catch(() => ({}))
let inline = 0
for (const [k, v] of Object.entries(raw)) {
const i = k.indexOf('/')
if (i <= 0 || k.startsWith('__')) continue // reserved/control key — never a record
if (isTombstone(v)) continue // a real deletion, which this guard must not block
if (v && typeof v === 'object' && !v.$oid) inline++
}
if (inline > 0) {
console.error(JSON.stringify({ msg: 'sign: anti-wipe guard — pointer set empty but INLINE records present in the log; skipping until the 2a-0 inline migration converts them', did: this.did, inline }))
return null
}
}
// ── FORWARD-ONLY SIGN GATE (anti-re-fork) ──────────────────────────────────────────────────────
// Re-anchor to the NETWORK-LATEST signed commit before continuing the chain, and never sign while
// BEHIND it. A stale/rejoined signer (lowest-key election winner with a lagging view) must not cause
// either regression that produced the stortron/tabertron split-brain:
// (1) REV regression — continuing from our own stale in-memory head (or restore()'s last-by-SEQ,
// which a fork can order behind by rev) publishes a rev ≤ one already served. Fix: adopt the
// HIGHEST-REV __commit__ frame as head; _nextRev() is monotonic so the next rev exceeds it.
// (2) CONTENT regression — signing our current (partially-replicated) view forward from that head
// would DROP records the net head already served (a delete-those-records commit). Fix: replay
// the log's ops up to net head for its record key set; if our view is missing ANY, SKIP this
// sign — a later tick (after more replication) catches up. A superset (our new records ∪ net's)
// is fine and signs forward. This is keet's "a behind peer silently catches up", made safe for
// our external (relay) projection. Non-blocking (listNow); `force` (bootstrap/reconcile/tests
// that own head deliberately) bypasses, matching the anti-wipe guard convention above.
if (!force && this.firehose && net) { // frames/net computed once at the top (shared with the election)
// ⛔ ALSO on an equal rev with a different commit: commitLine broke the tie the other way, so our head
// is a fork. Re-anchoring only on a strictly higher rev left a device on the losing side of a tie
// extending its own branch forever — sibling commits at every rev, each one a new fork that reopened
// the heal (test/multi-partition-relay, split heals).
if (!this.head || net.rev > this.head.rev || (net.rev === this.head.rev && String(net.commitCid) !== this.head.commitCid.toString())) {
// (1) adopt the network-latest head so we continue FORWARD from it, never from a stale head.
const signedCommit = dagCbor.decode(b4a.from(net.commit, 'hex'))
this.head = { rev: net.rev, commitCid: CID.parse(net.commitCid), dataCid: signedCommit.data, signedCommit }
// ⛔ The diff baseline is the record set the NET HEAD commits to, not this device's current view.
// Using the view absorbed any local write the net head lacks into the signed data root with no
// op: the diff was empty, no frame was appended, and head moved to a commit nobody ever received
// (test/reanchor-unsigned-writes.test.mjs). Same replay restore() uses; the view is the fallback
// only when a referenced block is missing, as there.
this._lastRecords = (await this._recordsFromFirehose(chain.line)) || records
if (this.sb.setSignerHead) this.sb.setSignerHead(this.head.rev) // re-anchored → update our currency
}
// (2) net head's record key set = replay every #commit op up to net, in rev order. Keyed by the
// rev that last CREATED each path, so the tombstone-epoch check below can tell a legacy
// hard-delete (pre-epoch) from a pointer that simply has not replicated to us (post-epoch).
const netKeys = new Map() // path -> rev of its most recent create/update frame
for (const f of frames.filter((f) => f.rev <= net.rev).sort((a, b) => (a.rev < b.rev ? -1 : 1))) {
for (const op of (f.ops || [])) { if (op.action === 'delete') netKeys.delete(op.path); else netKeys.set(op.path, f.rev) }
}
const curKeys = new Set(records.map((r) => r.collection + '/' + r.rkey))
// A netKey missing from our resolved view is one of two very different states, and the gate must NOT
// conflate them (doing so is the DELETE-THE-TIP WEDGE — see test/delete-tip-wedge.repro):
// - LEGITIMATE DELETE: the record's POINTER was tombstoned (del-entry converged into the shared
// autobase) → the record is genuinely gone → sign the delete forward. Dropping it is the intended
// mutation, not a content regression. (Without this, a sole writer that deletes its most-recent
// record can NEVER sign the delete — the create frame keeps the key in netKeys forever.)
// - BEHIND (content regression risk): the POINTER is still linearized but its bytes aren't fetched
// into the view yet → we're lagging replication → SKIP, else we'd tell the relay to delete a
// record that is actually live. The raw pointer set (this.sb, pre-byte-resolution) distinguishes
// them: pointer ABSENT → deleted; pointer PRESENT → behind.
// Absence is AMBIGUOUS in a replicated log, so it is no longer treated as evidence of deletion:
// a TOMBSTONE ({$deleted:true}, written by RepoBase.deleteRecord) is the positive signal. Three
// states, and only the first may be signed forward:
// tombstone present -> genuinely deleted -> sign the delete forward
// live pointer, unresolved (bytes unfetched) -> behind -> SKIP
// nothing at all (the pointer itself hasn't arrived) -> behind -> SKIP <-- was misread as a
// delete, which let a lagging co-owner publish a delete for a record that is actually live.
// LEGACY: records deleted before tombstones existed left no marker, so a path whose create frame
// predates __tombstone_epoch__ keeps the old "absent == deleted" reading — otherwise those would
// wedge the chain forever (the delete-tip wedge). With no epoch set, every path is legacy and
// behaviour is byte-for-byte what it was; installing the epoch switches on the fail-safe.
const rawMap = await this._listNow('')
const rawKeys = new Set(); const tombKeys = new Set()
for (const rk of Object.keys(rawMap)) {
if (rk.indexOf('/') <= 0 || rk.startsWith('__')) continue
if (isTombstone(rawMap[rk])) tombKeys.add(rk); else rawKeys.add(rk)
}
const epoch = await this._getNow(TOMBSTONE_EPOCH).catch(() => null)
for (const [k, createdAt] of netKeys) {
if (curKeys.has(k)) continue
if (tombKeys.has(k)) continue // explicit tombstone → legitimate delete, sign it forward
if (!rawKeys.has(k) && !(epoch && epoch.rev && createdAt >= epoch.rev)) continue // legacy hard-delete
console.error(JSON.stringify({ msg: 'sign: forward-only gate — view behind network head, skipping until replicated', did: this.did, missing: k, reason: rawKeys.has(k) ? 'pointer present, bytes unreplicated' : 'pointer not replicated', netRev: net.rev, have: curKeys.size, need: netKeys.size }))
return null
}
}
const rev = this._nextRev()
const prev = this.head ? this.head.commitCid : null
const out = materialize(records, { did: this.did, rev, prev, privateKey: this.privateKey })
// Forks to heal (atproto-room healPlan): a consumer behind a relay may hold ANY mix of branches, so the
// heal restates every path any frame touched since the earliest uncovered fork point.
const plan = (!force && chain) ? healPlan(chain) : null
if (this.head && out.dataCid.equals(this.head.dataCid) && !plan) return null // no new intents
// persist the #commit into the shared firehose log (option a) BEFORE advancing head, so every
// device + the membrane sees each signed commit. The diff is _lastRecords -> records, or for a heal
// the asserting state built by _healBefore.
if (this.firehose) {
const since = this.head ? this.head.rev : null
const slice = plan ? this._healSlice(plan, records) : null
const write = buildCommitWrite(slice ? this._healBefore(slice.paths, records) : this._lastRecords, records, { did: this.did, rev, prev, since, privateKey: this.privateKey })
if (slice && slice.final) {
write.heals = plan.heals
write.asserts = plan.from
if (plan.partsFrom) write.partsFrom = plan.partsFrom
} else if (slice) write.restates = plan.from
// Only append a #commit frame when the diff is non-empty. A 0-op frame can only arise from the
// rare restart-with-unsigned-pending-write edge (baseline == current yet head.dataCid differs);
// emitting it would be a malformed commit. head still advances below so we don't re-trigger.
if (write.ops.length > 0) {
// seq is informational now: two partitioned signers both write count+1 here.
const existing = await this.sb.listNow(FIREHOSE_PREFIX)
await this.sb.set(frameKey(rev, write.commitCid), serializeEntry(write, Object.keys(existing).length + 1))
}
}
this._lastRecords = records
this.head = { rev, commitCid: out.commitCid, dataCid: out.dataCid, signedCommit: out.signedCommit, recordBlocks: out.recordBlocks }
if (this.sb.setSignerHead) this.sb.setSignerHead(this.head.rev) // announce our advanced currency to peers
return this.head
}
// The CHAIN of signed #commit frames, in rev order — the sequence the membrane relays as
// subscribeRepos and restore() replays. Fork frames are excluded (commitLine); every device reads the
// same entries via replication and decides the same chain. It used to be every frame sorted by seq,
// which two partitioned signers both assign as count+1.
async firehoseLog () {
return commitLine(await this.firehoseFrames()).line
}
// Every signed frame, forks included — for a consumer that has to tell a fork it already projected
// from a log that restarted (serve-membrane).
async firehoseFrames () {
const map = await this.sb.list(FIREHOSE_PREFIX)
return Object.values(map).filter((f) => f && f.rev)
}
// The "before" state a heal commit diffs from, so that its ops RESTATE every path in the plan:
// a record in the merged set -> `update` (never `create`: the reference bsky AppView ignores a create
// for a record it already holds, so a consumer holding a stale copy would keep it). Where the chain's
// own previous value differs it is used; otherwise a fixed placeholder CID stands in, so the diff
// still yields an update.
// a path absent from the merged set -> `delete` (a no-op for a consumer that never had it).
// Every other path keeps the chain's previous value, so ordinary changes diff as usual. buildCommitWrite
// still proves the ops invert to this state's root; only `prevData` names a state no consumer held,
// which the relay logs and accepts (indigo verify.go:140-143, warn-only even in strict mode).
// Which paths THIS commit restates. indigo drops a #commit with more than MAX_COMMIT_OPS ops, so the ops
// this commit carries anyway (ordinary changes since the last commit) come out of the budget and the
// rest of the window waits for the next commit, in sorted order so every co-owner picks the same slice.
// `final` when the remainder fits: that commit asserts the window. An empty remainder (partials already
// restated everything, the final never landed) still restates one path, so the final has an op and is
// appended — a 0-op commit is never written, and the window would stay open forever.
_healSlice (plan, records) {
const now = new Map(records.map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
const was = new Map(this._lastRecords.map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
// Every path that changed since the last commit is an op whether or not the slice restates it, so a
// pending path that changes anyway rides free and the budget is what remains after ALL of them.
const changed = new Set()
for (const p of new Set([...now.keys(), ...was.keys()])) if (now.get(p) !== was.get(p)) changed.add(p)
let pending = [...plan.paths].sort()
if (pending.length === 0) pending = [[...plan.window].sort()[0]]
const free = pending.filter((p) => changed.has(p))
const rest = pending.filter((p) => !changed.has(p))
const budget = this._maxCommitOps - changed.size
if (budget < 0) return null // ordinary changes alone fill this commit; the heal continues on the next
if (rest.length <= budget) return { paths: new Set(pending), final: true }
return { paths: new Set(free.concat(rest.slice(0, budget))), final: false }
}
_healBefore (paths, records) {
const merged = new Map()
for (const r of records) merged.set(r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString())
const before = new Map()
for (const r of this._lastRecords) before.set(r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString())
for (const p of paths) {
const m = merged.get(p)
const was = before.get(p)
before.set(p, (was !== undefined && was !== m) ? was : HEAL_PLACEHOLDER)
}
return [...before].map(([path, cid]) => { const i = path.indexOf('/'); return { collection: path.slice(0, i), rkey: path.slice(i + 1), cid } })
}
// Re-seed head + the diff baseline from the shared __commit__ log after a (re)start. Without this,
// a reboot resets `head`/`_lastRecords` and the next sign re-authors the FULL record set as creates
// (the ops=[9,9,..,12] firehose bug) — breaking relay/AppView MST continuity. The head is the LAST
// signed #commit (decoded from the log, so its exact commitCid/dataCid are preserved for the next
// prev-link); the baseline is the current (quiescent) record set — the sign loop runs every couple
// seconds so at any restart it equals the record set that head represents. Returns the restored head,
// or null if nothing has been signed yet (truly fresh — the caller then authors the first commit).
async restore () {
const log = await this.firehoseLog()
if (!log.length) return null
const last = log[log.length - 1]
const signedCommit = dagCbor.decode(b4a.from(last.commit, 'hex'))
this.head = { rev: last.rev, commitCid: CID.parse(last.commitCid), dataCid: signedCommit.data, signedCommit }
// Re-seed the diff baseline from the HEAD's firehose record set, NOT the current pointer view. A record
// deleted-but-not-yet-signed (the DELETE-THE-TIP WEDGE) is ABSENT from the view yet MUST remain in the
// baseline so the next sign emits the proper delete op — and its bytes are already gone from the overlay
// store, so the firehose #commit blocks are the only surviving copy. Falls back to the view if any
// referenced block is missing (partial replication); the forward-only gate still guards a real regression.
this._lastRecords = (await this._recordsFromFirehose(log)) || await this.repo.listRecords()
return this.head
}
// Replay the firehose (#commit ops in rev order, resolving each create/update to its record block) to
// reconstruct the record set the HEAD commits to — the surviving copy of a record whose overlay bytes
// were deleted. Returns [{collection, rkey, record}], or null if a referenced block is missing (the
// caller then degrades to the pointer view).
async _recordsFromFirehose (log) {
const byKey = new Map()
for (const entry of log.slice().sort((a, b) => (a.rev < b.rev ? -1 : 1))) {
const blocks = new Map((entry.blocks || []).map((b) => [b.cid, b.bytes]))
for (const op of (entry.ops || [])) {
if (op.action === 'delete') { byKey.delete(op.path); continue }
const hex = op.cid ? blocks.get(op.cid) : null
if (!hex) return null // referenced record block absent — cannot faithfully reconstruct
byKey.set(op.path, dagCbor.decode(b4a.from(hex, 'hex')))
}
}
const out = []
for (const [path, record] of byKey) { const i = path.indexOf('/'); out.push({ collection: path.slice(0, i), rkey: path.slice(i + 1), record }) }
return out
}
// ── adopt an existing DID's repo into the swarm (continuity migration) ──
// Import an existing single-tenant repo's records + head so the swarm CONTINUES its commit chain
// (next sign: prev = the adopted commit, prevData = the adopted data root, rev monotonic) rather
// than re-authoring it. Hard-guards continuity: the imported records MUST reproduce the adopted
// commit's data root (else the import is incomplete and would fork/re-author — the migration wall,
// atproto-pds-ref §2/§7). Only meaningful on a writable (creator) device. records: [{collection,
// rkey, record}]; head: { rev, commitCid, dataCid, signedCommit? }.
async adopt (records, head) {
if (!head || !head.commitCid || !head.dataCid || !head.rev) throw new Error('adopt: head { rev, commitCid, dataCid } required')
for (const r of records) await this.repo.putRecord(r.collection, r.rkey, r.record)
const cur = await this.repo.listRecords()
const mv = materialize(cur, { did: this.did, rev: head.rev, prev: null, privateKey: this.privateKey })
if (!mv.dataCid.equals(head.dataCid)) throw new Error('adopt: imported records do not reproduce the adopted data root — import incomplete (would re-author wj62)')
this._lastRecords = cur
this.head = { rev: head.rev, commitCid: head.commitCid, dataCid: head.dataCid, signedCommit: head.signedCommit || null }
return this.head
}
// ── emit the ADOPTED head as a baseline #commit frame (adopt frontability) ──
// adopt() sets head + records but writes NO __commit__ frame, and the adopted repo's data root is
// unchanged so the next sign() is a no-op (returns null on out.dataCid == head.dataCid). Result: an
// adopted account with no subsequent write has an EMPTY firehose log, so serveMembrane.sync() early-
// returns (log.length <= seq.length) and projects NEITHER the repo NOR the firehose core → the keyless
// bridge can't front it (getRepo/getLatestCommit both fail) until the first record-changing post emits
// a frame. This closes that gap: write ONE baseline frame for the CURRENT head so the membrane projects
// immediately. buildCommitWrite over ([] -> current records) with the adopted head's rev+prev reproduces
// the adopted commit byte-for-byte (RFC6979-deterministic), so the served commitCid IS the canonical
// head. Idempotent (no-op if the log already has frames). Writable creator only. Returns the head or null.
async emitBaseline () {
if (!this.firehose || !this.head || !this.head.signedCommit) return null
const existing = await this.sb.listNow(FIREHOSE_PREFIX)
if (Object.keys(existing).length > 0) return null // already have frames — no baseline needed
const prev = this.head.signedCommit.prev || null
const write = buildCommitWrite([], this._lastRecords, { did: this.did, rev: this.head.rev, prev, since: null, privateKey: this.privateKey })
if (!write.commitCid.equals(this.head.commitCid)) throw new Error('emitBaseline: rebuilt commit CID != adopted head — refusing to emit an inconsistent baseline frame')
await this.sb.set(frameKey(this.head.rev, write.commitCid), serializeEntry(write, 1))
return this.head
}
// ── PART D: reconcile a divergent line onto an authoritative signed CAR (fork recovery) ──
// An autobase-level fork of a shared DID (two divergent authored lines, SAME base.key) CANNOT be
// merged in place — the oplog is append-only and a fresh NAMESPACE re-derives the same primaryKey
// writer (pear-git #4). The recovery is to re-adopt the NETWORK-AUTHORITATIVE signed repo (its CAR)
// into a FRESH line, then UNION any record that only landed on the LOSING line so nothing is lost.
// Run this on a device backed by a FRESH corestore (own primaryKey → clean writer). The caller passes
// the parsed CAR (`{records, head}` from readAdoptRepo) + the loser's record set. Steps: adopt the CAR
// (the anti-re-author guard proves the records reproduce its data root), union in each loser-only
// record (by collection/rkey — additive ONLY, never a delete), then sign FORWARD (rev > CAR rev,
// prev = CAR commit) so the relay accepts a monotonic continuation. Returns { head, adopted, unioned,
// total }. Deletion story: NOTHING is dropped — a record on the losing line the winner lacks is kept.
async reconcileFromCar ({ records, head } = {}, loserRecords = []) {
if (!Array.isArray(records) || !head) throw new Error('reconcileFromCar: { records, head } (from readAdoptRepo) required')
await this.adopt(records, head) // head = CAR head, _lastRecords = CAR records; guards against re-author
const have = new Set(records.map((r) => r.collection + '/' + r.rkey))
let unioned = 0
for (const r of loserRecords) {
if (!r || !r.collection || !r.rkey) continue
const key = r.collection + '/' + r.rkey
if (!have.has(key)) { await this.putRecord(r.collection, r.rkey, r.record); unioned++ } // loser-only → KEEP it
}
// Sign FORWARD only when we actually added records; else the adopted CAR head IS the reconciled
// state (loser was a strict subset) and re-signing an identical set would be a no-op.
const newHead = unioned > 0 ? await this.sign({ force: true }) : this.head
return { head: newHead, adopted: records.length, unioned, total: records.length + unioned }
}
// ── export the current signed repo as an atproto CAR (com.atproto.sync.getRepo shape) ──
// The input to reconcileFromCar on a PEER (fork recovery: fetch the authoritative node's CAR → adopt
// it into a fresh line) and to any getRepo consumer. Rebuilds the MST from the current record set
// (DAG-CBOR is canonical, so re-encoding is byte-exact) and reuses read-view's getRepoFromCommit,
// which ASSERTS the rebuilt MST root == the signed head's data root — refusing to serve a repo that
// doesn't match its own signature. Uses the head's EXISTING signature; never re-signs. Returns
// { carBytes, commitCid, dataCid, rev, recordCount }.
async exportCar () {
if (!this.head) await this.restore()
if (!this.head || !this.head.signedCommit) throw new Error('exportCar: no signed head to export (nothing signed yet)')
const recs = await this.repo.listRecords()
// Shim the record set into the { records()->[{key,bytes,record}] } shape getRepoFromCommit consumes.
const shim = { records: async () => recs.map((r) => ({ key: r.collection + '/' + r.rkey, bytes: dagCbor.encode(r.record), record: r.record })) }
const { carBytes, commitCid, dataCid } = await getRepoFromCommit(shim, this.head.signedCommit)
return { carBytes, commitCid: commitCid.toString(), dataCid: dataCid.toString(), rev: this.head.rev, recordCount: recs.length }
}
// ── ONE-TIME MIGRATION REPAIR ONLY — not routine recovery (atproto-base) ──
// ⚠ This REWRITES the firehose log, which is NOT what atproto expects: atproto firehoses are
// APPEND-ONLY with forward-only consumer cursors, and the idiomatic recovery from bad/discontinuous
// frames is to emit a #sync re-baseline → the relay/AppView resync via getRepo (the desync→resync
// path; see atproto specs/sync + docs/firehose-recovery.md). The bridge already does this
// (atproto-bridge control.js `emit-sync` + `set-seq-base`), and it is what heals a relay in
// practice. Reach for THAT, not this. resetFirehose exists solely to repair a log that accumulated
// MALFORMED frames from before the restore/dedup fixes (full-set re-authors whose prevData doesn't
// chain) — a one-shot historical cleanup during the wj62 migration. With restore()+the bridge dedup
// fix in place, new frames stay well-formed, so this should never be needed again.
//
// Mechanics: clear the __commit__ log, re-seed head + baseline from a known-good migration head (the
// single-tenant adoptCar — the last commit the relay validly holds), then a forced sign diffs the
// CURRENT record set against it → one incremental #commit continuing the chain. Writable only;
// marker-guarded (__firehose_reset__) so it runs exactly once. Returns the clean head.
async resetFirehose (baselineRecords, baselineHead) {
if (!baselineHead || !baselineHead.commitCid || !baselineHead.dataCid || !baselineHead.rev) throw new Error('resetFirehose: baselineHead { rev, commitCid, dataCid } required')
const log = await this.sb.list(FIREHOSE_PREFIX)
for (const key of Object.keys(log)) await this.sb.del(key)
this.head = { rev: baselineHead.rev, commitCid: baselineHead.commitCid, dataCid: baselineHead.dataCid, signedCommit: baselineHead.signedCommit || null }
this._lastRecords = baselineRecords
return this.sign({ force: true })
}
// ── control surface: submit a write, observe convergence ──
// applyWrite mirrors the single-tenant apply-write (lib/peer.js _attachRpc): a client/app hands a
// repo op to ANY device, which appends it to the shared intent-log; whichever device holds the
// lease materializes it into the signed chain on its next pass. Returns the post-write record
// count so a caller can confirm the intent landed on THIS device (it replicates to the others).
async applyWrite ({ action = 'create', collection, rkey, record } = {}) {
if (action === 'delete') await this.repo.deleteRecord(collection, rkey)
else await this.repo.putRecord(collection, rkey, record)
// recordCount from RAW pointer keys (StateBase), NOT this.repo.listRecords(): the latter RESOLVES every
// record's bytes via ensureLocal (an overlay fetch-by-oid that HANGS when a co-owner lacks the bytes
// locally) — which wedged apply-write for 20s ("UDS call timed out"). The write is already durable
// (putRecord appended the pointer); the count is only an ack, so count pointers without resolving bytes.
// this.sb.list (StateBaseV3, drained) gives the RAW pointer set — no PointerSubstrate byte resolution
// (the original went through this.repo.listRecords, resolving bytes). base.update itself was never the
// hang (get-state's {now} read skips it and still hung on ensureLocal); dropping the resolution is the fix.
let recordCount = 0
try { for (const k of Object.keys(await this.sb.list(''))) if (k.indexOf('/') > 0 && !k.startsWith('__')) recordCount++ } catch {}
return { ok: true, action, collection, rkey, recordCount }
}
// com.atproto.server.getServiceAuth — mint an ES256K inter-service JWT signed by the shared
// #atproto key (the SAME key that signs the repo's commits), so the KEYLESS membrane can proxy
// user-scoped AppView reads (getTimeline/listNotifications/getProfiles) authenticated AS this
// account. { iss:did, aud:<appview did>, exp, lxm:<method> }. signAssertion IS the atproto
// service-auth shape (atproto-identity/lib/auth.js). exp is short (default 10min, ≤1h w/ lxm).
getServiceAuth ({ aud, lxm, exp } = {}) {
if (!aud) throw new Error('getServiceAuth: aud required')
const now = Math.floor(Date.now() / 1000)
const e = exp || (now + 600)
const payload = { iss: this.did, aud, exp: e }
if (lxm) payload.lxm = lxm
return { token: signAssertion(payload, this.privateKey), exp: e }
}
// delegate — authorize a browser SUB-KEY to act as this DID for web sign-in: sign an ES256K
// assertion { iss:did, sub:<subKey hex>, exp } with the shared #atproto key. This is the atproto
// analog of adding a device — the DID's own key vouches for the sub-key, and a keyless verifier
// (favela web-auth) checks it against the DID's PLC-published key. The browser then signs login
// challenges with the sub-key. subKey = a 33-byte compressed secp256k1 pubkey (hex).
//
// PERSISTENT + REVOCABLE (Option 3): ttlSec defaults to effectively-permanent (10y) — the exp is a
// backstop, NOT the control. A delegated browser stays valid until EXPLICITLY revoked (revokeDevice),
// matching pgv3's persistent-until-revoked model. We ALSO record the sub-key in the device registry
// (addDevice) so it can be listed + revoked; revocation is enforced by the verifier (isDeviceRevoked),
// not by expiry. Privileged: gated by serveDevice's canWrite.
async delegate ({ subKey, ttlSec = 10 * 365 * 24 * 3600, label = null } = {}) {
if (!subKey || !/^[0-9a-fA-F]{66}$/.test(String(subKey))) throw new Error('delegate: subKey (33-byte compressed secp256k1 pubkey hex) required')
const sub = String(subKey).toLowerCase()
const now = Math.floor(Date.now() / 1000)
const exp = now + Number(ttlSec)
const delegation = signAssertion({ iss: this.did, sub, iat: now, exp }, this.privateKey)
// Record in the persistent registry (reserved StateBase key; never public). FIRE-AND-FORGET: addDevice
// does sb.set → _ensureWritable → base.update, which can BLOCK on an un-converged plain writer. The
// delegation is cryptographically valid regardless, so we must NOT block (and thus hang pairing) on the
// write — record in the background if/when the base is writable; listing/revocation pick it up then.
this.addDevice(sub, { label }).catch((e) => console.error(JSON.stringify({ msg: 'addDevice on delegate failed (delegation still valid)', error: e.message })))
return { did: this.did, subKey: sub, delegation, exp }
}
// state — the observable snapshot the read-RPC returns: who we are, whether we're the active
// signer + who holds the lease, the last signed head, and the current linearized record set
// (keys only). This is how a peer/test verifies convergence: submit to device A, read state on
// device B, compare the record sets.
async state () {
// NON-BLOCKING (Phase-2 #6): get-state must NEVER hang — it is the liveness probe favela-web +
// the CLI use. Both reads below avoid base.update(): listRecords({now}) reads the autoUpdate
// snapshot, and lease.current() is snapshot-based. On the live tabertron (writable=false,
// un-converged) the blocking versions hung this call for >8s.
// RAW pointer scan — never RESOLVE bytes here. This is the same trap applyWrite escaped (see its
// note below): under PointerSubstrate, repo.listRecords() calls store.ensureLocal() per record, an
// overlay fetch-by-oid that BLOCKS when this node lacks the bytes locally. That made get-state —
// the liveness probe favela-web and the CLI depend on — time out on wj62 (~129 records) while
// 3-record loser-club answered in 35ms, with members-diag responding in 3ms on both. collection and
// rkey come straight off the pointer key, so the returned shape is unchanged.
const records = []
for (const [k, v] of Object.entries(await this._listNow(''))) {
const i = k.indexOf('/')
if (i <= 0 || k.startsWith('__')) continue
if (isTombstone(v)) continue // deleted records are absent from state, as before
records.push({ collection: k.slice(0, i), rkey: k.slice(i + 1) })
}
const cur = await this.lease.current()
return {
did: this.did,
self: this.lease.self,
// ⚠ LEGACY. The __lease__ ts-heartbeat was REPLACED by the StateBase presence election (see the
// gate in sign(), ~line 358). These three are reported for continuity only and do NOT indicate
// whether this node signs — holdsLease is False on healthy signing nodes. Read holdsSigner.
holdsLease: await this.lease.holds(),
leaseHolder: cur ? cur.holder : null,
leaseEpoch: cur ? cur.epoch : null,
// The field that ACTUALLY gates signing: the presence election. Null when the substrate has no
// presence (unit mocks / single-writer), where sign() defaults to signing.
holdsSigner: (() => { try { return this.sb.holdsSigner ? !!this.sb.holdsSigner() : null } catch { return null } })(),
// ⛔ THE GATE'S VIEW (2026-10-14: both loser-club co-owners said holdsSigner:true). holdsSigner above asks
// WITHOUT the network head; sign() and the label tick ask WITH it (isSigner). Report both, with who is
// present and which heads are known, so a split election is a read, not an inference. Non-blocking (listNow).
isSigner: await this.isSigner().catch(() => null),
election: await (async () => {
if (!this.sb.signerState) return null
try {
let netRev = null
if (this.firehose) { const h = commitLine(Object.values(await this.sb.listNow(FIREHOSE_PREFIX)).filter((f) => f && f.rev)).head; netRev = h ? h.rev : null }
return { netRev, withHead: this.sb.signerState(undefined, netRev), noHead: this.sb.signerState(), presence: this.sb.presenceDiag ? this.sb.presenceDiag() : null }
} catch { return null }
})(),
indexers: (this.sb.base && this.sb.base.linearizer && this.sb.base.linearizer.indexers) ? this.sb.base.linearizer.indexers.length : null,
writable: !!(this.sb.base && this.sb.base.writable),
head: this.head ? { rev: this.head.rev, commitCid: this.head.commitCid.toString() } : null,
recordCount: records.length,
records: records.map((r) => ({ collection: r.collection, rkey: r.rkey }))
}
}
// Read-only membership diagnostic (NON-BLOCKING — getNow + flags only, no listRecords/base.update).
// Answers WHY writer admission fails: whether __members__/__devices__ are actually in the linearized
// view, plus the creator's projection state (_membersProjected / _roster / writable / base lengths).
async membersDiag () {
const sb = this.sb
let members = null; let devices = null
try { members = await sb.getNow('__members__') } catch {}
try { devices = await sb.getNow('__devices__') } catch {}
const roster = Array.isArray(sb._roster)
? sb._roster.map((r) => ({ key: String((r && (r.key || r.devicePubkey || r.identity || r.memberId)) || '').slice(0, 12), role: r && r.role }))
: (sb._roster || null)
return {
did: this.did,
writable: !!(sb.base && sb.base.writable),
// `_isCreator` (atproto-room 0.2.16), NOT `_amCreator`. The latter is a prototype METHOD, so
// `!!sb._amCreator` was `!!<function>` — this field reported `true` for every node, creator or not,
// and could never express the negative case. It was cited as evidence of a role flip in a live
// incident. `_isCreator` is the decision `_open` actually made.
amCreator: sb._isCreator === undefined ? null : !!sb._isCreator,
forcedRole: sb._forceCreator ? 'forced-creator' : (sb._forceNonCreator ? 'forced-non-creator' : null),
baseLength: sb.base ? sb.base.length : null,
indexedLength: sb.base ? sb.base.indexedLength : null,
// PER-WRITER BREAKDOWN. baseLength/indexedLength are AGGREGATES, and an aggregate that looks wrong
// cannot say WHO made it wrong. Re-joining loser-club left stortron frozen at indexed 34681 /
// base 34727 — a 46-block gap that was unanswerable from outside the process, because no surface
// said which writer those blocks came from; it could not be told apart from ordinary replication
// lag. The same shape at 3728 blocks turned out to be fork residue from an earlier attempt.
//
// `seen` is how much of a writer's core the linearized view has CONSUMED; `authored` is how much
// that writer has actually written. pending = authored-but-not-in-the-view, which is the number
// that explains a gap. A removed writer still carrying pending blocks, or a writer present on one
// node and absent on its peer, is the signature worth looking for.
writers: (sb.base && sb.base.activeWriters)
? [...sb.base.activeWriters].map((w) => {
const key = (w.core && w.core.key) ? b4a.toString(w.core.key, 'hex') : null
const localKey = (sb.base.local && sb.base.local.key) ? b4a.toString(sb.base.local.key, 'hex') : null
return {
key: key ? key.slice(0, 12) : null,
authored: w.core ? w.core.length : null,
seen: w.length,
pending: w.core ? Math.max(0, w.core.length - w.length) : null,
indexer: !!w.isActiveIndexer,
removed: !!w.isRemoved,
local: !!(key && localKey && key === localKey)
}
})
: null,
membersProjected: !!sb._membersProjected,
rosterLen: Array.isArray(sb._roster) ? sb._roster.length : (sb._roster ? -1 : 0),
roster,
membersInView: members || null,
devicesInView: devices || null
}
}
// Force a re-projection of __members__/__devices__ (idempotent, re-appends the retained roster). Clears
// the _membersProjected latch so the creator re-projects into the view even if a prior projection was
// re-linearized away (the false-positive that leaves admission stuck at "no __members__ in view"). A
// no-op off the writable creator (projectMembers returns false when not writable).
async reprojectMembers () {
const sb = this.sb
sb._membersProjected = false
let detail
try { detail = await sb.projectMembers(sb._roster) } catch (e) { detail = { error: e.message } }
let members = null
try { members = await sb.getNow('__members__') } catch {}
return { reprojected: detail === true, detail, writable: !!(sb.base && sb.base.writable), membersInView: members || null }
}
// ── private preferences (bsky-login onboarding: birthdate etc.) ──
// Stored in the shared statebase under a reserved key — NEVER leaks to the public repo/firehose
// (RepoBase skips "__"-prefixed keys). Any writable device writes; the autobase linearizes it. The
// atproto getPreferences/putPreferences shape is { preferences: [...] }.
async getPreferences () { const p = await this.sb.get(PREFS_KEY); return { preferences: (p && p.preferences) || [] } }
async putPreferences ({ preferences } = {}) { await this.sb.set(PREFS_KEY, { preferences: preferences || [] }); return {} }
// ── device registry (Option 3) — delegated browser sub-keys as reserved StateBase entries.
// Reserved (__-prefixed) → NEVER in the public repo/firehose (RepoBase skips them). Replicated to
// every DID device, so the linearized view is authoritative + revocation propagates. This is the
// atproto-keyless analog of hyper-identity's IdentityRegistry device records + `revoked:` tombstone,
// but it rides the StateBase we already run (no second autobase). Enforced by the verifier via the
// membrane RPCs (serve-device: list-devices / revoke-device / is-device-revoked).
async addDevice (subKey, { label = null } = {}) {
const sk = String(subKey).toLowerCase()
if (!/^[0-9a-f]{66}$/.test(sk)) throw new Error('addDevice: subKey (33-byte compressed secp256k1 hex) required')
const existing = await this.sb.get(DEVICE_PREFIX + sk).catch(() => null) // preserve addedAt on re-delegate
await this.sb.set(DEVICE_PREFIX + sk, { subKey: sk, label: label || (existing && existing.label) || null, addedAt: (existing && existing.addedAt) || Date.now() })
return { subKey: sk }
}
async revokeDevice (subKey, { reason = null } = {}) {
const sk = String(subKey).toLowerCase()
if (!/^[0-9a-f]{66}$/.test(sk)) throw new Error('revokeDevice: subKey (33-byte compressed secp256k1 hex) required')
await this.sb.set(DEVICE_REVOKED_PREFIX + sk, { subKey: sk, revokedAt: Date.now(), reason }) // tombstone (soft, reversible)
return { subKey: sk, revoked: true }
}
async restoreDevice (subKey) {
const sk = String(subKey).toLowerCase()
await this.sb.del(DEVICE_REVOKED_PREFIX + sk) // remove the tombstone
return { subKey: sk, revoked: false }
}
async isDeviceRevoked (subKey) {
const sk = String(subKey).toLowerCase()
// NON-BLOCKING read (getNow) — a revocation check MUST NOT hang the login flow if base.update() is
// stalled on an un-converged replica. Reads the current materialized view (autoUpdate HyperDB).
return !!(await this.sb.getNow(DEVICE_REVOKED_PREFIX + sk).catch(() => null))
}
async listDevices () {
const devs = await this.sb.listNow(DEVICE_PREFIX).catch(() => ({}))
const revoked = await this.sb.listNow(DEVICE_REVOKED_PREFIX).catch(() => ({}))
const revById = {}
for (const r of Object.values(revoked)) if (r && r.subKey) revById[r.subKey] = r
return Object.values(devs).map((d) => ({
subKey: d.subKey,
label: d.label || null,
addedAt: d.addedAt || null,
revoked: !!revById[d.subKey],
revokedAt: revById[d.subKey] ? revById[d.subKey].revokedAt : null
}))
}
}