OverviewFilesCommitsIssuesDocsReleasesPeers

atproto-did-peer commits / 9436095

0.2.15: label cores — open, seed and re-announce a labeler DID's labels
Member labels, hyper-social #80 step 3. hpp 0.3.128 announces a labeler DID's membrane-labels core as labelKey/labelLength. A DidPeer now: - opens and fully seeds it (provide({labelKey}), or adoptLabelKey when the key first appears on an already-open front; the router does both); - follows its appends and re-announces on each; - carries labelKey/labelLength in its own announce (one _announcement() for both paths), so a re-announcing seeder never hides the labels. Verification against the DID key is the consumer's job, as for commits. test/labels.test.mjs (testnet, router reader); full suite green. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
9436095e5357b3eb0efdfef697a5a866345c90fe
by plankatron <fleeky@gmail.com> · 2026-10-14T02:09:06.000Z

Changed files (3) +138 -18

lib/did-peer.js +48 -16
……@@ -75,6 +75,11 @@ export class DidPeer {
7575 this._rpcs = []
7676 this._repoKey = null
7777 this._firehoseKey = null
78 // A LABELER DID's signed label core (hpp 0.3.128 membrane-labels:<did>), when its holder announces one.
79 // Plain hypercore of encoded com.atproto.label blocks; verifying each against the DID key is the
80 // consumer's job (the gateway lens), exactly as it verifies commits.
81 this._labelKey = null
82 this.labels = null
7883 this._channels = new Set()
7984 this.head = null
8085 this._opening = null
……@@ -134,7 +139,28 @@ export class DidPeer {
134139
135140 // This node HOLDS the cores (a signing device, a relay that already seeded, or a bootstrap). Open
136141 // them so we can serve + announce. Idempotent.
137 async provide ({ repoKey, firehoseKey } = {}) { return this._openCores(repoKey, firehoseKey) }
142 async provide ({ repoKey, firehoseKey, labelKey = null } = {}) {
143 await this._openCores(repoKey, firehoseKey)
144 if (labelKey) await this.adoptLabelKey(labelKey)
145 }
146
147 // Open + fully seed a labeler's label core, follow its appends, and re-announce on each (a new label is
148 // news to downstream readers just like a new commit). Like adoptRepoKey: only on an established front,
149 // once. → true when it opened the core.
150 async adoptLabelKey (labelKey) {
151 if (this.labels || !labelKey || !this.seq || this._opening || this._labelOpening) return false
152 this._labelOpening = true
153 try {
154 this._labelKey = toKey(labelKey)
155 const core = this.corestore.get({ key: this._labelKey })
156 await core.ready()
157 try { core.download({ start: 0, end: -1 }) } catch {} // ONE live range: all + future appends
158 core.on('append', () => this._announceAll())
159 this.labels = core
160 this._announceAll()
161 return true
162 } finally { this._labelOpening = false }
163 }
138164
139165 // Re-open a repo core for a front that went back to PROJECTING after a stateless spell. The router
140166 // drops the replica when a peer announces repoKey:null; without this the drop is ONE-WAY, so a peer
……@@ -217,29 +243,32 @@ export class DidPeer {
217243 announceVia (message) {
218244 if (!this.seq || !message) return
219245 try {
220 message.send({
221 did: this.did,
222 repoKey: this._repoKey ? b4a.toString(this._repoKey, 'hex') : null,
223 firehoseKey: b4a.toString(this._firehoseKey, 'hex'),
224 repoLength: this.repo ? this.repo.core.length : null,
225 firehoseLength: this.seq.core.length
226 })
246 message.send(this._announcement())
227247 } catch (e) { this._fail(e) }
228248 }
229249
230250 _announceTo (chan) {
231251 if (!this.seq || !chan._announce) return
232252 try {
233 chan._announce.send({
234 did: this.did,
235 repoKey: this._repoKey ? b4a.toString(this._repoKey, 'hex') : null,
236 firehoseKey: b4a.toString(this._firehoseKey, 'hex'),
237 repoLength: this.repo ? this.repo.core.length : null,
238 firehoseLength: this.seq.core.length
239 })
253 chan._announce.send(this._announcement())
240254 } catch (e) { this._fail(e) }
241255 }
242256
257 // One payload for both announce paths. labelKey/labelLength are carried by EVERY announcer that holds
258 // the core — a re-announcing seeder that dropped them would hide the labels from everyone fronting
259 // through it. null when there is no label core (the field exists, so a consumer can branch).
260 _announcement () {
261 return {
262 did: this.did,
263 repoKey: this._repoKey ? b4a.toString(this._repoKey, 'hex') : null,
264 firehoseKey: b4a.toString(this._firehoseKey, 'hex'),
265 repoLength: this.repo ? this.repo.core.length : null,
266 firehoseLength: this.seq.core.length,
267 labelKey: this.labels ? b4a.toString(this._labelKey, 'hex') : null,
268 labelLength: this.labels ? this.labels.length : null
269 }
270 }
271
243272 _announceAll () { for (const chan of this._channels) this._announceTo(chan) }
244273
245274 async onAnnounce (a) {
……@@ -251,6 +280,7 @@ export class DidPeer {
251280 await this._openCores(a.repoKey, a.firehoseKey)
252281 this._announceAll() // we can help seed now — tell peers we have them too (symmetric)
253282 }
283 if (a.labelKey && !this.labels) await this.adoptLabelKey(a.labelKey)
254284 // The announce is a HINT (peer says its firehose is at length N). Replication pushes the blocks;
255285 // we react to the 'append' event to reconcile. One-shot update() nudges the fetch (event-triggered
256286 // on announce, NOT a poll loop).
……@@ -528,6 +558,7 @@ export class DidPeer {
528558 this._channels.clear()
529559 if (this.seq) { try { await this.seq.close() } catch {} }
530560 if (this.repo) { try { await this.repo.close() } catch {} }
561 if (this.labels) { try { await this.labels.close() } catch {} }
531562 }
532563}
533564
……@@ -719,6 +750,7 @@ export class DidPeerRouter {
719750 // above is one-way and a rolled-back flip never heals.
720751 existing.adoptRepoKey(a.repoKey).catch(() => {})
721752 }
753 if (a.labelKey && !existing.labels) existing.adoptLabelKey(a.labelKey).catch(() => {})
722754 if (existing.seq) existing.seq.core.update().catch(() => {})
723755 this._wake(did)
724756 return
……@@ -765,7 +797,7 @@ export class DidPeerRouter {
765797 if (entry.coords.handle) peer.handle = entry.coords.handle
766798 // EVERY peer announcing this (did, firehoseKey) is a candidate; _call tries them until one answers.
767799 if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r)
768 await peer.provide({ repoKey: entry.coords.repoKey, firehoseKey: entry.coords.firehoseKey })
800 await peer.provide({ repoKey: entry.coords.repoKey, firehoseKey: entry.coords.firehoseKey, labelKey: entry.coords.labelKey || null })
769801 // Re-sync: more peers may have announced this same (did, firehoseKey) during the open above.
770802 if (this._useRpc) for (const r of entry.conns.values()) if (r) peer.addRpc(r)
771803 if (!this._fronts.has(did)) this._fronts.set(did, new Map())
package.json +2 -2
……@@ -1,6 +1,6 @@
11{
22 "name": "atproto-did-peer",
3 "version": "0.2.14",
3 "version": "0.2.15",
44 "description": "Keyless, direction-free P2P reader/seeder for an atproto DID's repo+firehose Hypercores: announce-push discovery + event-driven head reconciliation + a keyless read surface. The shared primitive beneath atproto-bridge, atproto-relay and favela (extracted from atproto-bridge did-peer.js).",
55 "type": "module",
66 "main": "index.js",
……@@ -8,7 +8,7 @@
88 ".": "./index.js"
99 },
1010 "scripts": {
11 "test": "node test/commitops.test.mjs && node test/router.test.mjs && node test/co-owned.test.mjs && node test/coowner-set.test.mjs && node test/evict.test.mjs && node test/fork.test.mjs && node test/rpc-candidates.test.mjs && node test/no-repo.test.mjs && node test/get-repo-since.test.mjs && node test/rpc-timeout.test.mjs && node test/emit-sync.test.mjs",
11 "test": "node test/commitops.test.mjs && node test/router.test.mjs && node test/co-owned.test.mjs && node test/coowner-set.test.mjs && node test/evict.test.mjs && node test/fork.test.mjs && node test/rpc-candidates.test.mjs && node test/no-repo.test.mjs && node test/get-repo-since.test.mjs && node test/rpc-timeout.test.mjs && node test/emit-sync.test.mjs && node test/labels.test.mjs",
1212 "test:live": "node test/read-live.mjs"
1313 },
1414 "dependencies": {
test/labels.test.mjs +88 -0
……@@ -0,0 +1,88 @@
1// Label cores: a labeler DID's signed labels reach a keyless reader through the announce (member labels,
2// hyper-social #80 step 3).
3//
4// hpp (0.3.128) projects a labeler DID's signed label log into a membrane-labels core and announces
5// labelKey/labelLength beside the firehose. A reader — the gateway's lens, behind a DidPeerRouter — must
6// open and fully seed that core, follow its appends, and keep the field when IT re-announces (a seeder
7// that dropped labelKey would hide the labels from everyone fronting through it). A labelKey that first
8// appears on an already-open front is adopted, like a repoKey.
9//
10// Run: node test/labels.test.mjs
11import { mkdtempSync, rmSync } from 'node:fs'
12import { join } from 'node:path'
13import { tmpdir } from 'node:os'
14import createTestnet from '@hyperswarm/testnet'
15import Hyperswarm from 'hyperswarm'
16import b4a from 'b4a'
17import { RepoStore } from 'atproto-store/lib/repo-store.js'
18import { RepoManager } from 'atproto-store/lib/repo-manager.js'
19import { FirehoseSequencer } from 'atproto-store/lib/sequencer.js'
20import { openCorestore } from 'atproto-store/lib/corestore.js'
21import { createAtprotoIdentityFromKey } from 'atproto-identity/lib/atproto-identity.js'
22import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
23import { DidPeer, DidPeerRouter } from '../index.js'
24
25const assert = (c, m) => { if (!c) { console.error('FAIL:', m); process.exit(1) } console.log(' ok -', m) }
26const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms))
27const waitFor = async (fn, ms = 20000) => { const end = Date.now() + ms; while (Date.now() < end) { if (await fn()) return true; await sleep(100) } return false }
28const iso = '2026-10-13T00:00:00.000Z'
29const id = createAtprotoIdentityFromKey(new Uint8Array(32).fill(6), { handle: 'labeler.test', pdsUrl: 'https://pds.local' })
30const testnet = await createTestnet(3)
31const dirs = []
32const dir = (p) => { const d = mkdtempSync(join(tmpdir(), p)); dirs.push(d); return d }
33
34// The holder: a signed repo + firehose, and a label core holding two blocks.
35const hStore = openCorestore(dir('labels-holder-'))
36const repo = await new RepoStore(hStore, { name: `atproto-repo:${id.did}` }).ready()
37let micros = 1_700_000_000_000_000n
38const mgr = await new RepoManager(repo, { did: id.did, privateKey: id.privateKey, clock: () => (micros += 1_000_000n) }).init()
39const seq = await new FirehoseSequencer(hStore.get({ name: 'firehose' })).ready()
40await seq.sequenceCommit(await mgr.applyWrite({ action: 'create', collection: 'app.bsky.feed.post', rkey: '1', record: { $type: 'app.bsky.feed.post', text: 'hi', createdAt: iso } }), { time: iso })
41const labels = hStore.get({ name: 'membrane-labels' })
42await labels.ready()
43await labels.append([b4a.from('label-0'), b4a.from('label-1')])
44
45const holder = new DidPeer(hStore, { did: id.did, verifyKey: id.publicKey })
46await holder.provide({ repoKey: repo.core.key, firehoseKey: seq.core.key, labelKey: labels.key })
47assert(holder.labels && holder.labels.length === 2, 'provide opens the holder\'s label core')
48const hSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
49hSwarm.on('connection', (c) => holder.attach(c))
50await hSwarm.join(discoveryTopic(id.did), { server: true, client: true }).flushed()
51
52console.log('# a DidPeerRouter reader opens, seeds and follows the label core')
53const rStore = openCorestore(dir('labels-reader-'))
54const router = new DidPeerRouter(rStore)
55const rSwarm = new Hyperswarm({ bootstrap: testnet.bootstrap })
56rSwarm.on('connection', (c) => router.onconnection(c))
57router.add(id.did, { verifyKey: id.publicKey, noRepo: true })
58await rSwarm.join(discoveryTopic(id.did), { server: true, client: true }).flushed()
59const front = await router.front(id.did, { timeout: 20000 })
60assert(!!front, 'the reader fronts the DID')
61assert(await waitFor(() => front.labels && front.labels.contiguousLength === 2), 'the front holds both labels (fully downloaded)')
62assert(b4a.equals(front.labels.key, labels.key), 'the announced core, by key')
63await labels.append(b4a.from('label-2'))
64assert(await waitFor(() => front.labels.contiguousLength === 3), 'and follows a new label')
65
66console.log('\n# a re-announcing seeder keeps the field')
67let sent = null
68front.announceVia({ send: (m) => { sent = m } })
69assert(sent && sent.labelKey === b4a.toString(labels.key, 'hex'), 'the reader\'s own announce carries labelKey')
70assert(sent.labelLength === 3, 'and labelLength')
71
72console.log('\n# a labelKey that first appears on an open front is adopted')
73const late = new DidPeer(openCorestore(dir('labels-late-')), { did: id.did, verifyKey: id.publicKey, noRepo: true })
74await late.provide({ firehoseKey: seq.core.key })
75assert(late.labels === null, 'no labelKey, no label core')
76let none = null
77late.announceVia({ send: (m) => { none = m } })
78assert(none.labelKey === null && none.labelLength === null, 'announced as null (the field exists)')
79assert(await late.adoptLabelKey(labels.key) === true, 'adoptLabelKey opens it')
80assert(late.labels && b4a.equals(late.labels.key, labels.key), 'bound to the announced key')
81assert(await late.adoptLabelKey(labels.key) === false, 'idempotent')
82await late.close()
83
84console.log('\nLABELS OK')
85await rSwarm.destroy(); await router.close(); await rStore.close()
86await hSwarm.destroy(); await holder.close(); await seq.close(); await repo.close(); await hStore.close()
87await testnet.destroy()
88for (const d of dirs) rmSync(d, { recursive: true, force: true })