OverviewFilesCommitsIssuesDocsReleasesPeers

atproto-base commits / 275afbc

atproto-base: indexer topology — keet-aligned start-indexing (N=1→N=3 quorum)
Audited keet-core 4.16 members/index.js IN FULL. start-indexing/stop-indexing now mirror keet's handlers I/P EXACTLY: a device SELF-promotes/demotes its own writer, gated on its member being an admin (role 1 = keet canAdmin). NO preferFastForward, NO length-0 gate — keet promotes ACTIVE data-bearing writers routinely (the earlier "fresh-writer only" belief was wrong; the ~25% retraction was setLocal ROTATION, a different op). stop-indexing adds a removeable() guard so it never strands the last indexer. Deliberately did NOT port StateBaseV3's move-indexer: that was a MISREAD. keet's move-indexer is a SAME-MEMBER device→device indexer handoff (all three member keys must equal the caller's); StateBaseV3 repurposed it as a cross-member creator demotion, which keet never does — and the deterministic projection makes it unnecessary anyway. Proven: admin self-promote N=1→N=2 (base keeps finalizing); N=1→N=3 then the base SURVIVES the creator going offline (2-of-3 quorum finalizes — the durability N=1 cannot give); non-admin denied. N=1-until-3-then-quorum is a policy on who has admin. test/indexer-topology.test.mjs (3 tests / 9 asserts). Suite now 15 tests / 63 asserts. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
275afbc08802101bb258b22d2bcb9841e322dfbf
by plankatron <fleeky@gmail.com> · 2026-07-23T18:50:04.000Z

Changed files (2) +151 -1

