/**
* state-base.js — StateBaseV3: attestation-gated, encrypted multi-writer room (HyperDB view).
*
* The core substrate of atproto-room: an Autobase whose view is a HyperDB instance; ops are
* hyperdispatch-encoded and routed through a generated Router to handlers that write into HyperDB
* collections inside a per-apply transaction. Admission is cryptographic + in-apply (a joiner
* appends an optimistic block carrying a keet attestation proof; apply verifies it resolves to a
* device attested to the room's root identity + a projected role, then addWriters) — NOT
* owner-mediated. The base key is derived deterministically from the room's members-only
* encryptionKey and opened encrypted, so non-members can neither derive nor read it.
*
* Lineage: originally reskinned from keet-daemon/hyper-tps's per-room gamestate StateBaseV3;
* extracted here as the shared room engine consumed by hyper-proto-peer (repo devices) and the
* co-authored labeler. The class name StateBaseV3 is preserved to avoid churn across consumers.
*/
import Corestore from 'corestore'
import Autobase from 'autobase'
import Hypercore from 'hypercore'
import Verifier from 'hypercore/lib/verifier' // encodeManifest/decodeManifest for the keet-writer admission proof (A)
import Hyperswarm from 'hyperswarm'
import HyperDB from 'hyperdb'
import crypto from 'hypercore-crypto'
import fs from 'bare-fs'
import path from 'bare-path'
import process from 'bare-process'
import hyperdbDefinition from '../spec/state-base/contract/hyperdb/index.js'
import dispatch from '../spec/state-base/contract/hyperdispatch/index.js'
import { SignerPresence } from './signer-presence.js'
import { attachPresence } from './signer-presence-channel.js'
const { Router, encode } = dispatch
const STATE_VERSION = 3
// Key-derivation namespace. Renamed from the legacy 'hyper-tps/state/v3' (its keet/hyper-tps
// lineage) to 'atproto-room/state/v3' on the extraction. This is LOAD-BEARING: deriveGsKeyPair
// mixes it into the room's base key, so the rename yields a fresh base key for every room. Existing
// rooms are carried across by the re-adopt migration (the writable creator re-seeds the DID's repo
// records from the adopt bundle; the signed atproto chain continues from the adopted head — see
// hyper-proto-peer device-node.js bootSwarmDevice). Never change this string again without a
// migration story.
const GS_NAMESPACE = Buffer.from('atproto-room/state/v' + STATE_VERSION, 'utf-8')
// The pre-extraction namespace. A room created before the 'hyper-tps/state/v3' -> 'atproto-room/
// state/v3' rename derives its base key from this string. Kept solely so the one-time in-place
// re-adopt migration (_migrateFromLegacyNamespace) can open such a room and seed the renamed one.
const LEGACY_GS_NAMESPACE = Buffer.from('hyper-tps/state/v' + STATE_VERSION, 'utf-8')
const ENTRIES = '@state/entries'
// Reserved op key: a designated admin anchor claims the SOLE indexer slot
// (move-indexer). Carried as the `key` of an optimistic admit op; never inserted.
const MOVE_INDEXER = '__move_indexer__'
// Reserved op key: an admin GROWS the indexer set by adding ITSELF as an indexer
// (keet members/index.js op D `start-indexing`) — NO demote of anyone. This is the
// quorum-resilience path: with >=3 admin-indexers, autobase majority quorum
// ((n>>>1)+1, consensus.js:15) finalizes even when the genesis/creator is offline.
// Proven by test/probe-indexer-quorum.mjs (autobase 7.28.1 holds >2 indexers fine).
const START_INDEXING = '__start_indexing__'
// Reserved op key: BAN a device (atproto-base-REVOCATION.md §3/§6). Value = the target device's
// keet identity hex. Only a role-1 member may issue. Records the target in the durable REMOVED_SET,
// drops it from __members__, and removeWriters its writer(s) — so it can neither write now nor
// re-admit later (the Keet handleRemoveMember + add-guard mechanism). Never inserted as an entry.
const REMOVE_DEVICE = '__remove_device__'
// Reserved op key: ADMIT a member into the room, authored by an EXISTING role-1 member. Value =
// JSON { id: <keet identity hex>, role }. This is keet's addDevice/acceptJoinRequest shape — the room
// decides who belongs, and an existing member writes that decision INTO the room.
//
// It replaces the only mechanism we had: `projectMembers`, which reprojects a roster read from the
// CREATOR's local config file. That made membership a per-machine fact — a co-owner could not admit
// anyone (tabertron sits at rosterLen:0, membersProjected:false and is structurally unable to help),
// and growing a roster on the wrong box did nothing at all (hyper-social #40 §6). Authorization is the
// AUTHOR's role verified in apply, exactly like REMOVE_DEVICE below, so a non-member cannot use it to
// admit itself or anyone else.
//
// A BANNED identity is NOT re-admitted by this op — REMOVED_SET stays authoritative, or a ban would be
// reversible by any member simply re-adding the target. Un-banning is deliberately a separate act.
const ADD_MEMBER = '__add_member__'
// Room ABI. Bumped when an op changes what APPLY does, because apply must be deterministic across
// every node or their views diverge. ADD_MEMBER is exactly such a change: a node that predates it does
// not treat `__add_member__` as reserved, so it INSERTS it as an ordinary entry while a current node
// inserts nothing — same log, two different views. Keet gates this class of change with an in-band ABI
// stamp and refuses to proceed under an unsupported one (skills/keet-abi-upgrades; pear-git-v3 does the
// same for its org schema). We cannot retrofit old nodes, so the gate is on the AUTHOR: refuse to emit
// an op no current writer can apply identically.
const ROOM_ABI = 1
// Per-writer capability stamp: `__abi__/<writer key hex>` -> { v }. An ORDINARY entry, deliberately —
// old nodes store it byte-identically and stay convergent; they simply never write one of their own,
// which is precisely the signal we need.
const ABI_PREFIX = '__abi__/'
// Durable set of banned device identities (hex → { ts }). The admission guard checks it so a
// removed identity cannot (re)admit even with a valid attestation.
const REMOVED_SET = '__removed__'
// Reserved op key: set the MINIMUM attestation epoch (value = a keet receipt, hex). Only a role-1
// member may issue. Admission rejects any attestation whose epoch is below it — bump it + re-attest
// survivors to invalidate ALL stale proofs at once (atproto-base-REVOCATION §3). Never inserted.
const SET_EPOCH = '__set_epoch__'
// Durable current minimum-epoch receipt (hex). Absent → no epoch restriction.
const RECEIPT_KEY = '__receipt__'
function toBuf (k) {
if (Buffer.isBuffer(k)) return k
const s = String(k)
return /^[0-9a-fA-F]{64}$/.test(s) ? Buffer.from(s, 'hex') : Buffer.from(s)
}
// Normalize an identity key to lowercase hex (members keys arrive 64-hex from
// router._stateBaseProviders, which z32-decodes memberIds to hex).
function normKeyHex (k) {
if (!k) return null
if (Buffer.isBuffer(k)) return k.toString('hex')
const s = String(k)
return /^[0-9a-fA-F]{64}$/.test(s) ? s.toLowerCase() : s
}
// Order-insensitive equality for the small hex->value maps projected into the view (__members__ /
// __devices__), so a re-projection only APPENDS when the view would actually change. Every append
// grows the log forever, and these run on a timer + on every swarm connection.
function sameMap (a, b) {
const ka = Object.keys(a || {}).sort(); const kb = Object.keys(b || {}).sort()
if (ka.length !== kb.length) return false
for (let i = 0; i < ka.length; i++) { if (ka[i] !== kb[i] || String(a[ka[i]]) !== String(b[kb[i]])) return false }
return true
}
function deriveGsKeyPair (roomSeed, ns = GS_NAMESPACE) { return crypto.keyPair(crypto.hash([ns, toBuf(roomSeed)])) }
function deriveBaseKey (gsPublicKey) { return Hypercore.key({ version: 1, signers: [{ publicKey: gsPublicKey }] }) }
// Router → instance handlers (need instance state for admission). ctx carries
// { tx, base, node, sb }. LWW-by-ts; D1b admission on optimistic writes.
function makeRouter () {
const r = new Router()
r.add('@state/set-entry', (rec, ctx) => ctx.sb._applySet(rec, ctx))
r.add('@state/del-entry', (rec, ctx) => ctx.sb._applyDel(rec, ctx))
return r
}
export class StateBaseV3 {
constructor (storagePath, roomKey, opts = {}) {
this._roomKey = Buffer.isBuffer(roomKey) ? roomKey : Buffer.from(String(roomKey), 'hex')
if (this._roomKey.length !== 32) throw new Error('roomKey must be 32 bytes (or 64-char hex)')
this._roomKeyHex = this._roomKey.toString('hex')
// Distinct v3 path — v2 'state-bases/' is left untouched (clean cutover). Null when a shared
// Corestore is injected (opts.store, multi-DID): storage is a namespace of the daemon's store,
// so there is no per-room dir.
this._storagePath = storagePath ? path.join(storagePath, 'state-bases-v3', this._roomKeyHex) : null
this._network = opts.network !== false
this._swarmOpts = opts.swarmOpts || null
// Designated always-on anchor: on open, claim the SOLE indexer slot via
// move-indexer so the room finalizes even when the creator is offline.
this._claimIndexer = !!opts.claimIndexer
// Indexer eligibility (issue #133 — indexing≠signing). The creator opens AS the bootstrap sole
// indexer; a regular 2nd device must stay a PLAIN WRITER (making every admin an indexer creates
// the N=2 quorum trap — majority 2, zero fault tolerance, can't self-heal). Only a DESIGNATED
// indexer self-promotes: an always-on anchor (opts.claimIndexer/opts.indexer). Default: off.
this._indexerEligible = !!opts.indexer
// P3 (#103): admins self-promote to indexers not just at open but whenever their
// keet role changes at runtime (member→admin), driven by a low-frequency retry
// timer. _promoteRetryMs = how often to re-check; _promoting guards overlapping
// runs (timer + projectMembers can race). Q7 (#103): _maxIndexers bounds the
// ACTUAL indexer set so a large admin room doesn't push the majority-quorum
// threshold so high that finalization gets fragile — admins past the cap stay
// plain writers (still admitted, still write; they just don't index).
this._promoteRetryMs = Number(opts.promoteRetryMs) || Number(process.env.KEET_PROMOTE_RETRY_MS) || 30000
this._maxIndexers = Number(opts.maxIndexers) || Number(process.env.KEET_MAX_INDEXERS) || 5
this._promoting = false
this._promoteTimer = null
this._promoteFutileLogged = false
// Membership context. selfKey (this daemon's keet identity pubkey) drives the
// deterministic creator election; it comes from router._stateBaseProviders as
// `selfKey` (a 32-byte pubkey), NOT a keypair. selfKeyPair is optional (kept
// for v3.0b signing). CRITICAL: without a real selfKey the election would
// default amCreator=true on every daemon → multiple genesis → split-brain.
this._selfKeyPair = opts.selfKeyPair || null
this._selfKey = opts.selfKey
? toBuf(opts.selfKey)
: (this._selfKeyPair ? toBuf(this._selfKeyPair.publicKey) : null)
// Leak-free single-signer election from live CONNECTION PRESENCE (replaces the __lease__ ts-heartbeat,
// which appended to the log every ~8s forever). self = our member key; graceMs = failover blip-tolerance
// (default matches the old lease TTL). Fed by attachSignerPresence() on each swarm connection; the sign
// loop reads holdsSigner(). Null when we have no member identity (read-only opens).
// onChange routes a presence change (co-owner connect/disconnect) through the SAME watcher fan-out
// as a view advance, so a single sb.watch(tick) drives ZERO-TIMER failover: the daemon re-evaluates
// holdsSigner + signs on writes, replication, AND presence events — no polling timer.
this._presence = this._selfKey ? new SignerPresence({ self: this._selfKey.toString('hex'), graceMs: Number(opts.signerGraceMs) || 15000, onChange: () => this._fireWatchers() }) : null
this._selfHead = null // our current signed head rev — announced over presence so a BEHIND co-owner is
this._presenceSenders = new Set() // excluded from the election (anti-freeze); re-announced on head change
this._getMembers = typeof opts.getMembers === 'function' ? opts.getMembers : null
// RESYNC heal (force-non-creator): open as a NON-creator even when we're the lowest-key member, so a
// fresh-corestore re-join gets a FRESH (primaryKey-derived) local writer instead of re-opening the
// deterministic _gsKeyPair genesis writer (which a fresh store cannot shed and autobase will not
// linearize when backfilled — see _open + hyper-proto-peer resync). Mirrors keet: a device that lost
// its store re-joins as a NEW random writer via the doorbell (keet-core pairing/keys.js), never the
// old writer. Set on a resync boot only (device-node.js passes forceNonCreator: !!resyncStore).
this._forceNonCreator = opts.forceNonCreator === true
// forceCreator (assume-genesis heal): the healthy co-owner re-assumes the deterministic genesis
// writer/indexer after the original creator abandoned it via a resync — see _open. Mutually exclusive
// with forceNonCreator; exactly one co-owner holds gsKeyPair at a time.
this._forceCreator = opts.forceCreator === true
// Injectable crypto for membership-gated admission (keet attestation primitives).
// attest: sign localKey‖base.key as our keet identity; verifyAttest: IdentityKey.verify.
this._attest = typeof opts.attest === 'function' ? opts.attest : null
this._verifyAttest = typeof opts.verifyAttest === 'function' ? opts.verifyAttest : null
// Admission model (A): prove control of THIS daemon's keet room writer key, which
// keet maps (via @keet/devices) to our active member. roomWriterKeyPair signs the
// proof; roomWriterManifest binds the signer to the writer/device key
// (deviceKey = Hypercore.key(manifest)). Both come from room.base.local in the
// router. The verifier re-derives the device key and looks it up in __devices__
// (deviceKey→memberKey) → __members__ (memberKey→role). See hyper-fleeky-dev #99.
this._roomWriterKeyPair = opts.roomWriterKeyPair || this._selfKeyPair || null
this._roomWriterManifest = opts.roomWriterManifest || null
// Deterministic base identity from the room's members-only secret (v2 design).
this._roomSeed = opts.roomSeed ? toBuf(opts.roomSeed) : null
// gsNamespace override: defaults to the current GS_NAMESPACE. The one-time re-adopt migration
// and tests pass LEGACY_GS_NAMESPACE to open a pre-rename room in place.
this._gsNamespace = opts.gsNamespace ? toBuf(opts.gsNamespace) : GS_NAMESPACE
this._gsKeyPair = this._roomSeed ? deriveGsKeyPair(this._roomSeed, this._gsNamespace) : null
this._detBaseKey = this._gsKeyPair ? deriveBaseKey(this._gsKeyPair.publicKey) : null
this._encryptionKey = opts.encryptionKey ? toBuf(opts.encryptionKey) : null
// Multi-DID (multi-tenant) support. An injected SHARED Corestore + shared ProtomuxWakeup let ONE
// daemon host N DIDs: each StateBase opens on a per-DID NAMESPACE of the one store (one dir, one
// RocksDB lock, cryptographically-isolated cores) and shares one wakeup — exactly how keet-core
// fronts N rooms over one corestore/swarm. When opts.store is given, pass network:false too: this
// StateBase then creates NO Corestore and NO swarm; the daemon owns store+swarm+wakeup and drives
// `wakeup.addStream(conn); store.replicate(conn)` per connection + swarm.join(sb.base.discoveryKey).
this._injectedStore = opts.store || null
this._wakeup = opts.wakeup || null
this._router = makeRouter()
this.store = null
this.base = null
this.swarm = null
this._ready = false
this._closing = false
this._watchers = new Set()
// Writer-hex keys we've already admitted via addWriter at apply time. A remote
// member attaches its `admit` proof to EVERY optimistic write until it observes
// itself become writable, so without this guard apply re-runs addWriter/ackWriter
// (and logs) once per op — hundreds of redundant calls per peer. addWriter is
// idempotent, so de-duping the CALL leaves the writer set identical.
this._admitted = new Set()
// Cached "self is an ADMITTED, durable writer of this room": our local writer
// key (base.local.key) is in the autobase linearized system writer set
// (base.system). This is the signal `writable` reports — it answers "will my
// durable writes persist?" correctly: true once an indexer has actually
// admitted us (at/before base.writable's localWriter activation, so it still
// covers the post-admission lag), and FALSE for a peer that is merely an
// eligible room member in __members__ but was never admitted by an indexer
// (the false-green that keying off __members__ produced — see
// hyper-fleeky-dev #99: a local identity in the roster but absent from the
// indexer's membership read writable=true while its writes never finalized).
this._selfIsAdmittedWriter = false
// __members__ projection is EVENT-DRIVEN + self-terminating (NOT a bounded give-up loop). The
// creator retains the roster (set by projectMembers) and re-projects on the promote-timer tick +
// every new swarm connection until __members__ is confirmed in its view. A creator that becomes
// writable AFTER boot (re-adopt / swarm flap) therefore still lands __members__ and can admit
// members — the wj62 fork root was the app's 12-try boot loop giving up before writability settled,
// so __members__ never landed → every admit verified verifiedRole=null → members held their own
// lines. See test/admission-members-projection.test.mjs.
this._roster = null
this._membersProjected = false
}
// ── Lifecycle ──────────────────────────────────────────────
async ready () {
if (this._ready) return
if (this._closing) throw new Error('StateBaseV3 is closing')
if (this._injectedStore) {
// Shared-store (multi-DID) path: a per-DID NAMESPACE of the daemon's one Corestore — no new
// dir, no second RocksDB lock (the root's lock covers every namespace; a 2nd Corestore on the
// same dir fails to lock, a namespace does not). Proven end-to-end in test/multi-did.test.js
// (2 DIDs, one lock, encrypted admission over one shared wakeup, isolated).
this.store = this._injectedStore.namespace(this._roomKeyHex)
} else {
fs.mkdirSync(this._storagePath, { recursive: true })
this.store = new Corestore(this._storagePath)
}
await this.store.ready()
try {
await this._open()
} catch (err) {
if (!this._isCorruptionError(err)) throw err
// Diagnostic mode: report corruption WITHOUT quarantining (non-destructive
// probe of live storage). Used by tools/probe-room.mjs.
if (process.env.KEET_PROBE_NO_QUARANTINE) {
console.error('[state-base-v3] room=%s CORRUPT (%s) — probe mode, NOT quarantining',
this._roomKeyHex.slice(0, 12), err.message)
throw err
}
console.error('[state-base-v3] room=%s CORRUPT (%s) — quarantining + restarting',
this._roomKeyHex.slice(0, 12), err.message)
await this._quarantineAndReset()
}
this._ready = true
// PROJECT A ROSTER RETAINED BEFORE ready(). projectMembers() called before the base exists can only
// stash the roster (`if (!this._ready || !this.base || !this.base.writable) return false`), so the
// actual __members__ append waited for whichever came first: a swarm connection, or a promote-timer
// tick up to promoteRetryMs later.
//
// Under network:false — the multi-DID daemon mode — the connection hook that would drive this lives
// inside `if (this.swarm)` and NEVER RUNS, leaving only the timer. So __members__ landed late, and a
// member whose optimistic doorbell arrived first was refused by _verifyAdmissionInApply and never
// re-rang it: admission silently never happened, while __members__ and _membersProjected both read
// perfectly healthy afterwards — which is why this looked like a replication or daemon-wiring bug.
//
// Measured (test/members-projection-order.test.mjs): projectMembers BEFORE ready() left the member
// unadmitted; AFTER admitted immediately. Same wiring, same DID count — ordering was the ONLY
// variable. This makes the two equivalent, which is what callers already assume.
//
// Awaited so the roster is in the view before ready() returns, hence before any consumer can wire up
// connections. A no-op when nothing was retained, when not writable, or when already projected.
await this._ensureMembersProjected().catch(() => {})
// NO ABI announce at boot. Writing during _open is the fork hazard this file warns about throughout
// (the resync/joiner rule is to write NOTHING until replication converges) — and it reproduced here
// immediately: `[hypercore] conflict detected ... (writable=false,quorum=1)` followed by
// SESSION_CLOSED inside ApplyState._boot. The stamp is only ever needed when another writer is
// present to read it, so it is announced on connection and on admission instead.
// One-time in-place re-adopt migration for the namespace rename: if this is the writable creator
// and the (renamed) base is empty but a legacy-namespace base is present in the same store, copy
// its durable entries across. Best-effort + awaited so the room is seeded before consumers read.
await this._migrateFromLegacyNamespace()
// Admins/mods AUTO-promote to indexers on open (keet model: admins ARE the
// indexers — members/index.js addWriter(indexer:isIndexing)). Best-effort +
// background so it never blocks boot; needs an online indexer to land and
// retries on the next open. Non-admins are untouched (stay plain writers),
// so a room's indexer set = its online admins (quorum tolerates the creator
// dropping once ≥3 admins index).
this._maybePromote().catch(() => {})
// P3 (#103): the open-time call above only sees our role AS OF open. An admin
// GRANTED at runtime (member→admin while the daemon stays up) would otherwise
// need a full reopen to self-promote — the ashatron footgun. Re-fire on a
// low-frequency timer so a newly-promoted admin becomes an indexer on its own.
// Cheap when idle: _maybePromote early-returns once we're a durable indexer or
// while we're not (yet) an admin. unref so it never keeps the process alive.
if (!this._promoteTimer) {
this._promoteTimer = setInterval(() => { this._maybePromote().catch(() => {}); this._ensureMembersProjected().catch(() => {}) }, this._promoteRetryMs)
if (this._promoteTimer && typeof this._promoteTimer.unref === 'function') this._promoteTimer.unref()
}
}
// A DESIGNATED always-on anchor (opts.claimIndexer) automatically claims the
// sole indexer slot on open, so the room finalizes (quorum 1) even when the
// creator is offline. NOT every admin — autobase can't hold many indexers, so
// only the one designated anchor takes the slot; everyone else stays a plain
// writer (admitted on first write). Best-effort + background; needs the current
// indexer online once to sign the handoff, then it's durable in the oplog.
async _maybePromote () {
if (this._promoting) return // guard the timer/projectMembers race — one attempt at a time
this._promoting = true
try {
if (!this.base) return
if (this.base.writable && this._isLocalIndexer() === true) return // already an indexer
// Non-designated devices NEVER self-promote (the N=2-trap fix, #133). The creator is already
// the bootstrap sole indexer; a regular 2nd device stays a plain writer (it can still hold the
// lease + sign — signing needs the #atproto key, NOT indexer status). Only an anchor
// (claimIndexer / opts.indexer) grows the indexer set. See reference_statebase-indexing-vs-did-signing.
if (!this._claimIndexer && !this._indexerEligible) {
if (!this._promoteNonDesignatedLogged) { this._promoteNonDesignatedLogged = true; console.log('[state-base-v3] room=%s NOT self-promoting: non-designated device stays a PLAIN WRITER (N=1 posture; set opts.indexer for an anchor)', this._roomKeyHex.slice(0, 12)) }
return
}
// keet model (members/index.js:137 `isIndexing = canAdmin`): EVERY admin is an
// indexer, so the set grows to = the online admins and majority quorum tolerates
// any single node (incl. the genesis creator) dropping. Verify we're an admin
// (role 1); non-admins stay plain writers. Best-effort + background; needs an
// online indexer to sign the addWriter (autobase invariant) — retries next open.
if (this._getMembers && this._selfKey) {
let members; try { members = await this._getMembers() } catch { members = null }
const selfHex = this._selfKey.toString('hex')
const me = Array.isArray(members) ? members.find(m => normKeyHex(m.key || m.devicePubkey || m.identity || m.memberId || m.id) === selfHex) : null
if (!me || Number(me.role) !== 1) return // non-admin → stays a plain writer
}
// Q7 (#103): bound the ACTUAL indexer set. Every admin is ELIGIBLE (keet model),
// but once the live set has _maxIndexers members we stop growing it — a room with
// many admins would otherwise push the majority threshold ((n>>>1)+1) so high that
// a couple of offline indexers stall finalization. Admins past the cap stay plain
// writers (admitted on their writes; they just don't index). We're not already an
// indexer here (early-returned above), so a full set means: leave it alone.
let idxLen = 0; try { idxLen = this.base.linearizer.indexers.length } catch { idxLen = 0 }
if (idxLen >= this._maxIndexers) {
console.log('[state-base-v3] room=%s admin NOT self-promoting: indexer set full (%d/%d) — staying a plain writer', this._roomKeyHex.slice(0, 12), idxLen, this._maxIndexers)
return
}
// AUTOBASE CONSTRAINT (7.28.1): a writer can only be elevated to indexer while it is
// FRESH — base.local.length === 0. Reliable promotion is a two-phase protocol anchored
// at length 0 (addWriter parks the key in pendingIndexers, the writer self-acks — gated
// on local.length===0, apply-state.js:356 — and the manifest fork fires anchored to that
// writer's own first confirmed node). A writer that has ALREADY appended data
// (_seenLength>0) takes system._updateIndexer's immediate branch, binding the fork to an
// un-anchored tip node authored by another indexer; the reorganising linearizer keeps
// rolling it back (_rollbackViews) → indexerUpdate is set but the indexer set never
// advances. So a data-bearing writer CANNOT be promoted in place — retrying is futile and
// just spams the log. keet has the same constraint: it mints indexers only from FRESH
// devices/joins (addDevice: addWriter(deviceKey,{indexer:isIndexing})); change-member-role
// sets the admin CAPABILITY only, never addWriter. So we promote ONLY while fresh; an
// admin that already wrote game data keeps its admin capability but stays a plain writer
// (durability is provided by the fresh/anchor indexers, not by every admin). See
// hyper-fleeky-dev #103 + memory/project_statebase-inplace-indexer-promotion-*.
let localLen = 0; try { localLen = this.base.local ? this.base.local.length : 0 } catch { localLen = 0 }
if (localLen > 0) {
if (!this._promoteFutileLogged) {
this._promoteFutileLogged = true
console.log('[state-base-v3] room=%s admin NOT self-promoting: writer already has data (local.length=%d) — autobase only elevates a FRESH (length 0) writer to indexer; keeping admin as a plain writer', this._roomKeyHex.slice(0, 12), localLen)
}
return
}
// Every admin GROWS the set via start-indexing (no demote → quorum). A DESIGNATED
// anchor (opts.claimIndexer) still does the SOLE-slot move-indexer for back-compat.
const mode = this._claimIndexer ? { claimIndexer: true } : { startIndexing: true }
console.log('[state-base-v3] room=%s admin self-promoting to indexer (%s)', this._roomKeyHex.slice(0, 12), this._claimIndexer ? 'move-indexer anchor' : 'start-indexing grow')
const r = await this.requestWriter({ timeoutMs: 25000, ...mode })
console.log('[state-base-v3] room=%s promote result writable=%s indexer=%s', this._roomKeyHex.slice(0, 12), r.writable, r.indexer)
} catch (e) { console.log('[state-base-v3] promote failed: %s', e.message) } finally { this._promoting = false }
}
async _open () {
if (!this._detBaseKey) throw new Error('StateBaseV3 requires roomSeed (deterministic key)')
// forceNonCreator (resync heal): never elect as creator → open with keyPair=null → a FRESH local
// writer; the deterministic genesis (_gsKeyPair) becomes a REMOTE writer that replicates + linearizes
// normally (the proven non-creator path), instead of an un-sheddable, un-linearizable local writer.
//
// forceCreator (notary revival): re-assume the deterministic genesis writer/indexer on a HEALTHY
// co-owner even when it is NOT the lowest key. Used when the original creator (lowest-key) abandoned
// the genesis via a resync (forceNonCreator) — leaving the base with NO live indexer (indexedLength
// frozen, new writes optimistic-only). Because _gsKeyPair is roomSeed-derived (derivable by any
// co-owner), a healthy co-owner that holds the authoritative state can become the sole live indexer
// (quorum 1) — breaking the "need-an-indexer-to-make-an-indexer" deadlock keet cannot escape (keet
// writers are random). Once ONE notary is live again, keet's dynamic start/move/stop-indexing take over.
//
// CANONICAL usage = REBUILD-FROM-AUTHORITATIVE: pair forceCreator with a FRESH store + adopt(records,
// head) — the notary AUTHORS the genesis from the snapshot (adopt continues the atproto chain from the
// current head, so the relay chain is unbroken). This sidesteps two walls the in-place variant hit:
// a fresh store has NO stale view core (no autobase view-reindex checkout crash), and AUTHORING (not
// backfilling) an empty genesis is the normal creation path (diagnostic-1 N/A). Proven in
// hyper-proto-peer/test/rebuild-from-authoritative.test.js (new post FINALIZES). Do NOT forceCreator an
// EXISTING store to "continue" the genesis in place — that reindexes the view and crashes on write.
// EXACTLY ONE co-owner may hold gsKeyPair at a time — the resynced peer MUST be forceNonCreator.
const amCreator = this._forceCreator ? true : (this._forceNonCreator ? false : ((!this._getMembers || !this._selfKey) ? true : await this._amCreator()))
// RECORD the decision. `_amCreator` is a METHOD, so every consumer doing `!!sb._amCreator` to ask
// "is this node the creator?" was reading a truthy function and getting `true` unconditionally — which
// is exactly what hyper-proto-peer's members-diag did, so that field has always said `true` for every
// node, creator or not. A diagnostic that cannot report the negative case is worse than none: it was
// cited as evidence of a role flip during a real incident. This is the answer to that question.
this._isCreator = amCreator
const roleTag = this._forceCreator ? ' (assume-genesis: forced creator)' : (this._forceNonCreator ? ' (resync: forced non-creator)' : '')
console.log('[state-base-v3] room=%s %s%s base=%s', this._roomKeyHex.slice(0, 12),
amCreator ? 'creator' : 'member', roleTag, this._detBaseKey.toString('hex').slice(0, 12))
await this._openBase(this._detBaseKey, amCreator ? this._gsKeyPair : null)
}
async _openBase (bootstrapKey, keyPair) {
const baseOpts = {
// The Autobase view IS a HyperDB. autoUpdate is load-bearing: HyperDB reads
// a snapshot frozen at construction, advanced only by core 'append'.
open: (store) => HyperDB.bee(store.get('view'), hyperdbDefinition, { autoUpdate: true }),
apply: this._apply.bind(this),
close: async (view) => { if (view) await view.close() },
valueEncoding: 'binary', // ops are hyperdispatch-encoded buffers
optimistic: true,
ackInterval: 1000,
ackThreshold: 4
}
if (keyPair) baseOpts.keyPair = keyPair
if (this._encryptionKey) { baseOpts.encryptionKey = this._encryptionKey; baseOpts.encrypted = true }
// Shared wakeup (multi-DID): all DIDs on the daemon share ONE ProtomuxWakeup, so a single
// `wakeup.addStream(conn)` carries every DID's optimistic/pending-writer announcements — the
// admission doorbell works over the shared connection with no per-base wakeup (autobase sets
// wakeupOwner=false when handlers.wakeup is injected). Pair with network:false (daemon-driven repl).
if (this._wakeup) baseOpts.wakeup = this._wakeup
this.base = new Autobase(this.store, bootstrapKey, baseOpts)
await this.base.ready()
// Fire the watcher fan-out on EVERY autobase view advance — crucially including the BACKGROUND
// apply-completion of THIS node's OWN optimistic write. Without this, a solo write (e.g. a direct
// apply-write on the notary) appends but the sb.watch -> tick -> sign() chain never re-fires: the
// apply-path _fireWatchers() only runs on an explicit base.update() drain with new nodes, and the
// autoUpdate view advance does not go through _apply. So the write would finalize only when some
// OTHER event (a co-owner's replication, another write, a presence change) happens to drive an apply
// — the "direct notary write never finalizes" lag. See test/finalization-lag.repro.test.js. sign() is
// idempotent (no-ops on an unchanged data root) and the tick coalesces, so extra fires are cheap.
this.base.on('update', () => this._fireWatchers())
if (this._network) {
if (!this.swarm) {
this.swarm = new Hyperswarm(this._swarmOpts || undefined)
this.swarm.on('connection', (socket) => {
// Observability: StateBase swarm peers were invisible in logs, so a device coming up
// ISOLATED (no base-topic peer) looked identical to connected-but-not-replicating. Log it.
const rk = socket.remotePublicKey ? socket.remotePublicKey.toString('hex').slice(0, 12) : '?'
console.log('[state-base-v3] swarm peer CONNECTED room=%s peer=%s conns=%s', this._roomKeyHex.slice(0, 8), rk, this.swarm.connections ? this.swarm.connections.size : '?')
socket.on('close', () => console.log('[state-base-v3] swarm peer CLOSED room=%s peer=%s', this._roomKeyHex.slice(0, 8), rk))
// MUST replicate via base.replicate (not store.replicate): it ALSO
// wires the autobase wakeupProtocol onto the stream, which announces
// optimistic/pending writer cores. With bare store.replicate the
// indexer never discovers a not-yet-admitted member's optimistic
// onboarding block → admission never runs → member stays non-writable
// (the testy1/AI-Testing-Town "can't write" deadlock). cf. pear-git-v3
// pear-git.js:447 ("corestore.replicate unreliable for writer cores").
try { this.base.replicate(socket) } catch (err) { console.error('[state-base-v3] replicate threw:', err.message) }
try { this.attachSignerPresence(socket) } catch {} // ephemeral signer-presence handshake — feeds the election, appends nothing
// EVENT-DRIVEN __members__ projection: a peer (a member) just connected. If we're the writable
// creator and haven't projected the roster yet, (re)project now so its admission verifies —
// instead of waiting on the slow promote-timer tick. Self-terminating + idempotent.
this._ensureMembersProjected().catch(() => {})
this._announceAbi().catch(() => {}) // a peer arrived: stamp our ABI if we have not yet
})
}
this.swarm.join(this.base.discoveryKey)
}
console.log('[state-base-v3] room=%s base.key=%s writable=%s', this._roomKeyHex.slice(0, 12),
this.base.key.toString('hex').slice(0, 12), this.base.writable)
}
// Attach the ephemeral signer-presence handshake to a connection (announce our member key, learn the
// peer's → feed the election). Public so the multi-DID daemon can also attach it on the shared swarm;
// the single-DID path calls it from the internal connection handler. Returns a detach fn; no-op without a
// member identity. Appends NOTHING — presence lives and dies with the connection.
// id: this._roomKey — a per-DID channel id so N DIDs' presence coexist on ONE shared multi-DID daemon
// connection (protomux keys on (protocol, id); without it every DID collides on a null id and only the
// first opens). Single-DID and multi-DID both derive the same roomKey per DID, so peers of a DID match.
attachSignerPresence (conn) {
if (!this._presence) return () => {}
// Diagnostic counts (presenceDiag): attaches that got the channel, attaches that were a silent duplicate
// no-op, and announces received per peer — "channel taken" vs "peer never sends", told apart by a read.
const d = this._presenceDiagState || (this._presenceDiagState = { attached: 0, duplicate: 0, received: 0, peers: {} })
const handle = attachPresence(conn, {
selfKey: this._selfKey.toString('hex'),
presence: this._presence,
id: this._roomKey,
getHead: () => this._selfHead,
onReceive: (k) => { d.received++; d.peers[String(k).slice(0, 16)] = Date.now() }
})
if (handle.attached) d.attached++
else d.duplicate++
this._presenceSenders.add(handle.send)
return () => { this._presenceSenders.delete(handle.send); try { handle() } catch {} } // handle() detaches
}
// Feed our current signed head rev into the presence election (currency) + RE-ANNOUNCE it to peers, so a
// co-owner BEHIND the network head is excluded from the election while it is gate-blocked from signing —
// the caught-up peer signs instead of the chain freezing. Called by the sign loop after sign/restore.
setSignerHead (rev) {
const r = rev || null
if (r === this._selfHead) return
this._selfHead = r
if (this._presence) this._presence.setSelfHead(r)
for (const send of this._presenceSenders) { try { send(r) } catch {} }
}
// Do WE hold the active-signer role right now? Drop-in for the old lease.holds() in the sign loop:
// election is a deterministic function of who is CONNECTED (lowest-key effective co-owner), computed in
// RAM. A device with no co-owners connected is the sole signer (holds=true). No writes, ever.
holdsSigner (now, netHead = null) { return this._presence ? this._presence.holds(now, netHead) : true }
currentSigner (now, netHead = null) { return this._presence ? this._presence.currentSigner(now, netHead) : (this._selfKey ? this._selfKey.toString('hex') : null) }
signerState (now, netHead = null) { return this._presence ? this._presence.state(now, netHead) : { self: null, signer: null, effective: [], holds: true } }
// Presence wiring diagnostics (see attachSignerPresence). Read-only; null without a presence election.
presenceDiag () {
if (!this._presence) return null
const d = this._presenceDiagState || { attached: 0, duplicate: 0, received: 0, peers: {} }
return { attached: d.attached, duplicate: d.duplicate, received: d.received, peers: { ...d.peers } }
}
// keet's _autobaseApply commit shape: a per-batch transaction; handlers write
// into it; flush on success / close on error (never insert on the db directly).
async _apply (nodes, view, base) {
if (process.env.SBV3_DIAG) {
const who = base.writable ? 'WRITABLE' : 'member'
const froms = nodes.map(n => (n.from && n.from.key ? n.from.key.toString('hex').slice(0, 8) : 'sys')).join(',')
console.log('[SBV3_DIAG] _apply self=%s(%s) nodes=%d from=[%s]',
(this._selfKey ? this._selfKey.toString('hex').slice(0, 8) : '?'), who, nodes.length, froms)
}
const tx = view.transaction()
try {
for (const node of nodes) {
const op = node.value
if (op == null || !Buffer.isBuffer(op)) continue // null/system nodes
await this._router.dispatch(op, { tx, base, node, sb: this })
}
await tx.flush()
} catch (err) {
try { await tx.close() } catch {}
throw err
}
this._fireWatchers()
}
// Notify every sb.watch(fn) subscriber. Fired on a view advance (above) AND on a presence change
// (SignerPresence.onChange), so one subscription covers writes/replication + signer failover.
_fireWatchers () { for (const w of this._watchers) { try { w() } catch {} } }
// ── Public API (generic KV — unchanged surface vs v2) ──────
async set (key, value) {
if (!this._ready) throw new Error('StateBaseV3 not ready')
// DOORBELL (keet block-until-admitted). NEVER fuse durable CONTENT into a pre-admission
// OPTIMISTIC block: autobase keeps an optimistic block ONLY if the SAME apply admits+acks its
// writer, else it is rolled back every drain and discarded (apply-state.js:1110-1141). Fusing
// content+admit therefore reports success while the write silently VANISHES when no indexer is
// online to ack it — and the author can never read it back (test/silent-loss.repro.mjs). Every
// durable autobase app (autopass/keet/hyper-rooms/hypernote) instead admits FIRST, writes AFTER.
// So: if we are not yet an admitted writer, drive a CONTENT-FREE admission and block until we
// are genuinely in the system writer set; only then append content non-optimistically. If
// admission cannot be reached, _ensureWritable THROWS (never a silent loss). cf. skills/autobase.
await this._ensureWritable()
const rec = { key: String(key), value: this._enc(value), ts: Date.now() }
await this.base.append(encode('@state/set-entry', rec))
// BOUNDED: the content is appended (durable-local + replicated); a stalled drain must not hang
// the writer indefinitely (Phase-2 #6). The value is still readable back via the autoUpdate view.
await this._updateBounded()
}
async del (key) {
if (!this._ready) throw new Error('StateBaseV3 not ready')
await this._ensureWritable()
await this.base.append(encode('@state/del-entry', { key: String(key), ts: Date.now() }))
await this._updateBounded()
}
// Block until we are a genuinely ADMITTED writer (our local writer key is in the linearized
// system writer set), then return. No-op if already writable. Otherwise ring the CONTENT-FREE
// admission doorbell (requestWriter, which re-rings until admitted) and wait; if no indexer
// admits us within the timeout, THROW rather than let a caller write content that autobase would
// silently roll back. This is the guard that makes set()/del() durable (block-until-admitted).
async _ensureWritable ({ timeoutMs = 20000 } = {}) {
if (this.base && this.base.writable) return
await this.requestWriter({ timeoutMs })
if (!this.base || !this.base.writable) {
throw new Error('StateBaseV3: not admitted as a writer (no indexer online within ' + timeoutMs + 'ms) — refusing to write content that would be silently lost')
}
}
// Trigger our OWN admission without writing any game data: append a keyless
// optimistic block carrying just the admission proof. When an online indexer
// applies it, _applySet addWriters us (INDEXER if our keet role is admin/mod,
// role 1/2). So an always-on `--pin-room` anchor that is an admin becomes a
// durable indexer and can then admit other members even when the original
// room creator is offline (fixes the "creator went away → room stuck" class).
// No-op if already writable. Requires an online indexer to land the first
// promotion (bootstrap); thereafter the writer status is durable in the oplog.
// claimIndexer: a DESIGNATED admin anchor takes the sole indexer slot (move-
// indexer). Without it, just become a plain writer. Goal reached when writable
// (and, if claiming, when we are the indexer).
async requestWriter ({ timeoutMs = 15000, claimIndexer = false, startIndexing = false } = {}) {
if (!this._ready) throw new Error('StateBaseV3 not ready')
const wantIndexer = claimIndexer || startIndexing
const done = () => this.base.writable && (!wantIndexer || this._isLocalIndexer() === true)
if (done()) return { writable: true, already: true, indexer: this._isLocalIndexer() }
// FAIL FAST if this device has been BANNED. Its identity is in the durable, replicated
// __removed__ set, so the admission guard (_verifyAdmissionInApply) will reject every admit
// proof it ever sends — ringing the doorbell for the full timeoutMs is pointless and makes every
// write a removed device attempts stall ~timeoutMs before throwing. Throw immediately instead.
// Best-effort (see _isSelfRemoved): false on any uncertainty, so a non-banned device is never
// wrongly rejected — it just falls back to the normal doorbell.
if (await this._isSelfRemoved()) throw new Error('StateBaseV3: this device has been removed (banned) — cannot become a writer')
// DO NOT WRITE INTO A ROOM THAT DOES NOT LIST US. This is keet's rule (RoomFirewall.allow resolves
// a peer through @keet/devices-by-swarm-key and admits only an ACTIVE device; membership is written
// by an existing member via addDevice). An unlisted device asking to be let in is the 26,953-block
// loop on wj62: apply resolves its role to null every time, so every append is futile, and the
// doorbell re-rings on each base advance — including the advance its own append caused.
//
// POSITIVE EVIDENCE OF EXCLUSION ONLY. `_isSelfListed` returns null when the room has no roster in
// our view at all — which is a fresh joiner mid-replication as much as it is an outsider, and those
// two are indistinguishable from here. Refusing on absence broke a genuine member whose __members__
// had not landed yet (test/doorbell-durable-write.test.js: its set() threw instead of waiting), so
// absence must mean "no information" and ring as before. Only a roster that EXISTS and omits us is
// a refusal — which is the flooding node's state exactly: long converged, listed nobody's business,
// 26,953 blocks of asking. A device admitted later rings on the very next advance, because
// __members__ changing IS an advance — event-driven, no timer.
if ((await this._isSelfListed()) === false) throw new Error('StateBaseV3: this room lists members and this device is not among them — wait for a member to admit it (addMember)')
const admit = await this._buildAdmissionProof()
if (!admit) throw new Error('StateBaseV3 cannot request writer: no membership/identity to build an admission proof')
// key MUST be a string (encoder always writes it; undefined crashes compact-
// encoding). MOVE_INDEXER = indexer swap (deliberate hand-off); START_INDEXING =
// grow the indexer set (self-promote, quorum path); '' = plain-writer admission.
// None insert a game entry.
const key = claimIndexer ? MOVE_INDEXER : (startIndexing ? START_INDEXING : '')
await this.base.append(encode('@state/set-entry', { key, admit, ts: Date.now() }), { optimistic: true })
// BOUNDED, not blocking: an un-converged replica's base.update() can hang forever (Phase-2 #6).
// The optimistic doorbell block is already appended (durable-local + replicated via the swarm);
// bounding the drive lets the loop keep re-ringing + re-polling base.writable instead of freezing.
await this._updateBounded()
const deadline = Date.now() + timeoutMs
while (!done() && Date.now() < deadline) {
await new Promise(r => setTimeout(r, 250))
// RE-RING the doorbell while still un-admitted: a single optimistic block can be rolled back
// before an intermittent / just-connected indexer acks it (silent-loss RECOVER mode), so keep
// a live optimistic block present to re-anchor the tip until admission lands. Once writable,
// stop — further appends go through the normal (non-optimistic) writer path.
if (!this.base.writable) {
try { await this.base.append(encode('@state/set-entry', { key, admit, ts: Date.now() }), { optimistic: true }) } catch {}
}
await this._updateBounded()
}
if (this.base.writable) this._announceAbi().catch(() => {}) // just admitted: we can stamp now
return { writable: this.base.writable, indexer: this._isLocalIndexer() }
}
// Is THIS daemon's local writer an indexer? autobase tracks it directly.
_isLocalIndexer () {
try { return this.base.isIndexer === true } catch { return null }
}
// Drive the linearizer, but NEVER hang the caller. base.update() awaits a full drain, which can
// stall INDEFINITELY on an un-converged plain-writer replica (device-swarm still catching up) —
// the Phase-2 hang (hyper-social #6): the sign/admission path awaited it and froze, so the
// re-admission doorbell never re-rang. This bounds how long we AWAIT one drain. It is SAFE to
// time out: convergence does NOT depend on our awaiting update() — incoming replication triggers
// background drains (autobase _onwakeup/_onremotewriterchange → _queueBump) that still advance
// base.writable — so a bounded await lets the doorbell loop keep re-ringing + re-polling the
// (sync) writable getter while the real work proceeds in the background. Same posture as the
// getNow/listNow snapshot reads and the boot __members__ time-box.
async _updateBounded (ms = 800) {
if (!this.base) return
let t = null
try {
await Promise.race([
this.base.update().catch(() => {}),
new Promise((r) => { t = setTimeout(r, ms); if (t && typeof t.unref === 'function') t.unref() })
])
} finally { if (t) clearTimeout(t) }
}
// Background, guarded re-admission doorbell for a booted-but-not-yet-writable plain writer.
// Rationale (Phase 2): a plain writer admitted at RUNTIME may reboot writable=false if an indexer
// never SIGNED a checkpoint covering its admission that the writer persisted (the power-cut
// window). Re-admission needs the writer to re-ring requestWriter — but the lease/sign path only
// writes when it can hold the lease, and a healthy peer (stortron) holds it, so that path never
// fires the doorbell. This is the independent trigger: the sign loop calls it each tick; it rings
// requestWriter ONCE at a time (a call while one is in flight is a no-op), so an online indexer
// re-admits us and signs a durable checkpoint we then persist. No-op once writable.
async ensureAdmitted ({ timeoutMs = 15000 } = {}) {
if (!this._ready || !this.base || this.base.writable) return this.base ? this.base.writable : false
if (this._ensuringAdmit) return false
this._ensuringAdmit = true
try { await this.requestWriter({ timeoutMs }) } catch {} finally { this._ensuringAdmit = false }
return this.base ? this.base.writable : false
}
async get (key) {
if (!this._ready) return null
await this.base.update()
const rec = await this.base.view.get(ENTRIES, { key: String(key) })
return rec ? this._dec(rec.value) : null
}
async list (prefix = '') {
if (!this._ready) return {}
await this.base.update()
const out = {}
const stream = this.base.view.find(ENTRIES, { gte: { key: prefix }, lt: { key: prefix + '~' } })
for await (const rec of stream) out[rec.key] = this._dec(rec.value)
return out
}
// Non-blocking reads — same as get()/list() but WITHOUT `await base.update()`. base.update() drives the
// linearizer fully up-to-date, which can HANG on an un-converged plain-writer replica (device-swarm
// still catching up), stalling any caller (e.g. the device-registry revocation check on login). The
// view is a HyperDB opened `autoUpdate:true`, so it already reflects the latest MATERIALIZED state —
// these read that current snapshot instantly (at most a few ms stale). Use for liveness-critical reads
// that must never block; use get()/list() when you need the freshest fully-drained view.
async getNow (key) {
if (!this._ready) return null
const rec = await this.base.view.get(ENTRIES, { key: String(key) })
return rec ? this._dec(rec.value) : null
}
async listNow (prefix = '') {
if (!this._ready) return {}
const out = {}
const stream = this.base.view.find(ENTRIES, { gte: { key: prefix }, lt: { key: prefix + '~' } })
for await (const rec of stream) out[rec.key] = this._dec(rec.value)
return out
}
// value is an arbitrary JSON-able Variant from the game; stored as a JSON string.
_enc (v) { try { return JSON.stringify(v) } catch { return String(v) } }
_dec (s) { if (typeof s !== 'string') return s; try { return JSON.parse(s) } catch { return s } }
// ── Apply-side op handlers (called by the router) ──────────
async _applySet (rec, ctx) {
// D1b admission: an optimistic write from a not-yet-admitted member carries
// an `admit` proof. Verify it deterministically (pure crypto + projected
// __members__, NO external lookup) and ackWriter so the block commits.
// Unverified optimistic blocks are NOT acked → autobase never applies them.
if (rec.admit && ctx.node && ctx.node.from && ctx.node.from.key) {
const fromKeyHex = ctx.node.from.key.toString('hex')
// Already admitted this writer — addWriter is idempotent, so skip the
// re-verify + re-addWriter + re-ack (and the log line) that the peer's
// every-op `admit` proof would otherwise trigger. Once they're a permanent
// writer their blocks commit via normal acking; no per-block ackWriter.
const fromKey = ctx.node.from.key
const fromHex = fromKeyHex.slice(0, 12)
// Record writer->identity in __devices__ for ANY admit-carrying op (plain admission,
// START_INDEXING, MOVE_INDEXER all addWriter this key) so a ban (REMOVE_DEVICE) can
// removeWriter this device by its identity (atproto-base-REVOCATION §6). Harmless if the op is
// later denied — the key never becomes a writer so removeWriter no-ops; admit.identity is the
// caller's OWN attested id (it cannot claim another's).
if (rec.key !== REMOVE_DEVICE && rec.key !== ADD_MEMBER && rec.key !== SET_EPOCH) {
try {
const a = typeof rec.admit === 'string' ? JSON.parse(rec.admit) : rec.admit
const idHex = a && a.identity ? a.identity : null
if (idHex) {
const drec = await ctx.tx.get(ENTRIES, { key: '__devices__' })
const devmap = drec && drec.value ? JSON.parse(drec.value) : {}
if (devmap[fromKeyHex] !== idHex) { devmap[fromKeyHex] = idHex; await ctx.tx.insert(ENTRIES, { key: '__devices__', value: JSON.stringify(devmap), ts: rec.ts || 0 }) }
}
} catch (e) { console.log('[state-base-v3] record device-writer FAIL: %s', e.message) }
}
if (rec.key === MOVE_INDEXER) {
// MOVE-INDEXER (keet members/index.js:258-275): a DESIGNATED admin anchor
// takes the SOLE indexer slot via an atomic swap — promote the anchor +
// demote the genesis creator (whose writer key is the deterministic
// base.key). Count stays 1, so the always-on anchor finalizes (quorum 1)
// even with the creator offline. We do NOT grow the indexer set: autobase
// can't reliably exceed ~2 indexers (the 3rd-indexer manifest hang, proven).
// Re-runnable (idempotent addWriter); not gated by _admitted (it's a
// promotion, not a first admission). Only admins (role 1) may claim.
const role = await this._verifyAdmissionInApply(fromKey, rec.admit, ctx.tx)
if (role !== 1) { console.log('[state-base-v3] MOVE-INDEXER DENIED from=%s role=%s', fromHex, String(role)); return }
try {
await ctx.base.addWriter(fromKey, { indexer: true })
this._admitted.add(fromKeyHex)
if (this._detBaseKey && Buffer.compare(fromKey, this._detBaseKey) !== 0) {
await ctx.base.addWriter(this._detBaseKey, { indexer: false }) // demote genesis creator
}
// NO preferFastForward: a fast-forwarding indexer refuses to ack/sign (autobase
// index.js:952), which deadlocks the indexer-set migration so signedLength stalls on
// ALL peers (audit 2026-07-02). keet's indexer ops don't call it. Removing it lets the
// new indexer ack and quorum reform. Proven on START_INDEXING (probe-indexer-quorum 8/8
// + probe-rotation-phantom-check 3/3 inline); MOVE-INDEXER (anchor path) same rationale.
console.log('[state-base-v3] MOVE-INDEXER: anchor=%s is now SOLE indexer (genesis demoted)', fromHex)
} catch (e) { console.log('[state-base-v3] move-indexer FAIL from=%s: %s', fromHex, e.message) }
try { await ctx.base.ackWriter(fromKey) } catch (e) { console.log('[state-base-v3] ackWriter FAIL: %s', e.message) }
} else if (rec.key === START_INDEXING) {
// START-INDEXING (keet members/index.js op D): an admin GROWS the indexer set
// by adding ITSELF as an indexer — NO demote of anyone. With >=3 admin-indexers,
// autobase majority quorum ((n>>>1)+1) finalizes even when the genesis/creator
// is offline. Only admins (role 1) may promote; idempotent addWriter (not gated
// by _admitted — it's a promotion, and combines first-admission + indexer grow).
const role = await this._verifyAdmissionInApply(fromKey, rec.admit, ctx.tx)
if (role !== 1) { console.log('[state-base-v3] START-INDEXING DENIED from=%s role=%s', fromHex, String(role)); return }
try {
await ctx.base.addWriter(fromKey, { indexer: true })
this._admitted.add(fromKeyHex)
// NO preferFastForward — see MOVE-INDEXER note above (FF-deadlock, autobase index.js:952).
console.log('[state-base-v3] START-INDEXING: %s is now an indexer (grow, no demote)', fromHex)
} catch (e) { console.log('[state-base-v3] start-indexing FAIL from=%s: %s', fromHex, e.message) }
try { await ctx.base.ackWriter(fromKey) } catch (e) { console.log('[state-base-v3] ackWriter FAIL: %s', e.message) }
} else if (rec.key === REMOVE_DEVICE) {
// BAN (atproto-base-REVOCATION §6): only a role-1 member may issue (authorized via the admit
// proof, like START_INDEXING). Record the target in REMOVED_SET, drop it from __members__,
// and removeWriter its writer(s) — so it can neither write now nor re-admit later.
const role = await this._verifyAdmissionInApply(fromKey, rec.admit, ctx.tx)
if (role !== 1) { console.log('[state-base-v3] REMOVE-DEVICE DENIED from=%s role=%s', fromHex, String(role)); return }
const target = String(this._dec(rec.value) || '')
if (target) {
const rrec = await ctx.tx.get(ENTRIES, { key: REMOVED_SET }); const removed = rrec && rrec.value ? JSON.parse(rrec.value) : {}
removed[target] = { ts: rec.ts || 0 }; await ctx.tx.insert(ENTRIES, { key: REMOVED_SET, value: JSON.stringify(removed), ts: rec.ts || 0 })
const mrec = await ctx.tx.get(ENTRIES, { key: '__members__' }); const members = mrec && mrec.value ? JSON.parse(mrec.value) : {}
if (members[target] != null || members[normKeyHex(target)] != null) { delete members[target]; delete members[normKeyHex(target)]; await ctx.tx.insert(ENTRIES, { key: '__members__', value: JSON.stringify(members), ts: rec.ts || 0 }) }
const drec = await ctx.tx.get(ENTRIES, { key: '__devices__' }); const devmap = drec && drec.value ? JSON.parse(drec.value) : {}
for (const [wk, id] of Object.entries(devmap)) {
if (id === target || id === normKeyHex(target)) {
try { await ctx.base.removeWriter(Buffer.from(wk, 'hex'), { indexer: false }); this._admitted.delete(wk); console.log('[state-base-v3] REMOVE-DEVICE: removeWriter %s (identity %s)', wk.slice(0, 12), target.slice(0, 12)) } catch (e) { console.log('[state-base-v3] removeWriter FAIL %s: %s', wk.slice(0, 12), e.message) }
}
}
}
} else if (rec.key === ADD_MEMBER) {
// ADMIT A MEMBER (keet contract/members addDevice): authorized by the AUTHOR's role, exactly
// like REMOVE_DEVICE above. The room records who belongs; the joiner never writes to ask.
const role = await this._verifyAdmissionInApply(fromKey, rec.admit, ctx.tx)
if (role !== 1) { console.log('[state-base-v3] ADD-MEMBER DENIED from=%s role=%s', fromHex, String(role)); return }
let target = null; let targetRole = 1
try { const v = this._dec(rec.value); if (v && typeof v === 'object') { target = String(v.id || ''); targetRole = Number(v.role) || 1 } else if (v) { target = String(v) } } catch {}
if (!target) { console.log('[state-base-v3] ADD-MEMBER: no target'); return }
// A ban outranks an add — otherwise any member could undo a removal by re-adding.
if (await this._isRemoved(ctx.tx, target)) { console.log('[state-base-v3] ADD-MEMBER REFUSED: %s is in __removed__ (unban first)', target.slice(0, 12)); return }
const mrec = await ctx.tx.get(ENTRIES, { key: '__members__' })
const members = mrec && mrec.value ? JSON.parse(mrec.value) : {}
if (members[target] === targetRole || members[normKeyHex(target)] === targetRole) return // idempotent
members[normKeyHex(target)] = targetRole
await ctx.tx.insert(ENTRIES, { key: '__members__', value: JSON.stringify(members), ts: rec.ts || 0 })
console.log('[state-base-v3] ADD-MEMBER: %s admitted role=%s by %s', target.slice(0, 12), String(targetRole), fromHex)
} else if (rec.key === SET_EPOCH) {
// Set the minimum attestation epoch (atproto-base-REVOCATION §3): only a role-1 member. After
// this, admission rejects any attestation whose epoch is below the receipt — re-attest
// survivors at the new epoch to keep them, invalidating all stale/revoked proofs at once.
const role = await this._verifyAdmissionInApply(fromKey, rec.admit, ctx.tx)
if (role !== 1) { console.log('[state-base-v3] SET-EPOCH DENIED from=%s role=%s', fromHex, String(role)); return }
const receiptHex = String(this._dec(rec.value) || '')
if (receiptHex) { await ctx.tx.insert(ENTRIES, { key: RECEIPT_KEY, value: this._enc(receiptHex), ts: rec.ts || 0 }); console.log('[state-base-v3] SET-EPOCH: min-epoch receipt set by %s', fromHex) }
} else if (!this._admitted.has(fromKeyHex)) {
const role = await this._verifyAdmissionInApply(fromKey, rec.admit, ctx.tx)
console.log('[state-base-v3] ADMIT attempt from=%s verifiedRole=%s', fromHex, String(role))
if (role == null) return
// Admit as a PLAIN WRITER (indexer:false). We deliberately never grow the
// indexer set; the sole indexer is the genesis creator OR a move-indexer
// anchor. Everyone else writes and the indexer signs.
try { await ctx.base.addWriter(fromKey, { indexer: false }); this._admitted.add(fromKeyHex); console.log('[state-base-v3] addWriter OK from=%s indexer=false', fromHex) } catch (e) { console.log('[state-base-v3] addWriter FAIL from=%s: %s', fromHex, e.message) }
try { await ctx.base.ackWriter(fromKey) } catch (e) { console.log('[state-base-v3] ackWriter FAIL: %s', e.message) }
}
}
// Empty/null key = a pure admission request; MOVE_INDEXER / START_INDEXING / REMOVE_DEVICE =
// reserved ops. All ran their effect above; none write a game entry.
if (!rec.key || rec.key === MOVE_INDEXER || rec.key === START_INDEXING || rec.key === REMOVE_DEVICE || rec.key === ADD_MEMBER || rec.key === SET_EPOCH) return
const existing = await ctx.tx.get(ENTRIES, { key: rec.key })
if (existing && typeof existing.ts === 'number' && existing.ts > (rec.ts || 0)) return
await ctx.tx.insert(ENTRIES, { key: rec.key, value: rec.value, ts: rec.ts || 0 })
}
async _applyDel (rec, ctx) {
const existing = await ctx.tx.get(ENTRIES, { key: rec.key })
if (existing && typeof existing.ts === 'number' && existing.ts > (rec.ts || 0)) return
await ctx.tx.delete(ENTRIES, { key: rec.key })
}
// Ban a device by its keet identity hex (atproto-base-REVOCATION §6): append a REMOVE_DEVICE op
// carrying an admission proof so apply authorizes by the CALLER's role (like requestWriter). Apply
// records the target in REMOVED_SET, drops it from __members__, and removeWriters its writer(s).
async banDevice (identityHex) {
if (!this._ready) throw new Error('StateBaseV3 not ready')
const admit = await this._buildAdmissionProof()
if (!admit) throw new Error('banDevice: no identity to authorize the ban')
await this.base.append(encode('@state/set-entry', { key: REMOVE_DEVICE, value: this._enc(String(identityHex)), admit, ts: Date.now() }), { optimistic: true })
await this.base.update()
}
// Admit a member into the room (keet contract/members addDevice). An EXISTING role-1 member authors
// it; apply authorizes by THIS caller's role. That is the whole point: membership becomes a fact of
// the ROOM, replicated to every device, instead of a roster in one machine's config file that only
// the creator could project (hyper-social #40 §6).
async addMember (identityHex, role = 1, { force = false } = {}) {
if (!this._ready) throw new Error('StateBaseV3 not ready')
// ABI GATE. Emitting this op into a room containing a writer that predates it makes the two apply
// it differently — the old node inserts `__add_member__` as an entry, we insert nothing — and a
// divergent apply is a forked view, which is far worse than the problem this op solves. Refuse and
// name the laggards. `force` exists for an operator who has verified the room out of band; it is
// never the default, and the message says who is missing so the check can be answered rather than
// bypassed blindly.
if (!force) {
const below = await this._writersBelowAbi(ROOM_ABI)
if (below.length) throw new Error('addMember: room ABI ' + ROOM_ABI + ' not supported by writer(s) ' + below.join(', ') + ' — upgrade them first (or pass force after verifying)')
}
const admit = await this._buildAdmissionProof()
if (!admit) throw new Error('addMember: no identity to authorize the admission')
await this.base.append(encode('@state/set-entry', { key: ADD_MEMBER, value: this._enc({ id: String(identityHex), role: Number(role) || 1 }), admit, ts: Date.now() }), { optimistic: true })
await this._updateBounded()
}
// Set the minimum attestation epoch (atproto-base-REVOCATION §3). receiptHex = a keet receipt
// encoding the new minimum epoch (membership.encodeReceipt). Authorized by the caller's admit
// proof (role 1). After this, attestations below the epoch are rejected on admission.
async bumpEpoch (receiptHex) {
if (!this._ready) throw new Error('StateBaseV3 not ready')
const admit = await this._buildAdmissionProof()
if (!admit) throw new Error('bumpEpoch: no identity to authorize')
await this.base.append(encode('@state/set-entry', { key: SET_EPOCH, value: this._enc(String(receiptHex)), admit, ts: Date.now() }), { optimistic: true })
await this.base.update()
}
// Ban guard: is this identity/member in the durable REMOVED_SET? (atproto-base-REVOCATION §3.)
async _isRemoved (tx, ...keys) {
const rrec = await tx.get(ENTRIES, { key: REMOVED_SET })
const removed = rrec && rrec.value ? JSON.parse(rrec.value) : {}
for (const k of keys) { if (k && (removed[k] || removed[normKeyHex(k)])) return true }
return false
}
// Out-of-apply self-ban check: is THIS device's own identity in the durable __removed__ set (a ban
// that has replicated to us)? Read against the linearized view exactly like getProjectedRole. Used
// by requestWriter to fail fast instead of ringing a doorbell the admission guard will never
// answer. Best-effort: returns false on any uncertainty (no selfKey / view not ready / parse
// error), so it can never wrongly reject a non-banned device — worst case falls back to the
// normal doorbell timeout.
async _isSelfRemoved () {
try {
if (!this._ready || !this.base || !this._selfKey) return false
// BOUNDED (Phase-2 #6): this runs at the TOP of requestWriter, before the doorbell append —
// a blocking base.update() here would hang the doorbell on an un-converged replica and it
// would never re-ring. Best-effort anyway (false on uncertainty), so reading a slightly stale
// ban snapshot is acceptable; the autoUpdate view is still current within a few ms.
await this._updateBounded()
const rrec = await this.base.view.get(ENTRIES, { key: REMOVED_SET })
const removed = rrec && rrec.value ? JSON.parse(rrec.value) : {}
const hex = this._selfKey.toString('hex')
return !!(removed[hex] || removed[normKeyHex(hex)])
} catch { return false }
}
// Does the room LIST this device — i.e. could an admission actually succeed? Read against the
// linearized view exactly like _isSelfRemoved. Two ways to be listed, matching what
// _verifyAdmissionInApply consults: our keet identity in __members__, or one of our writer keys
// mapped to a listed member in __devices__ (a device admitted earlier, now rebooting unwritable —
// the Phase-2 case, which MUST keep ringing).
//
// TRI-STATE: true = listed, false = the roster exists and omits us (refuse), null = we cannot tell
// yet (ring as before). Collapsing null into false is what broke a genuine member whose roster had
// not replicated — its set() threw instead of waiting. Only positive evidence of exclusion silences
// a device; that is precisely the flooding node's state and not a joiner's.
async _isSelfListed () {
try {
if (!this._ready || !this.base) return null // unknown — not "no"
await this._updateBounded()
const mrec = await this.base.view.get(ENTRIES, { key: '__members__' })
// NULL, not false: no roster in OUR view is a fresh joiner mid-replication as often as it is an
// outsider, and the caller must not treat the two alike. Only an existing roster that omits us
// is a refusal.
if (!mrec || !mrec.value) return null
const members = JSON.parse(mrec.value)
if (this._selfKey) {
const hex = this._selfKey.toString('hex')
if (members[hex] != null || members[normKeyHex(hex)] != null) return true
}
const drec = await this.base.view.get(ENTRIES, { key: '__devices__' })
const devmap = drec && drec.value ? JSON.parse(drec.value) : {}
const localHex = this.base.local && this.base.local.key ? this.base.local.key.toString('hex') : null
if (localHex) {
const memberKey = devmap[localHex] != null ? devmap[localHex] : devmap[normKeyHex(localHex)]
if (memberKey && (members[memberKey] != null || members[normKeyHex(memberKey)] != null)) return true
}
return false // the roster EXISTS and does not name us — the one case that is a real refusal
} catch { return null }
}
// Announce THIS writer's room ABI. Written once per writer key, on the events that can make it true
// (open-while-writable, and admission succeeding) — never on a timer. Idempotent: a re-announce of an
// unchanged value is skipped, so this cannot become an append loop of its own.
async _announceAbi () {
try {
if (!this._ready || !this.base || !this.base.writable) return false
const localHex = this.base.local && this.base.local.key ? this.base.local.key.toString('hex') : null
if (!localHex) return false
const key = ABI_PREFIX + localHex
const cur = await this.getNow(key).catch(() => null)
if (cur && Number(cur.v) === ROOM_ABI) return false
await this.set(key, { v: ROOM_ABI, ts: Date.now() })
return true
} catch { return false }
}
// Which active writers have NOT announced support for `abi`? Returns an array of short keys; empty
// means every writer in the room can apply the op identically. Fails CLOSED: a writer we cannot read
// a stamp for counts as unsupported, because "no stamp" is exactly what a pre-ABI node looks like.
async _writersBelowAbi (abi = ROOM_ABI) {
const below = []
try {
await this._updateBounded()
const writers = (this.base && this.base.activeWriters) ? [...this.base.activeWriters] : []
for (const w of writers) {
if (!w || w.isRemoved) continue // a removed writer applies nothing that matters
const key = w.core && w.core.key ? w.core.key.toString('hex') : null
if (!key) continue
const stamp = await this.getNow(ABI_PREFIX + key).catch(() => null)
if (!stamp || Number(stamp.v) < abi) below.push(key.slice(0, 12))
}
} catch { /* fail closed via whatever was collected */ }
return below
}
// ── One-time in-place re-adopt migration (namespace rename) ─────────────────
// The 'hyper-tps/state/v3' -> 'atproto-room/state/v3' rename changes the derived base key, so a
// room created under the old namespace opens here as a FRESH, empty base that shares the same
// on-disk Corestore as its old cores. Seed the renamed base from the legacy one so the room is
// reconstituted — repo records + the signed __commit__ firehose + membership. Only the writable
// creator seeds (peers replicate); no-op if the renamed base already has entries, if this instance
// IS the legacy namespace, or if no legacy base is present. Idempotent across reboots.
//
// Carries every DIRECT set-entry entry (repo records, __members__, __devices__, __commit__/*,
// prefs) verbatim. It does NOT carry the apply-DERIVED governance entries __removed__ (bans) /
// __receipt__ (min-epoch) — those are produced inside apply from admission-proof ops, not stored as
// plain set-entry ops, so they re-establish from config (re-issue a ban after migration if needed).
async _migrateFromLegacyNamespace () {
try {
if (!this._roomSeed || !this.base || !this.base.writable) return
if (Buffer.compare(this._gsNamespace, LEGACY_GS_NAMESPACE) === 0) return // already the legacy view
const existing = await this.list('')
if (Object.keys(existing).length > 0) return // renamed base already populated — nothing to do
const legacyBaseKey = deriveBaseKey(deriveGsKeyPair(this._roomSeed, LEGACY_GS_NAMESPACE).publicKey)
if (Buffer.compare(legacyBaseKey, this._detBaseKey) === 0) return // no rename in effect
// Open the legacy base READ-ONLY on the SAME corestore (no fd-lock conflict — one store, two
// autobases), with a minimal apply that only materializes direct set/del entries.
const router = this._makeLegacyReadRouter()
const opts = {
open: (store) => HyperDB.bee(store.get('view'), hyperdbDefinition, { autoUpdate: true }),
apply: async (nodes, view) => {
const tx = view.transaction()
try {
for (const node of nodes) { const op = node.value; if (Buffer.isBuffer(op)) await router.dispatch(op, { tx }) }
await tx.flush()
} catch (e) { try { await tx.close() } catch {} ; throw e }
},
close: async (view) => { if (view) await view.close() },
valueEncoding: 'binary', optimistic: true, ackInterval: 1000, ackThreshold: 4
}
if (this._encryptionKey) { opts.encryptionKey = this._encryptionKey; opts.encrypted = true }
// Open on a Corestore SESSION, not this.store directly: a session shares the same storage +
// cores (so it sees the legacy base) but has its own ref-counting, so closing it releases only
// the migration's core sessions — it never tears down cores the live base is still using
// (SESSION_CLOSED). No second fd-lock (same underlying storage).
const legacyStore = this.store.session()
// Cheap presence probe FIRST: the legacy creator's ops live in the bootstrap core (it opened
// with that keyPair), so a real legacy room has length > 0 there. A genuinely fresh room has
// no such data — skip WITHOUT opening a full legacy Autobase (which would spawn ~10 empty
// cores on every first boot). One tiny empty probe core is the only footprint.
const probe = legacyStore.get({ key: legacyBaseKey })
await probe.ready()
const hasLegacy = probe.length > 0
if (!hasLegacy) { try { await legacyStore.close() } catch {} ; return }
const legacy = new Autobase(legacyStore, legacyBaseKey, opts)
try {
await legacy.ready()
await legacy.update()
const entries = []
const stream = legacy.view.find(ENTRIES, { gte: { key: '' }, lt: { key: '~' } })
for await (const rec of stream) entries.push(rec)
if (entries.length === 0) return
let n = 0
for (const rec of entries) {
// Copy verbatim (value is already the stored encoded string). Direct append — the creator
// is writable, so no doorbell needed (mirrors projectMembers).
await this.base.append(encode('@state/set-entry', { key: rec.key, value: rec.value, ts: rec.ts || 0 }))
n++
}
await this.base.update()
console.log('[state-base-v3] room=%s RE-ADOPT MIGRATED %d entries from legacy namespace (base %s)',
this._roomKeyHex.slice(0, 12), n, legacyBaseKey.toString('hex').slice(0, 12))
} finally { try { await legacy.close() } catch {} ; try { await legacyStore.close() } catch {} }
} catch (e) { console.log('[state-base-v3] legacy migration skipped: %s', e && e.message) }
}
// Minimal read-only router for materializing a legacy base's view: LWW-insert direct set-entry
// ops, LWW-delete del-entry ops. Skips the transient admission/indexer/ban/epoch REQUEST ops
// (empty key + the reserved __*__ command keys) — those carry no durable entry (their effects are
// applied by the full router). Enough to reproduce every stored entry we migrate.
_makeLegacyReadRouter () {
const TRANSIENT = new Set(['', MOVE_INDEXER, START_INDEXING, REMOVE_DEVICE, SET_EPOCH])
const r = new Router()
r.add('@state/set-entry', async (rec, ctx) => {
if (!rec.key || TRANSIENT.has(rec.key)) return
const existing = await ctx.tx.get(ENTRIES, { key: rec.key })
if (existing && typeof existing.ts === 'number' && existing.ts > (rec.ts || 0)) return
await ctx.tx.insert(ENTRIES, { key: rec.key, value: rec.value, ts: rec.ts || 0 })
})
r.add('@state/del-entry', async (rec, ctx) => {
const existing = await ctx.tx.get(ENTRIES, { key: rec.key })
if (existing && typeof existing.ts === 'number' && existing.ts > (rec.ts || 0)) return
await ctx.tx.delete(ENTRIES, { key: rec.key })
})
return r
}
// ── Membership projection (cross-base sync) ────────────────
// Daemon's out-of-apply watcher copies keet-room membership in so apply can
// Event-driven, self-terminating __members__ re-projection (mirrors _maybePromote). Fires on the
// promote-timer tick + every new swarm connection. If we retain a roster (set by projectMembers),
// haven't confirmed __members__ yet, and are writable, project it now; once landed it early-returns
// forever (cheap when done). This is the fix for the wj62 fork: it replaces the app's bounded 12-try
// boot loop, so a creator that becomes writable AFTER boot still lands __members__ and admits members.
// No hang: projectMembers's base.update() drains OUR OWN writable base (clean); a non-writable device
// no-ops here and retries on the next event.
async _ensureMembersProjected () {
if (this._membersProjected || !this._roster || !this._ready || !this.base || !this.base.writable) return
try { await this.projectMembers(this._roster) } catch {}
}
// gate admission deterministically. Stored under the reserved key __members__
// (a memberId-hex → role map). No-op until we're a writer (another writer
// projects; peers replicate it).
async projectMembers (members) {
if (members) this._roster = members // retain the roster for event-driven, self-terminating re-projection
if (!this._ready || !this.base || !this.base.writable) return false // not writable yet → a later event retries
const map = {} // memberKey(hex) -> role
const devmap = {} // keet writer/device key(hex) -> memberKey(hex) (for admission model A)
if (Array.isArray(members)) {
for (const m of members) {
const id = normKeyHex(m.key || m.devicePubkey || m.identity || m.memberId || m.id)
if (!id) continue
map[id] = Number(m.role) || 3
const devs = Array.isArray(m.devices) ? m.devices : []
for (const d of devs) { const dh = normKeyHex(d); if (dh) devmap[dh] = id }
}
} else if (members && typeof members === 'object') {
for (const id of Object.keys(members)) { const h = normKeyHex(id); if (h) map[h] = Number(members[id]) || 3 }
}
// SELF-HEAL: record THIS node's own writer->identity mapping. __devices__ is otherwise written
// only at ADMISSION time (_applySet, while a writer still carries an `admit` proof), so once an
// entry is lost it is never rewritten — a device that is already admitted never re-announces. Each
// node knows its own pair authoritatively, so projecting it here repopulates the map organically.
// Mirrors atproto-base's projectMembers (base.js:313). ctx.node.from.key in _applySet IS this
// base.local.key, so the two writers agree on the key space.
if (this._selfKey && this.base.local && this.base.local.key) {
devmap[this.base.local.key.toString('hex')] = this._selfKey.toString('hex')
}
// MERGE __devices__, never overwrite. hpp's roster is [{key, role}] with NO `devices` array
// (device-node.js), so devmap was always {} — and an unconditional overwrite stamped with a fresh
// ts therefore WIPED every entry admission had accumulated. ban()/REMOVE_DEVICE iterates this map
// to removeWriter a banned member's devices, so an empty map made revocation silently remove
// NOTHING (observed on both live DIDs: devicesInView {}). Additive only — never drops an entry.
const prevDev = (await this.getNow('__devices__').catch(() => null)) || {}
const mergedDev = { ...prevDev, ...devmap }
// Never write an EMPTY roster over a populated one. Admission verifies every joiner's proof against
// __members__ (_verifyAdmissionInApply), so wiping it denies every future admission — that is the
// wj62 fork signature. A caller with no roster must be a no-op, not a wipe.
const prevMembers = (await this.getNow('__members__').catch(() => null)) || {}
if (Object.keys(map).length === 0 && Object.keys(prevMembers).length > 0) {
console.log('[state-base-v3] room=%s projectMembers REFUSED: empty roster would wipe %d existing member(s)',
this._roomKeyHex.slice(0, 12), Object.keys(prevMembers).length)
return false
}
// Append only what actually changes — these run on the promote timer AND every swarm connection.
if (!sameMap(prevMembers, map)) await this.base.append(encode('@state/set-entry', { key: '__members__', value: JSON.stringify(map), ts: Date.now() }))
if (!sameMap(prevDev, mergedDev)) await this.base.append(encode('@state/set-entry', { key: '__devices__', value: JSON.stringify(mergedDev), ts: Date.now() }))
await this.base.update()
this._membersProjected = true // __members__ appended + drained into our view → stop re-projecting (early-returns cheaply)
// P3 (#103): a re-projection is the moment a role change becomes visible. If THIS
// node was just granted admin, re-check promotion immediately rather than waiting
// for the next timer tick. Background + best-effort; the timer is the general
// fallback for role changes replicated in from another writer's projection.
this._maybePromote().catch(() => {})
return true
}
async getProjectedRole (id) {
if (!this._ready || !this.base) return null
await this.base.update()
const mrec = await this.base.view.get(ENTRIES, { key: '__members__' })
if (!mrec || !mrec.value) return null
let members; try { members = JSON.parse(mrec.value) } catch { return null }
const h = normKeyHex(id)
const role = members[h] != null ? members[h] : members[id]
return role != null ? Number(role) : null
}
// ── Admission proof (op-embedded) ──────────────────────────
// Proof attests (over localKey‖base.key, so it can't be replayed onto another
// base) that `identity` vouches for this writer's local key. keet attestation
// (preferred) or raw ed25519 (fallback).
async _buildAdmissionProof () {
if (!this.base || !this.base.local || !this.base.local.key || !this.base.key) return null
const localKey = this.base.local.key
const signData = Buffer.concat([localKey, this.base.key])
// (A) keet-writer-control: sign signData with our keet ROOM writer keypair and
// carry that writer's manifest. The verifier derives deviceKey=Hypercore.key(manifest),
// checks sig against manifest.signers[0].publicKey, maps device→member→role.
if (this._roomWriterKeyPair && this._roomWriterKeyPair.secretKey && this._roomWriterManifest) {
const sig = crypto.sign(signData, toBuf(this._roomWriterKeyPair.secretKey))
const manifestEnc = Verifier.encodeManifest(this._roomWriterManifest)
return JSON.stringify({ manifest: Buffer.from(manifestEnc).toString('hex'), sig: sig.toString('hex') })
}
// legacy keet-attestation / raw-ed25519 fallback (older callers + synthetic tests)
if (this._attest && this._selfKey) {
const proof = await this._attest(signData)
if (!proof) return null
const pbuf = Buffer.isBuffer(proof) ? proof : Buffer.from(proof)
return JSON.stringify({ identity: this._selfKey.toString('hex'), proof: pbuf.toString('hex') })
}
if (this._selfKeyPair && this._selfKeyPair.secretKey) {
const sig = crypto.sign(signData, toBuf(this._selfKeyPair.secretKey))
return JSON.stringify({ identity: toBuf(this._selfKeyPair.publicKey).toString('hex'), proof: sig.toString('hex') })
}
return null
}
// DETERMINISTIC in-apply verification (pure crypto + projected __members__).
async _verifyAdmissionInApply (localKey, admitStr, tx) {
try {
const admit = typeof admitStr === 'string' ? JSON.parse(admitStr) : admitStr
if (!admit || !this.base || !this.base.key) return null
const signData = Buffer.concat([localKey, this.base.key])
// (A) keet-writer-control proof: { manifest, sig }. Re-derive the keet writer/
// device key from the manifest, verify the sig against the manifest's signer,
// then resolve device→member (__devices__) → role (__members__).
if (admit.manifest && admit.sig) {
let manifest, device, signerPub
try {
manifest = Verifier.decodeManifest(Buffer.from(admit.manifest, 'hex'))
device = Hypercore.key(manifest)
signerPub = manifest.signers && manifest.signers[0] && manifest.signers[0].publicKey
} catch (e) { console.log('[state-base-v3] verify: bad manifest %s', e.message); return null }
if (!signerPub) { console.log('[state-base-v3] verify: manifest has no signer'); return null }
if (!crypto.verify(signData, Buffer.from(admit.sig, 'hex'), signerPub)) {
console.log('[state-base-v3] verify: writer sig INVALID for device=%s', device.toString('hex').slice(0, 12)); return null
}
const drec = await tx.get(ENTRIES, { key: '__devices__' })
const devmap = drec && drec.value ? JSON.parse(drec.value) : {}
const deviceHex = device.toString('hex')
const memberKey = devmap[deviceHex] != null ? devmap[deviceHex] : devmap[normKeyHex(deviceHex)]
if (!memberKey) { console.log('[state-base-v3] verify: device=%s NOT in __devices__', deviceHex.slice(0, 12)); return null }
const mrec = await tx.get(ENTRIES, { key: '__members__' })
const members = mrec && mrec.value ? JSON.parse(mrec.value) : {}
const role = members[memberKey] != null ? members[memberKey] : members[normKeyHex(memberKey)]
if (role == null) console.log('[state-base-v3] verify: member=%s (device %s) NOT in __members__', String(memberKey).slice(0, 12), deviceHex.slice(0, 12))
if (role != null && await this._isRemoved(tx, memberKey, deviceHex)) { console.log('[state-base-v3] verify: member=%s is REMOVED (banned)', String(memberKey).slice(0, 12)); return null }
return role != null ? Number(role) : null
}
// legacy keet-attestation / raw-ed25519 proof: { identity, proof }
if (!admit.identity || !admit.proof) return null
const identity = Buffer.from(admit.identity, 'hex')
const proof = Buffer.from(admit.proof, 'hex')
let ok = false
if (this._verifyAttest) {
// Pass the current minimum-epoch receipt: verifyAttest rejects a proof whose epoch is below
// it (atproto-base-REVOCATION §3 — invalidate stale/revoked proofs after an epoch bump).
const rcpt = await tx.get(ENTRIES, { key: RECEIPT_KEY })
const receipt = rcpt && rcpt.value ? Buffer.from(this._dec(rcpt.value), 'hex') : null
ok = await this._verifyAttest(proof, signData, identity, receipt)
} else { if (proof.length !== 64) { console.log('[state-base-v3] verify: bad proof len'); return null }; ok = crypto.verify(signData, proof, identity) }
if (!ok) { console.log('[state-base-v3] verify: proof INVALID or below min-epoch for identity=%s (verifyAttest=%s)', admit.identity.slice(0, 12), !!this._verifyAttest); return null }
const mrec = await tx.get(ENTRIES, { key: '__members__' })
if (!mrec || !mrec.value) { console.log('[state-base-v3] verify: no __members__ in view'); return null }
const members = JSON.parse(mrec.value)
const role = members[admit.identity] != null ? members[admit.identity]
: (members[normKeyHex(admit.identity)] != null ? members[normKeyHex(admit.identity)] : null)
if (role == null) console.log('[state-base-v3] verify: identity=%s NOT in __members__ (keys=%s)', admit.identity.slice(0, 12), Object.keys(members).map(k => k.slice(0, 8)).join(','))
if (role != null && await this._isRemoved(tx, admit.identity)) { console.log('[state-base-v3] verify: identity=%s is REMOVED (banned)', admit.identity.slice(0, 12)); return null }
return role != null ? Number(role) : null
} catch (e) { console.log('[state-base-v3] verify: threw %s', e.message); return null }
}
watch (fn) { this._watchers.add(fn); return () => this._watchers.delete(fn) }
// `writable` answers "will my durable writes persist?" — true once autobase has
// activated our local writer (base.writable), OR an indexer has already admitted
// our local writer into the linearized system writer set (cached
// _selfIsAdmittedWriter). The admitted-writer signal is true at/before
// base.writable's localWriter activation, so it still covers the post-admission
// lag WITHOUT the false-positive of keying off __members__: room membership is a
// SUPERSET of admitted writers, so an eligible member the indexer never admitted
// (e.g. an identity absent from the indexer's own membership) must NOT read
// writable. Sync getter uses the cache; callers wanting a fresh answer await
// effectiveWritable(). See hyper-fleeky-dev #99.
get writable () { return this.base?.writable || false }
// Fresh "will my writes persist?" — refreshes the _selfIsAdmittedWriter cache.
async effectiveWritable () {
if (this.base && this.base.writable) return true
return await this._refreshSelfWriter()
}
// Re-derive whether our LOCAL WRITER KEY (base.local.key) is in the autobase
// linearized system writer set (base.system) — i.e. an indexer has durably
// admitted us, so our writes finalize. This is the authoritative "admitted"
// signal — the same one autobase's own fast-forward consults
// (lib/fast-forward.js: system.get(base.local.key)). It is NOT mere room
// membership: __members__ gates admission (_verifyAdmissionInApply), but being
// in it does not mean an indexer actually admitted us (#99). base.system
// (=_applyState.system) is null transiently on reboot/close → keep the cache.
async _refreshSelfWriter () {
try {
if (!this._ready || !this.base || !this.base.local || !this.base.local.key) return this._selfIsAdmittedWriter
await this.base.update()
const sys = this.base.system
if (!sys) return this._selfIsAdmittedWriter
this._selfIsAdmittedWriter = await sys.has(this.base.local.key)
return this._selfIsAdmittedWriter
} catch { return this._selfIsAdmittedWriter }
}
get stateBaseKey () { return this.base?.key || null }
get hasBootstrap () { return !!this._detBaseKey }
// ── creator election (v2 logic) ────────────────────────────
async _amCreator () {
if (!this._getMembers || !this._selfKey) return false
let members
try { members = await this._getMembers() } catch { return false }
if (!Array.isArray(members) || members.length === 0) return false
const norm = (k) => (Buffer.isBuffer(k) ? k.toString('hex') : String(k)).toLowerCase()
const eligible = members
.map(m => ({ key: norm(m.key || m.devicePubkey || m.identity || m.memberId), role: Number(m.role) }))
.filter(m => m.key && (m.role === 1 || m.role === 2))
.sort((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0))
if (eligible.length === 0) return false
return eligible[0].key === this._selfKey.toString('hex')
}
// ── corruption recovery (same posture as v2 1.0.2) ─────────
_isCorruptionError (err) {
const m = (err && err.message ? err.message : String(err)).toUpperCase()
return m.includes('DECODING_ERROR') || m.includes('UNKNOWN WIRE TYPE') || m.includes('CORRUPT') ||
m.includes('INVALID DEVICE FILE') || m.includes('CHECKSUM') || m.includes('INVALID OPLOG') || m.includes('INVALID HEADER')
}
async _quarantineAndReset () {
if (this.swarm) { try { await this.swarm.destroy() } catch {} ; this.swarm = null }
if (this.base) { try { await this.base.close() } catch {} ; this.base = null }
if (this.store) { try { await this.store.close() } catch {} ; this.store = null }
const quarantine = this._storagePath + '.corrupt-' + Date.now()
fs.renameSync(this._storagePath, quarantine)
console.error('[state-base-v3] room=%s quarantined -> %s', this._roomKeyHex.slice(0, 12), quarantine)
fs.mkdirSync(this._storagePath, { recursive: true })
this.store = new Corestore(this._storagePath)
await this.store.ready()
await this._open()
}
async close () {
if (this._closing) return
this._closing = true
if (this._promoteTimer) { clearInterval(this._promoteTimer); this._promoteTimer = null }
if (this.swarm) { await this.swarm.destroy(); this.swarm = null }
if (this.base) { await this.base.close(); this.base = null }
if (this.store) { await this.store.close(); this.store = null }
this._watchers.clear()
this._ready = false
this._closing = false
}
}