OverviewFilesCommitsIssuesDocsReleasesPeers

atproto-base commits / e058166

atproto-base: finalized-view checkout — project only the durably-signed view
indexedRecords()/head() now read the view Hyperbee CHECKED OUT at the VIEW CORE's signedLength (base.view.core.signedLength — NOT base.signedLength, which is the system/oplog core; verified by probe against autobase 7.28.1). So an un-finalized optimistic tip — a lagging co-owner's writes, or writes during an N=2 stall — NEVER leaks into the signed atproto chain (engine-level reorg-safety). The atproto chain intentionally advances on FINALIZATION, not on write (durability); added settled() to await that boundary + made ackInterval configurable. New proof finalized-view.test.mjs: a plain writer's un-acked write is in its optimistic view but ABSENT from the signed head until an indexer finalizes it — then it appears (stall resolved, never forked). Suite now 12 tests / 54 asserts, all green. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
e058166eaeef64f34afa92596074d8e9f97df9f4
by plankatron <fleeky@gmail.com> · 2026-07-23T17:36:49.000Z

Changed files (4) +100 -14

lib/base.js +31 -7
……@@ -45,6 +45,7 @@ export class AtprotoBase {
4545 this._encryptionKey = opts.encryptionKey || null
4646 this._identity = opts.identity ? { publicKey: toBuf(opts.identity.publicKey), secretKey: toBuf(opts.identity.secretKey) } : null
4747 this._members = opts.members || null
48 this._ackInterval = opts.ackInterval || 1000
4849 this._admitted = new Set()
4950 this.store = null
5051 this.base = null
……@@ -61,7 +62,7 @@ export class AtprotoBase {
6162 close: async (view) => { if (view) await view.close() },
6263 valueEncoding: 'binary',
6364 optimistic: true,
64 ackInterval: 1000,
65 ackInterval: this._ackInterval,
6566 ackThreshold: 4
6667 }
6768 if (this._keyPair) baseOpts.keyPair = this._keyPair
……@@ -200,18 +201,41 @@ export class AtprotoBase {
200201 await this.base.update()
201202 }
202203
203 // The INDEXED (finalized) atproto record set — the input to the projection. Single-writer creator:
204 // full view == indexed. Multi-writer indexed-view checkout (read at base.indexedLength) is the next slice.
204 // The FINALIZED (durably multisig-signed) atproto record set — the input to the projection. Reads the
205 // view Hyperbee CHECKED OUT at the VIEW CORE's signedLength, so an un-finalized optimistic tip (a
206 // lagging co-owner's writes, or writes during an N=2 stall) NEVER leaks into the signed atproto chain
207 // (engine-level reorg-safety; see DESIGN §hard-requirements). NB: base.view.core.signedLength — NOT
208 // base.signedLength, which is the SYSTEM/oplog core, a different core (verified against autobase 7.28.1).
205209 async indexedRecords () {
206210 await this.base.update()
211 const signed = (this.base.view && this.base.view.core) ? this.base.view.core.signedLength : 0
212 if (!signed) return [] // nothing finalized yet (also avoids Hyperbee checkout(0) → full-view footgun)
213 const co = this.base.view.checkout(signed)
207214 const out = []
208 for await (const { key, value } of this.base.view.createReadStream()) {
209 if (String(key).startsWith('__')) continue // reserved governance keys
210 out.push({ key: String(key), value: value.record })
211 }
215 try {
216 for await (const { key, value } of co.createReadStream()) {
217 if (String(key).startsWith('__')) continue // reserved governance keys
218 out.push({ key: String(key), value: value.record })
219 }
220 } finally { try { await co.close() } catch {} }
212221 return out
213222 }
214223
224 // Drive finalization until the view core's signedLength stabilizes — so a caller can await the point
225 // where recent writes are durably finalized (and thus reflected in indexedRecords()/head()). The
226 // atproto chain intentionally advances on FINALIZATION, not on write (durability), so this lag is
227 // by-design. Bounded; returns once signedLength holds steady (or the timeout hits, e.g. an N=2 stall).
228 async settled ({ timeoutMs = 5000 } = {}) {
229 const deadline = Date.now() + timeoutMs
230 let last = -1; let stable = 0
231 while (Date.now() < deadline) {
232 try { await this.base.update() } catch {}
233 const s = (this.base.view && this.base.view.core) ? this.base.view.core.signedLength : 0
234 if (s === last) { if (++stable >= 3) return } else { stable = 0; last = s }
235 await new Promise((r) => setTimeout(r, 40))
236 }
237 }
238
215239 async head ({ prevRev = null, prev = null } = {}) {
216240 const records = await this.indexedRecords()
217241 const rev = nextRev(prevRev)
test/base.test.mjs +7 -6
……@@ -18,9 +18,10 @@ const KEY = deriveAtprotoKey(MNEMONIC)
1818
1919function mkBase () {
2020 const dir = mkdtempSync(join(tmpdir(), 'atbase-'))
21 const base = new AtprotoBase({ path: dir, did: ID.did, privateKey: KEY.privateKey })
21 const base = new AtprotoBase({ path: dir, did: ID.did, privateKey: KEY.privateKey, ackInterval: 100 })
2222 return { base, dir, close: async () => { await base.close(); rmSync(dir, { recursive: true, force: true }) } }
2323}
24const head = async (bs) => { await bs.base.settled(); return bs.base.head() } // finalized head
2425const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-07-23T00:00:00.000Z', langs: ['en'] })
2526
2627test('AtprotoBase: typed record writes → relay-valid, deterministic, advancing atproto chain', async (t) => {
……@@ -31,7 +32,7 @@ test('AtprotoBase: typed record writes → relay-valid, deterministic, advancing
3132 await A.base.putRecord('app.bsky.feed.post', '3aaa', post('one'))
3233 await A.base.putRecord('app.bsky.feed.post', '3bbb', post('two'))
3334
34 const h1 = await A.base.head()
35 const h1 = await head(A)
3536 const v = await verifyRepo(await readCar(h1.carBytes))
3637 t.ok(v.ok, `head CAR is relay-valid (verifyRepo), records=${v.records}`)
3738 t.ok(verifyCommit(h1.signedCommit, KEY.publicKey), 'head commit verifies against #atproto key')
……@@ -39,13 +40,13 @@ test('AtprotoBase: typed record writes → relay-valid, deterministic, advancing
3940
4041 // Advances on a new write.
4142 await A.base.putRecord('app.bsky.feed.post', '3ccc', post('three'))
42 const h2 = await A.base.head()
43 const h2 = await head(A)
4344 t.not(h1.commitCid.toString(), h2.commitCid.toString(), 'head advances when a record is added')
4445 t.is((await verifyRepo(await readCar(h2.carBytes))).records, 3, 'now 3 records')
4546
4647 // Delete is reflected.
4748 await A.base.delRecord('app.bsky.feed.post', '3aaa')
48 const h3 = await A.base.head()
49 const h3 = await head(A)
4950 t.is((await verifyRepo(await readCar(h3.carBytes))).records, 2, 'record deleted → 2 records in projection')
5051})
5152
……@@ -60,8 +61,8 @@ test('AtprotoBase: two independent bases with the same records project the IDENT
6061 await B.base.putRecord('app.bsky.feed.post', '3bbb', post('two'))
6162 await B.base.putRecord('app.bsky.feed.post', '3aaa', post('one'))
6263
63 const ha = await A.base.head()
64 const hb = await B.base.head()
64 const ha = await head(A)
65 const hb = await head(B)
6566 t.is(ha.commitCid.toString(), hb.commitCid.toString(), 'identical head commit CID across co-owners')
6667 t.is(Buffer.compare(Buffer.from(ha.signedCommit.sig), Buffer.from(hb.signedCommit.sig)), 0, 'identical signature (no signer election needed)')
6768})
test/finalized-view.test.mjs +59 -0
……@@ -0,0 +1,59 @@
1// ENGINE-LEVEL REORG-SAFETY — head()/indexedRecords() project ONLY the FINALIZED view (checked out
2// at the view core's signedLength), so an un-finalized optimistic write NEVER leaks into the signed
3// atproto chain. Proof: a plain writer, disconnected from the sole indexer, writes a record — it lands
4// in the writer's optimistic view but is ABSENT from its finalized head, until an indexer finalizes it.
5// (This is the head-level counterpart of proof #3/#5; it guarantees the projection can't sign a tip
6// that later reorgs, and that an N=2 stall simply omits the un-finalized write rather than forking.)
7
8import test from 'brittle'
9import { mkdtempSync, rmSync } from 'node:fs'
10import { tmpdir } from 'node:os'
11import { join } from 'node:path'
12import crypto from 'hypercore-crypto'
13import { deriveAtprotoKey, createAtprotoIdentity } from 'atproto-identity'
14import { readCar } from 'atproto-repo/lib/car.js'
15import { verifyRepo } from 'atproto-repo'
16import { AtprotoBase } from '../lib/base.js'
17
18const MNEMONIC = 'test test test test test test test test test test test junk'
19const ID = createAtprotoIdentity(MNEMONIC, { handle: 'x.7rnx.net', pdsUrl: 'https://pds.7rnx.net' })
20const KEY = deriveAtprotoKey(MNEMONIC)
21const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-07-23T00:00:00.000Z', langs: ['en'] })
22const hex = (b) => b.toString('hex')
23const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
24function mk (opts) { const dir = mkdtempSync(join(tmpdir(), 'fv-')); 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 }) } } }
25const 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 {} } }
26async function settle (...ws) { for (let i = 0; i < 15; i++) { for (const w of ws) { try { await w.base.base.update() } catch {} } await sleep(60) } }
27async function fullViewCount (bs) { let n = 0; for await (const { key } of bs.base.base.view.createReadStream()) { if (!String(key).startsWith('__')) n++ } return n }
28async function recordCount (car) { return (await verifyRepo(await readCar(car))).records }
29
30test('finalized-view: an un-finalized write is in the optimistic view but NEVER in the signed head', async (t) => {
31 const creatorId = crypto.keyPair()
32 const writerId = crypto.keyPair()
33 const roster = [{ key: hex(creatorId.publicKey), role: 1 }, { key: hex(writerId.publicKey), role: 3 }]
34
35 const C = mk({ identity: creatorId, members: roster }) // creator = sole indexer
36 await C.base.ready()
37 const P = mk({ bootstrap: C.base.base.key, identity: writerId }) // P = plain writer
38 await P.base.ready()
39 let unpipe = pipe(C, P)
40 t.teardown(async () => { try { unpipe() } catch {} ; await C.close(); await P.close() })
41
42 await C.base.projectMembers(roster); await settle(C, P)
43 await P.base.putRecord('app.bsky.feed.post', '3one', post('finalized')) // admitted + written
44 await settle(C, P); await P.base.settled()
45 t.is(await recordCount((await P.base.head()).carBytes), 1, 'first write is finalized and in the signed head')
46
47 // Disconnect P from the sole indexer; P writes again — this write CANNOT finalize.
48 unpipe()
49 await P.base.putRecord('app.bsky.feed.post', '3two', post('un-finalized'))
50 await P.base.settled({ timeoutMs: 1500 }) // will not advance — nothing to finalize it
51
52 t.is(await fullViewCount(P), 2, "the un-finalized write IS in P's optimistic view (durable-local)")
53 t.is(await recordCount((await P.base.head()).carBytes), 1, 'but the un-finalized write is ABSENT from the finalized signed head (no tip leaks into the chain)')
54
55 // Reconnect → the sole indexer finalizes it → it now appears in the head (no fork, just delayed).
56 unpipe = pipe(C, P)
57 await settle(C, P); await P.base.settled()
58 t.is(await recordCount((await P.base.head()).carBytes), 2, 'after reconnect the indexer finalizes it → now in the signed head (stall resolved, never forked)')
59})
test/governance.test.mjs +3 -1
……@@ -20,7 +20,7 @@ const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
2020
2121function mk (opts) {
2222 const dir = mkdtempSync(join(tmpdir(), 'gov-'))
23 const base = new AtprotoBase({ path: dir, did: ID.did, privateKey: KEY.privateKey, ...opts })
23 const base = new AtprotoBase({ path: dir, did: ID.did, privateKey: KEY.privateKey, ackInterval: 100, ...opts })
2424 return { base, dir, close: async () => { await base.close(); rmSync(dir, { recursive: true, force: true }) } }
2525}
2626const 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 {} } }
……@@ -46,9 +46,11 @@ test('governance: a member joiner is admitted (block-until-admitted) → writes
4646 t.ok(J.base.writable, 'joiner was admitted as a writer (block-until-admitted)')
4747
4848 await settle(C, J)
49 await C.base.settled()
4950 const recsC = await C.base.indexedRecords()
5051 t.ok(recsC.find((r) => r.key.endsWith('3joiner')), "joiner's record replicated into the creator's shared view")
5152
53 await C.base.settled(); await J.base.settled()
5254 const hC = await C.base.head()
5355 const hJ = await J.base.head()
5456 t.is(hC.commitCid.toString(), hJ.commitCid.toString(), 'identical projected head across co-owners after admission (deterministic, no signer election)')