lib/base.js +63 -1
……@@ -23,7 +23,13 @@ import b4a from 'b4a'
2323import crypto from 'hypercore-crypto'
2424import { projectCommit, nextRev } from './project.js'
2525
26const OP = { SET_RECORD: 'set-record', DEL_RECORD: 'del-record', ADMIT: 'admit', SET_MEMBERS: 'set-members' }
26// Typed ops. start-indexing/stop-indexing mirror keet's members/index.js handlers I/P EXACTLY: a
27// device SELF-promotes/demotes its own writer, gated on its member being an admin (role 1 = keet
28// canAdmin). NO preferFastForward, NO length-0 gate (keet promotes ACTIVE data-bearing writers —
29// verified by reading keet-core 4.16 members/index.js in full). This is NOT StateBaseV3's move-indexer,
30// which was a MISREAD (it demoted the creator cross-member; keet's move-indexer is a same-member
31// device→device handoff — deliberately NOT ported).
32const OP = { SET_RECORD: 'set-record', DEL_RECORD: 'del-record', ADMIT: 'admit', SET_MEMBERS: 'set-members', START_INDEXING: 'start-indexing', STOP_INDEXING: 'stop-indexing' }
2733const MEMBERS = '__members__' // reserved view key: { identityHex: role }
2834const DEVICES = '__devices__' // reserved view key: { writerKeyHex: identityHex }
2935const RKEY = (collection, rkey) => `${collection}/${rkey}`
……@@ -109,6 +115,25 @@ export class AtprotoBase {
109115 continue
110116 }
111117
118 if (op.t === OP.START_INDEXING) {
119 // keet members/index.js `I`: a device self-promotes to indexer iff its member is an admin.
120 const fromKey = node.from && node.from.key
121 if (!fromKey) continue
122 if (await this._writerRole(view, fromKey.toString('hex')) !== 1) continue // canAdmin only
123 try { await host.addWriter(fromKey, { indexer: true }); await host.ackWriter(fromKey) } catch {}
124 continue
125 }
126
127 if (op.t === OP.STOP_INDEXING) {
128 // keet members/index.js `P`: self-demote (admin-gated). We add a removeable() guard so we never
129 // demote the last indexer and strand finalization (keet leaves this to policy).
130 const fromKey = node.from && node.from.key
131 if (!fromKey) continue
132 if (await this._writerRole(view, fromKey.toString('hex')) !== 1) continue
133 try { if (host.removeable(fromKey)) await host.addWriter(fromKey, { indexer: false }) } catch {}
134 continue
135 }
136
112137 if (op.t === OP.SET_RECORD) {
113138 const existing = await view.get(op.key)
114139 if (existing && existing.value && typeof existing.value.w === 'number' && existing.value.w > op.w) continue // LWW
……@@ -142,6 +167,17 @@ export class AtprotoBase {
142167 return role != null ? Number(role) : null
143168 }
144169
170 // Role of the member owning a given writer key (via __devices__ → __members__), for in-apply
171 // authorization of self-promote/demote (keet's memberFor(writer.key).status.canAdmin).
172 async _writerRole (view, writerHex) {
173 const drec = await view.get(DEVICES); const dev = (drec && drec.value) || {}
174 const idHex = dev[writerHex]
175 if (idHex == null) return null
176 const mrec = await view.get(MEMBERS); const members = (mrec && mrec.value) || {}
177 const role = members[idHex]
178 return role != null ? Number(role) : null
179 }
180
145181 // The creator/admin projects the roster into the shared view (memberKey→role) + its own device
146182 // mapping (writerKey→identity), so admission can resolve a joiner's role deterministically in apply.
147183 async projectMembers (members) {
……@@ -156,6 +192,32 @@ export class AtprotoBase {
156192 return true
157193 }
158194
195 // ── Indexer topology (keet-aligned: self-promote, canAdmin-gated) ────────────────────────────────
196 _isLocalIndexer () { try { return this.base.isIndexer === true } catch { return false } }
197
198 // Self-promote this device to an indexer (keet `@keet/start-indexing`). MUST already be an admitted
199 // writer (keet: an active device) — ensureWritable() first — and this member must be an admin (role 1)
200 // in the roster. A current indexer must be online to apply the promotion. Grows the indexer set toward
201 // quorum (N=1→N=3; keep it OFF the 2-rest by granting admin to the 2nd+3rd co-owners together, exactly
202 // as keet gates it via canAdmin). Data-bearing writers ARE promotable (keet does this routinely).
203 async requestIndexer ({ timeoutMs = 20000 } = {}) {
204 await this.ensureWritable()
205 if (this._isLocalIndexer()) return true
206 await this.base.append(b4a.from(JSON.stringify({ t: OP.START_INDEXING }))).catch(() => {})
207 await this._updateBounded()
208 const deadline = Date.now() + timeoutMs
209 while (!this._isLocalIndexer() && Date.now() < deadline) { await new Promise((r) => setTimeout(r, 150)); await this._updateBounded() }
210 return this._isLocalIndexer()
211 }
212
213 // Self-demote (keet `@keet/stop-indexing`); admin-gated + removeable-guarded in apply so it never
214 // strands the last indexer.
215 async stopIndexing () {
216 if (!this.base || !this.base.writable) return
217 await this.base.append(b4a.from(JSON.stringify({ t: OP.STOP_INDEXING })))
218 await this.base.update()
219 }
220
159221 // Block until this co-owner is a genuinely ADMITTED writer, then return. No-op if already writable.
160222 // Rings the CONTENT-FREE admission doorbell and waits; THROWS if no indexer admits us within the
161223 // timeout (never a silent content loss). This is what makes putRecord/delRecord durable.
test/indexer-topology.test.mjs +88 -0
……@@ -0,0 +1,88 @@
1// INDEXER TOPOLOGY (keet-aligned start-indexing). An ADMIN co-owner self-promotes to indexer
2// (@keet/start-indexing handler I), growing N=1→N=2; a NON-admin is denied. The base keeps
3// finalizing. (This is keet's self-promote-if-canAdmin — NOT StateBaseV3's cross-member
4// move-indexer, which was the misread.)
5
6import test from 'brittle'
7import { mkdtempSync, rmSync } from 'node:fs'
8import { tmpdir } from 'node:os'
9import { join } from 'node:path'
10import crypto from 'hypercore-crypto'
11import { deriveAtprotoKey, createAtprotoIdentity } from 'atproto-identity'
12import { readCar } from 'atproto-repo/lib/car.js'
13import { verifyRepo } from 'atproto-repo'
14import { AtprotoBase } from '../lib/base.js'
15
16const MNEMONIC = 'test test test test test test test test test test test junk'
17const ID = createAtprotoIdentity(MNEMONIC, { handle: 'x.7rnx.net', pdsUrl: 'https://pds.7rnx.net' })
18const KEY = deriveAtprotoKey(MNEMONIC)
19const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-07-23T00:00:00.000Z', langs: ['en'] })
20const hex = (b) => b.toString('hex')
21const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
22function mk (opts) { const dir = mkdtempSync(join(tmpdir(), 'idx-')); const base = new AtprotoBase({ path: dir, did: ID.did, privateKey: KEY.privateKey, ackInterval: 100, ...opts }); return { base, dir, close: async () => { await base.close(); rmSync(dir, { recursive: true, force: true }) } } }
23const pipe = (a, b) => { const s1 = a.base.base.replicate(true); const s2 = b.base.base.replicate(false); s1.pipe(s2).pipe(s1); return () => { try { s1.destroy() } catch {} ; try { s2.destroy() } catch {} } }
24async function settle (...ws) { for (let i = 0; i < 20; i++) { for (const w of ws) { try { await w.base.base.update() } catch {} } await sleep(60) } }
25const indexers = (bs) => { try { return bs.base.base.linearizer.indexers.length } catch { return -1 } }
26async function recordCount (car) { return (await verifyRepo(await readCar(car))).records }
27
28test('start-indexing: an ADMIN co-owner self-promotes to indexer (N=1→N=2); base keeps finalizing', async (t) => {
29 const cId = crypto.keyPair(); const aId = crypto.keyPair()
30 const roster = [{ key: hex(cId.publicKey), role: 1 }, { key: hex(aId.publicKey), role: 1 }] // both admins
31
32 const C = mk({ identity: cId, members: roster }); await C.base.ready()
33 const A = mk({ bootstrap: C.base.base.key, identity: aId }); await A.base.ready()
34 const unpipe = pipe(C, A)
35 t.teardown(async () => { unpipe(); await C.close(); await A.close() })
36
37 await C.base.projectMembers(roster); await settle(C, A)
38 t.is(indexers(C), 1, 'starts at N=1 (creator sole indexer)')
39
40 const ok = await A.base.requestIndexer({ timeoutMs: 8000 })
41 await settle(C, A)
42 t.ok(ok && A.base._isLocalIndexer(), 'admin co-owner became an indexer (self-promote, canAdmin)')
43 t.is(indexers(C), 2, 'indexer set grew to N=2')
44
45 await A.base.putRecord('app.bsky.feed.post', '3a', post('from A'))
46 await settle(C, A); await C.base.settled()
47 t.is(await recordCount((await C.base.head()).carBytes), 1, 'base still finalizes with 2 indexers online')
48})
49
50test('N=1→N=3 quorum: after growing to 3 indexers, the base survives the creator going OFFLINE', async (t) => {
51 const cId = crypto.keyPair(); const aId = crypto.keyPair(); const bId = crypto.keyPair()
52 const roster = [{ key: hex(cId.publicKey), role: 1 }, { key: hex(aId.publicKey), role: 1 }, { key: hex(bId.publicKey), role: 1 }] // 3 admins
53
54 const C = mk({ identity: cId, members: roster }); await C.base.ready()
55 const A = mk({ bootstrap: C.base.base.key, identity: aId }); await A.base.ready()
56 const B = mk({ bootstrap: C.base.base.key, identity: bId }); await B.base.ready()
57 // full mesh
58 let pCA = pipe(C, A); let pCB = pipe(C, B); const pAB = pipe(A, B)
59 t.teardown(async () => { try { pCA() } catch {} ; try { pCB() } catch {} ; try { pAB() } catch {} ; await C.close(); await A.close(); await B.close() })
60
61 await C.base.projectMembers(roster); await settle(C, A, B)
62 await A.base.requestIndexer({ timeoutMs: 8000 }); await settle(C, A, B) // N=2
63 await B.base.requestIndexer({ timeoutMs: 8000 }); await settle(C, A, B) // N=3
64 t.is(indexers(A), 3, 'grew N=1→N=3 (three admin indexers)')
65
66 // Creator goes OFFLINE. A + B (quorum 2 of 3) must still finalize.
67 pCA(); pCB()
68 await A.base.putRecord('app.bsky.feed.post', '3survive', post('written with creator offline'))
69 await settle(A, B); await A.base.settled()
70 t.is(await recordCount((await A.base.head()).carBytes), 1, 'A+B (2 of 3 quorum) finalized a write with the creator OFFLINE — the durability N=1 cannot provide')
71})
72
73test('start-indexing: a NON-admin is denied (stays a plain writer)', async (t) => {
74 const cId = crypto.keyPair(); const nId = crypto.keyPair()
75 const roster = [{ key: hex(cId.publicKey), role: 1 }, { key: hex(nId.publicKey), role: 3 }] // n = non-admin
76
77 const C = mk({ identity: cId, members: roster }); await C.base.ready()
78 const N = mk({ bootstrap: C.base.base.key, identity: nId }); await N.base.ready()
79 const unpipe = pipe(C, N)
80 t.teardown(async () => { unpipe(); await C.close(); await N.close() })
81
82 await C.base.projectMembers(roster); await settle(C, N)
83 const ok = await N.base.requestIndexer({ timeoutMs: 3000 })
84 await settle(C, N)
85 t.absent(ok, 'requestIndexer did not succeed for a non-admin')
86 t.absent(N.base._isLocalIndexer(), 'non-admin never became an indexer')
87 t.is(indexers(C), 1, 'indexer set stayed at N=1')
88})