OverviewFilesCommitsIssuesDocsReleasesPeers

atproto-room commits / 18a988f

state-base: never hang the read/lease/admission path on a stalled base.update() (Phase 2, hyper-social #6)
A plain writer that rebooted writable=false (admission not durably signed before shutdown — the power-cut window) hung on base.update(): the sign loop's first read (lease.current → sb.get → base.update) stalled on the un-converged replica, so the re-admission doorbell never re-rang → never re-admitted → circular. get-state hung >8s on live tabertron. - state-base: _updateBounded() races base.update() against an 800ms timeout (safe: incoming replication drives background drains that still advance base.writable). requestWriter's _isSelfRemoved + post-append updates and set()/del() now use it, so the doorbell block is always appended even when a drain stalls. - ensureAdmitted(): guarded background re-admission doorbell — a booted-but-not-writable plain writer re-rings requestWriter so an online indexer re-admits + signs a durable checkpoint. - lease.current(): read __lease__ via non-blocking getNow snapshot (the sign-loop first read). - repo-base: listRecords({ now:true }) reads the listNow snapshot for liveness-critical callers. - test: deterministic proof (patch base.update to hang forever → read/lease/doorbell path still returns + doorbell block still fires); + happy-path durability repro. All regressions green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
18a988fc05122ff80bf8bf92be5d1cf0ad909cfa
by plankatron <fleeky@gmail.com> · 2026-07-17T02:44:48.000Z

Changed files (7) +251 -9

lib/lease.js +9 -1
……@@ -24,8 +24,16 @@ export class Lease {
2424 }
2525
2626 // The current lease record from the linearized view, or null.
27 //
28 // NON-BLOCKING read (Phase-2 #6): use the statebase's getNow snapshot when available. current()
29 // is the FIRST call the sign loop makes every tick (renewOrClaim → renew/claim → current). If it
30 // went through the blocking get()→base.update(), a single stalled drain on an un-converged replica
31 // froze the ENTIRE sign loop — so it never reached the re-admission doorbell (the live tabertron
32 // hang: get-state over UDS hung here). The lease is wall-clock-TTL + "never steal a live lease",
33 // so a few-ms-stale snapshot is safe (the autobase still linearizes concurrent claims to one
34 // winner). Falls back to get() for injected substrates without getNow.
2735 async current () {
28 const v = await this.sb.get(LEASE_KEY)
36 const v = typeof this.sb.getNow === 'function' ? await this.sb.getNow(LEASE_KEY) : await this.sb.get(LEASE_KEY)
2937 return v && v.holder ? v : null
3038 }
3139
lib/repo-base.js +6 -2
……@@ -72,9 +72,13 @@ export class RepoBase {
7272
7373 // List the repo's records (optionally scoped to one collection) as [{collection, rkey, record}],
7474 // skipping any reserved/non-record keys the underlying StateBase may hold (e.g. __members__).
75 async listRecords (collection = null) {
75 //
76 // { now:true } (Phase-2 #6) reads the non-blocking autoUpdate snapshot (sb.listNow) instead of the
77 // blocking sb.list()→base.update(), so liveness-critical callers (device.state()/get-state) never
78 // hang on an un-converged replica. Falls back to list() if the substrate has no listNow.
79 async listRecords (collection = null, { now = false } = {}) {
7680 const prefix = collection ? (assertSegment('collection', collection), collection + SEP) : ''
77 const map = await this.sb.list(prefix)
81 const map = (now && typeof this.sb.listNow === 'function') ? await this.sb.listNow(prefix) : await this.sb.list(prefix)
7882 const out = []
7983 for (const key of Object.keys(map)) {
8084 const split = splitKey(key)
lib/state-base.js +50 -5
……@@ -393,14 +393,16 @@ export class StateBaseV3 {
393393 await this._ensureWritable()
394394 const rec = { key: String(key), value: this._enc(value), ts: Date.now() }
395395 await this.base.append(encode('@state/set-entry', rec))
396 await this.base.update()
396 // BOUNDED: the content is appended (durable-local + replicated); a stalled drain must not hang
397 // the writer indefinitely (Phase-2 #6). The value is still readable back via the autoUpdate view.
398 await this._updateBounded()
397399 }
398400
399401 async del (key) {
400402 if (!this._ready) throw new Error('StateBaseV3 not ready')
401403 await this._ensureWritable()
402404 await this.base.append(encode('@state/del-entry', { key: String(key), ts: Date.now() }))
403 await this.base.update()
405 await this._updateBounded()
404406 }
405407
406408 // Block until we are a genuinely ADMITTED writer (our local writer key is in the linearized
……@@ -447,7 +449,10 @@ export class StateBaseV3 {
447449 // None insert a game entry.
448450 const key = claimIndexer ? MOVE_INDEXER : (startIndexing ? START_INDEXING : '')
449451 await this.base.append(encode('@state/set-entry', { key, admit, ts: Date.now() }), { optimistic: true })
450 await this.base.update()
452 // BOUNDED, not blocking: an un-converged replica's base.update() can hang forever (Phase-2 #6).
453 // The optimistic doorbell block is already appended (durable-local + replicated via the swarm);
454 // bounding the drive lets the loop keep re-ringing + re-polling base.writable instead of freezing.
455 await this._updateBounded()
451456 const deadline = Date.now() + timeoutMs
452457 while (!done() && Date.now() < deadline) {
453458 await new Promise(r => setTimeout(r, 250))
……@@ -458,7 +463,7 @@ export class StateBaseV3 {
458463 if (!this.base.writable) {
459464 try { await this.base.append(encode('@state/set-entry', { key, admit, ts: Date.now() }), { optimistic: true }) } catch {}
460465 }
461 await this.base.update().catch(() => {})
466 await this._updateBounded()
462467 }
463468 return { writable: this.base.writable, indexer: this._isLocalIndexer() }
464469 }
……@@ -468,6 +473,42 @@ export class StateBaseV3 {
468473 try { return this.base.isIndexer === true } catch { return null }
469474 }
470475
476 // Drive the linearizer, but NEVER hang the caller. base.update() awaits a full drain, which can
477 // stall INDEFINITELY on an un-converged plain-writer replica (device-swarm still catching up) —
478 // the Phase-2 hang (hyper-social #6): the sign/admission path awaited it and froze, so the
479 // re-admission doorbell never re-rang. This bounds how long we AWAIT one drain. It is SAFE to
480 // time out: convergence does NOT depend on our awaiting update() — incoming replication triggers
481 // background drains (autobase _onwakeup/_onremotewriterchange → _queueBump) that still advance
482 // base.writable — so a bounded await lets the doorbell loop keep re-ringing + re-polling the
483 // (sync) writable getter while the real work proceeds in the background. Same posture as the
484 // getNow/listNow snapshot reads and the boot __members__ time-box.
485 async _updateBounded (ms = 800) {
486 if (!this.base) return
487 let t = null
488 try {
489 await Promise.race([
490 this.base.update().catch(() => {}),
491 new Promise((r) => { t = setTimeout(r, ms); if (t && typeof t.unref === 'function') t.unref() })
492 ])
493 } finally { if (t) clearTimeout(t) }
494 }
495
496 // Background, guarded re-admission doorbell for a booted-but-not-yet-writable plain writer.
497 // Rationale (Phase 2): a plain writer admitted at RUNTIME may reboot writable=false if an indexer
498 // never SIGNED a checkpoint covering its admission that the writer persisted (the power-cut
499 // window). Re-admission needs the writer to re-ring requestWriter — but the lease/sign path only
500 // writes when it can hold the lease, and a healthy peer (stortron) holds it, so that path never
501 // fires the doorbell. This is the independent trigger: the sign loop calls it each tick; it rings
502 // requestWriter ONCE at a time (a call while one is in flight is a no-op), so an online indexer
503 // re-admits us and signs a durable checkpoint we then persist. No-op once writable.
504 async ensureAdmitted ({ timeoutMs = 15000 } = {}) {
505 if (!this._ready || !this.base || this.base.writable) return this.base ? this.base.writable : false
506 if (this._ensuringAdmit) return false
507 this._ensuringAdmit = true
508 try { await this.requestWriter({ timeoutMs }) } catch {} finally { this._ensuringAdmit = false }
509 return this.base ? this.base.writable : false
510 }
511
471512 async get (key) {
472513 if (!this._ready) return null
473514 await this.base.update()
……@@ -669,7 +710,11 @@ export class StateBaseV3 {
669710 async _isSelfRemoved () {
670711 try {
671712 if (!this._ready || !this.base || !this._selfKey) return false
672 await this.base.update()
713 // BOUNDED (Phase-2 #6): this runs at the TOP of requestWriter, before the doorbell append —
714 // a blocking base.update() here would hang the doorbell on an un-converged replica and it
715 // would never re-ring. Best-effort anyway (false on uncertainty), so reading a slightly stale
716 // ban snapshot is acceptable; the autoUpdate view is still current within a few ms.
717 await this._updateBounded()
673718 const rrec = await this.base.view.get(ENTRIES, { key: REMOVED_SET })
674719 const removed = rrec && rrec.value ? JSON.parse(rrec.value) : {}
675720 const hex = this._selfKey.toString('hex')
package.json +1 -1
……@@ -1,6 +1,6 @@
11{
22 "name": "atproto-room",
3 "version": "0.2.1",
3 "version": "0.2.2",
44 "description": "Attestation-gated, encrypted multi-writer room engine \u2014 StateBase (keet-model admission) + Lease + signer + RepoBase + LabelBase/LabelRoom + serve-membrane. Extracted from hyper-proto-peer; consumed by the atproto peer (repo devices) and the co-authored labeler.",
55 "type": "module",
66 "main": "index.js",
test/all.js +1 -0
……@@ -10,6 +10,7 @@ import './signer.test.js'
1010import './signer-conformance.test.js'
1111import './membership.test.js'
1212import './doorbell-durable-write.test.js'
13import './phase2-nonblocking.test.js'
1314import './non-indexer-default.test.js'
1415import './co-authored-labeler.test.js'
1516import './migration-readopt.test.js'
test/phase2-admission-durability.repro.mjs +96 -0
……@@ -0,0 +1,96 @@
1// Phase-2 admission-durability OFFLINE repro (hyper-social #6).
2//
3// Reproduces the tabertron symptom: a plain writer (B) that was admitted once boots writable=false
4// on RESTART, and base.update() HANGS — so the sign/read/admission path (which awaits base.update())
5// never re-rings the doorbell and B never re-admits. A (creator/sole N=1 indexer) stays ONLINE the
6// whole time, mirroring stortron being healthy.
7//
8// This is a PROBE, not a pass/fail test — it prints exactly what happens at each step so we
9// characterize the bug before fixing (never guess-fix the live wj62 StateBase). Run:
10// cd atproto-room && bare test/phase2-admission-durability.repro.mjs
11import '../lib/bare-globals.js'
12import crypto from 'hypercore-crypto'
13import fs from 'bare-fs'
14import b4a from 'b4a'
15import createTestnet from '@hyperswarm/testnet'
16import { StateBaseV3 } from '../lib/state-base.js'
17import { makeAttest, makeVerifyAttest, attestDeviceChain, deviceKeetKeyPair, rootIdentityPublicKey } from '../lib/membership.js'
18
19const ROOT = 'abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about'
20const DEVA = 'legal winner thank year wave sausage worth useful legal winner thank yellow'
21const DEVB = 'zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo wrong'
22const sleep = (ms) => new Promise(r => setTimeout(r, ms))
23const RUN = '/tmp/hpp-phase2-' + Math.floor(Math.random() * 1e9)
24
25// Race any promise against a timeout so a HANG is observable instead of hanging the whole script.
26async function timed (label, p, ms = 4000) {
27 const t0 = Date.now()
28 const HUNG = Symbol('hung')
29 const r = await Promise.race([p.then(v => ({ v })).catch(e => ({ err: e })), sleep(ms).then(() => HUNG)])
30 const dt = Date.now() - t0
31 if (r === HUNG) { console.log(` [${label}] HUNG (>${ms}ms) <-- did not resolve`); return { hung: true } }
32 if (r.err) { console.log(` [${label}] threw after ${dt}ms: ${r.err.message}`); return { err: r.err } }
33 console.log(` [${label}] ok in ${dt}ms`)
34 return { v: r.v }
35}
36
37async function main () {
38 fs.mkdirSync(RUN, { recursive: true })
39 const testnet = await createTestnet(3)
40 const rootPub = await rootIdentityPublicKey(ROOT)
41 let da = await deviceKeetKeyPair(DEVA); let pa = await attestDeviceChain({ rootMnemonic: ROOT, devicePublicKey: da.publicKey })
42 let db = await deviceKeetKeyPair(DEVB); let pb = await attestDeviceChain({ rootMnemonic: ROOT, devicePublicKey: db.publicKey })
43 let A = { dev: da, proof: pa }, B = { dev: db, proof: pb }
44 if (b4a.compare(A.dev.publicKey, B.dev.publicKey) > 0) { const t = A; A = B; B = t } // creator = lowest key
45 const members = [
46 { key: b4a.toString(A.dev.publicKey, 'hex'), role: 1 },
47 { key: b4a.toString(B.dev.publicKey, 'hex'), role: 1 }
48 ]
49 const roomSeed = crypto.randomBytes(32); const roomKey = crypto.randomBytes(32); const encryptionKey = crypto.randomBytes(32)
50 const dirA = RUN + '/A'; const dirB = RUN + '/B'
51 const mk = (dir, dev, proof) => new StateBaseV3(dir, roomKey, {
52 roomSeed, encryptionKey, swarmOpts: { bootstrap: testnet.bootstrap },
53 selfKey: dev.publicKey, attest: makeAttest(dev, proof), verifyAttest: makeVerifyAttest(rootPub),
54 getMembers: async () => members
55 })
56
57 console.log('\n=== STEP 1: bring up A (creator/sole N=1 indexer) + admit B ===')
58 const sbA = mk(dirA, A.dev, A.proof); await sbA.ready(); await sbA.projectMembers(members)
59 let sbB = mk(dirB, B.dev, B.proof); await sbB.ready()
60 console.log('A.writable=%s A.isIndexer=%s', sbA.base.writable, sbA.base.isIndexer)
61 await sbB.set('post/1', { text: 'first-post' }) // doorbell → admit → RUNTIME writable
62 console.log('B.writable after first set()=%s B.local.length=%s', sbB.base.writable, sbB.base.local.length)
63
64 console.log('\n=== STEP 2: DO NOT converge — simulate a power cut right after admission ===')
65 // B is writable at RUNTIME (linearized system) but A (ackInterval 1000ms) may not yet have SIGNED
66 // an indexed checkpoint covering B's admission that B persisted. That is the power-cut window.
67 const persignHasB = sbB.base.system ? await sbB.base.system.has(sbB.base.local.key) : null
68 console.log('B.system.has(B.local) [durably signed?]=%s (false => reboot will be writable=false)', persignHasB)
69
70 console.log('\n=== STEP 3: CLOSE B abruptly (power cut), A STAYS ONLINE ===')
71 await sbB.close()
72 await sleep(500)
73
74 console.log('\n=== STEP 4: REOPEN B on the SAME storage, A still online — the restart under test ===')
75 sbB = mk(dirB, B.dev, B.proof)
76 await sbB.ready()
77 console.log('B.writable at boot (from persisted signed checkpoint)=%s', sbB.base.writable)
78 console.log('B.local.length=%s B.local.key=%s', sbB.base.local.length, sbB.base.local.key.toString('hex').slice(0, 12))
79
80 console.log('\n=== STEP 5: probe the sign/read path (each raced against a 4s timeout) ===')
81 await timed('B.base.update()', sbB.base.update())
82 await timed('B.get(post/1) [blocking, via base.update]', sbB.get('post/1'))
83 await timed('B.getNow(post/1) [non-blocking snapshot]', sbB.getNow('post/1'))
84 await timed('B.lease.current() [sign-loop first read]', (async () => { const { Lease } = await import('../lib/lease.js'); return new Lease(sbB, { self: b4a.toString(B.dev.publicKey, 'hex') }).current() })())
85 await timed('B.set(post/2) [_ensureWritable→requestWriter doorbell]', sbB.set('post/2', { text: 'second-post' }), 25000)
86 console.log('B.writable after set(post/2)=%s', sbB.base.writable)
87
88 console.log('\n=== STEP 6: give convergence more time, re-check ===')
89 for (let i = 0; i < 20; i++) { await sbA.base.update().catch(() => {}); await sbB.base.update().catch(() => {}); await sleep(300) }
90 console.log('B.writable=%s B.effectiveWritable=%s', sbB.base.writable, await sbB.effectiveWritable())
91
92 await sbA.close().catch(() => {}); await sbB.close().catch(() => {}); await testnet.destroy()
93 console.log('\n=== done ===')
94 process.exit(0)
95}
96main().catch(e => { console.error('REPRO ERROR', e); process.exit(1) })
test/phase2-nonblocking.test.js +88 -0
……@@ -0,0 +1,88 @@
1// Phase-2 admission-durability mechanism test (hyper-social #6).
2//
3// Deterministic proof that the sign/read/admission path NEVER hangs on a stalled base.update() —
4// which is the live tabertron failure (get-state hung >8s; the sign loop's first read hung, so the
5// re-admission doorbell never re-rang). We simulate the stall by replacing base.update() with a
6// promise that NEVER resolves, then assert every liveness-critical call still returns quickly and
7// the doorbell block is still appended. This isolates the FIX from the (hard-to-reproduce) exact
8// convergence fork that causes the real stall — the guarantee we ship is "a stalled drain can't
9// freeze these paths", and that is what this proves.
10import '../lib/bare-globals.js'
11import test from 'brittle'
12import crypto from 'hypercore-crypto'
13import fs from 'bare-fs'
14import b4a from 'b4a'
15import { StateBaseV3 } from '../lib/state-base.js'
16import { Lease } from '../lib/lease.js'
17import { RepoBase } from '../lib/repo-base.js'
18import { makeAttest, makeVerifyAttest, attestDeviceChain, deviceKeetKeyPair, rootIdentityPublicKey } from '../lib/membership.js'
19
20const ROOT = 'abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon abandon about'
21const DEVA = 'legal winner thank year wave sausage worth useful legal winner thank yellow'
22const DEVB = 'zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo wrong'
23const tmp = () => { const d = '/tmp/hpp-phase2-nb-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }
24
25// Resolve p, but fail the assertion if it takes longer than ms (i.e. it HUNG).
26async function within (t, label, p, ms) {
27 const HUNG = Symbol('hung'); const t0 = Date.now()
28 const r = await Promise.race([p.then(v => ({ v })).catch(e => ({ err: e })), new Promise(res => setTimeout(() => res(HUNG), ms))])
29 const dt = Date.now() - t0
30 t.ok(r !== HUNG, `${label} returned in ${dt}ms (< ${ms}ms) — did not hang`)
31 return r === HUNG ? undefined : (r.err ? { threw: r.err } : r.v)
32}
33
34test('a stalled base.update() cannot freeze the read / lease / doorbell path', { timeout: 60000 }, async (t) => {
35 const rootPub = await rootIdentityPublicKey(ROOT)
36 const da = await deviceKeetKeyPair(DEVA); const pa = await attestDeviceChain({ rootMnemonic: ROOT, devicePublicKey: da.publicKey })
37 const db = await deviceKeetKeyPair(DEVB); const pb = await attestDeviceChain({ rootMnemonic: ROOT, devicePublicKey: db.publicKey })
38 // member = the NOT-lowest key, so it opens writable=false (member, no indexer present) — exactly
39 // the un-converged plain-writer posture tabertron reboots into.
40 let A = da, B = db
41 if (b4a.compare(A.publicKey, B.publicKey) > 0) { const t2 = A; A = B; B = t2 } // ensure A<B so B = higher key => member (opens writable=false)
42 const members = [
43 { key: b4a.toString(A.publicKey, 'hex'), role: 1 },
44 { key: b4a.toString(B.publicKey, 'hex'), role: 1 }
45 ]
46 const roomSeed = crypto.randomBytes(32); const roomKey = crypto.randomBytes(32); const encryptionKey = crypto.randomBytes(32)
47 const sb = new StateBaseV3(tmp(), roomKey, {
48 roomSeed, encryptionKey, network: false, // no swarm — isolated, so it can never converge (writable stays false)
49 selfKey: B.publicKey, attest: makeAttest(B, pb), verifyAttest: makeVerifyAttest(rootPub),
50 getMembers: async () => members
51 })
52 await sb.ready()
53 t.teardown(async () => { try { await sb.close() } catch {} })
54 t.absent(sb.base.writable, 'B opens writable=false (un-converged plain writer)')
55
56 // ── HARD-STALL base.update(): replace it with a promise that never resolves. Any code path that
57 // AWAITS base.update() directly (the pre-fix bug) now hangs forever; the fix must route through
58 // _updateBounded / snapshot reads so these calls still return.
59 const realUpdate = sb.base.update.bind(sb.base)
60 sb.base.update = () => new Promise(() => {}) // never resolves
61 t.teardown(() => { try { sb.base.update = realUpdate } catch {} })
62
63 // control: the explicitly non-blocking snapshot reads (already shipped) must be instant
64 await within(t, 'sb.getNow()', sb.getNow('nope'), 2000)
65 await within(t, 'sb.listNow()', sb.listNow(''), 2000)
66
67 // the sign-loop first read: Lease.current() → must be snapshot-based, not base.update()
68 const lease = new Lease(sb, { self: b4a.toString(B.publicKey, 'hex') })
69 await within(t, 'lease.current()', lease.current(), 2000)
70 await within(t, 'lease.holds()', lease.holds(), 2000)
71
72 // get-state's record read: RepoBase.listRecords({ now:true }) → snapshot, not base.update()
73 const repo = new RepoBase(sb, { did: 'did:plc:test' })
74 await within(t, 'repo.listRecords({now:true})', repo.listRecords(null, { now: true }), 2000)
75
76 // THE DOORBELL: requestWriter must (a) append the content-free optimistic admit block and
77 // (b) RETURN within its timeout even though base.update() is hung — so the sign loop keeps
78 // re-ringing instead of freezing. It will not become writable (isolated, no indexer), so it
79 // loops the full timeoutMs then returns.
80 const lenBefore = sb.base.local.length
81 const r = await within(t, 'requestWriter({timeoutMs:1500})', sb.requestWriter({ timeoutMs: 1500 }), 8000)
82 t.ok(r && r.writable === false, 'requestWriter returned writable=false (no indexer) — but returned, not hung')
83 t.ok(sb.base.local.length > lenBefore, `doorbell optimistic block was appended (local.length ${lenBefore} -> ${sb.base.local.length})`)
84
85 // ensureAdmitted (the boot re-admission trigger) must also return, and be guarded (2nd concurrent
86 // call is a no-op that returns immediately).
87 await within(t, 'ensureAdmitted()', sb.ensureAdmitted({ timeoutMs: 1200 }), 8000)
88})