OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-proto-peer commits / 42c4121

0.3.127: heals fit the relay's op limit; a tied fork re-anchors to the chain head
1. Split heals. A heal over more than 200 touched paths was one commit the relay drops (indigo MaxCommitOps). sign() now restates the window in slices within the budget left by ordinary changes (every changed path costs an op, restated or not), sorted so co-owners pick the same slice; the last slice asserts the window (atproto-room 0.2.25). maxCommitOps is a constructor option (default 200) so tests can force splitting. test/heal-op-limit: 250-record fork window -> 2 commits, largest 200 ops, both forks' consumers converge (red on 0.3.126: one 251-op commit). 2. Equal-rev tie. sign() re-anchored only to a strictly HIGHER network head, so the device whose fork lost commitLine's tie-break kept extending its own branch: siblings at every rev, each a new fork. Found by the split- heal fuzz (a 5-op cap livelocked: every round re-forked and reopened the heal). Now it also re-anchors on an equal rev with a different commit. test/partition-fork-heal: red without the fix. Verified: multi-partition-relay with 200- and 5-op caps, 150 seeds x 24 steps; autobee fuzz seeds 1-10; full suite 168/168. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
42c41215a4a72f6a65feb5b613a27af3dcb67a0b
by plankatron <fleeky@gmail.com> · 2026-10-13T16:12:33.000Z

Changed files (5) +187 -13

lib/swarm-device.js +44 -6
……@@ -33,6 +33,8 @@ import { encodeReceipt, receiptEpoch } from 'atproto-room/lib/membership.js'
3333// every device; frames off it are forks that the next commit heals.
3434import { commitLine, frameKey, healPlan, CHAIN_MARK } from 'atproto-room/lib/commit-line.js'
3535// The stand-in old value that turns a restated path into an `update` op. Never a real record's CID.
36// indigo verify.go MaxCommitOps — a #commit with more ops is an error in strict mode (the default).
37const MAX_COMMIT_OPS = 200
3638const HEAL_PLACEHOLDER = cidForBytesSync(dagCbor.encode({ $type: 'hyper-proto-peer.healPlaceholder' })).toString()
3739
3840// StateBaseV3's durable banned-identity set + minimum-epoch receipt (see state-base.js).
……@@ -75,6 +77,9 @@ function serializeEntry (write, seq) {
7577 // A heal: the fork commits it knew (heals) and the fork-point rev from which it restated every touched
7678 // path (asserts). atproto-room healPlan decides from these whether a heal can have been skipped.
7779 ...(write.asserts !== undefined ? { heals: write.heals || [], asserts: write.asserts } : {}),
80 ...(write.partsFrom ? { partsFrom: write.partsFrom } : {}),
81 // A PARTIAL heal: restated a slice of the window that starts at this fork-point rev.
82 ...(write.restates !== undefined ? { restates: write.restates } : {}),
7883 commit: hex(write.commitBlock.bytes),
7984 ops: write.ops.map((o) => ({ action: o.action, path: o.path, cid: o.cid ? o.cid.toString() : null, prev: o.prev ? o.prev.toString() : null })),
8085 blocks: write.newBlocks.map((b) => ({ cid: b.cid.toString(), bytes: hex(b.bytes) }))
……@@ -82,7 +87,7 @@ function serializeEntry (write, seq) {
8287}
8388
8489export class SwarmDevice {
85 constructor (statebase, { did, privateKey, self, ttlMs = 15000, clock, firehose = false, recordStore = null } = {}) {
90 constructor (statebase, { did, privateKey, self, ttlMs = 15000, clock, firehose = false, recordStore = null, maxCommitOps = MAX_COMMIT_OPS } = {}) {
8691 if (!did || !privateKey || !self) throw new Error('SwarmDevice: did, privateKey, self required')
8792 this.sb = statebase
8893 this.did = did
……@@ -102,6 +107,7 @@ export class SwarmDevice {
102107 this.lease = new Lease(statebase, { self, ttlMs })
103108 this._clock = clock // optional injectable micros clock for deterministic revs in tests
104109 this.firehose = firehose // when true, sign() persists the signed #commit into __commit__/<rev>
110 this._maxCommitOps = maxCommitOps // the relay's per-commit op limit; a heal is split to fit it (tests lower it)
105111 this._lastRecords = [] // record set at the last sign — the `before` side of the #commit diff
106112 this.head = null // { rev, commitCid, dataCid, signedCommit } — the last signed commit
107113 }
……@@ -437,7 +443,11 @@ export class SwarmDevice {
437443 // our external (relay) projection. Non-blocking (listNow); `force` (bootstrap/reconcile/tests
438444 // that own head deliberately) bypasses, matching the anti-wipe guard convention above.
439445 if (!force && this.firehose && net) { // frames/net computed once at the top (shared with the election)
440 if (!this.head || net.rev > this.head.rev) {
446 // ⛔ ALSO on an equal rev with a different commit: commitLine broke the tie the other way, so our head
447 // is a fork. Re-anchoring only on a strictly higher rev left a device on the losing side of a tie
448 // extending its own branch forever — sibling commits at every rev, each one a new fork that reopened
449 // the heal (test/multi-partition-relay, split heals).
450 if (!this.head || net.rev > this.head.rev || (net.rev === this.head.rev && String(net.commitCid) !== this.head.commitCid.toString())) {
441451 // (1) adopt the network-latest head so we continue FORWARD from it, never from a stale head.
442452 const signedCommit = dagCbor.decode(b4a.from(net.commit, 'hex'))
443453 this.head = { rev: net.rev, commitCid: CID.parse(net.commitCid), dataCid: signedCommit.data, signedCommit }
……@@ -505,8 +515,13 @@ export class SwarmDevice {
505515 // the asserting state built by _healBefore.
506516 if (this.firehose) {
507517 const since = this.head ? this.head.rev : null
508 const write = buildCommitWrite(plan ? this._healBefore(plan, records) : this._lastRecords, records, { did: this.did, rev, prev, since, privateKey: this.privateKey })
509 if (plan) { write.heals = plan.heals; write.asserts = plan.from }
518 const slice = plan ? this._healSlice(plan, records) : null
519 const write = buildCommitWrite(slice ? this._healBefore(slice.paths, records) : this._lastRecords, records, { did: this.did, rev, prev, since, privateKey: this.privateKey })
520 if (slice && slice.final) {
521 write.heals = plan.heals
522 write.asserts = plan.from
523 if (plan.partsFrom) write.partsFrom = plan.partsFrom
524 } else if (slice) write.restates = plan.from
510525 // Only append a #commit frame when the diff is non-empty. A 0-op frame can only arise from the
511526 // rare restart-with-unsigned-pending-write edge (baseline == current yet head.dataCid differs);
512527 // emitting it would be a malformed commit. head still advances below so we don't re-trigger.
……@@ -546,12 +561,35 @@ export class SwarmDevice {
546561 // Every other path keeps the chain's previous value, so ordinary changes diff as usual. buildCommitWrite
547562 // still proves the ops invert to this state's root; only `prevData` names a state no consumer held,
548563 // which the relay logs and accepts (indigo verify.go:140-143, warn-only even in strict mode).
549 _healBefore (plan, records) {
564 // Which paths THIS commit restates. indigo drops a #commit with more than MAX_COMMIT_OPS ops, so the ops
565 // this commit carries anyway (ordinary changes since the last commit) come out of the budget and the
566 // rest of the window waits for the next commit, in sorted order so every co-owner picks the same slice.
567 // `final` when the remainder fits: that commit asserts the window. An empty remainder (partials already
568 // restated everything, the final never landed) still restates one path, so the final has an op and is
569 // appended — a 0-op commit is never written, and the window would stay open forever.
570 _healSlice (plan, records) {
571 const now = new Map(records.map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
572 const was = new Map(this._lastRecords.map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
573 // Every path that changed since the last commit is an op whether or not the slice restates it, so a
574 // pending path that changes anyway rides free and the budget is what remains after ALL of them.
575 const changed = new Set()
576 for (const p of new Set([...now.keys(), ...was.keys()])) if (now.get(p) !== was.get(p)) changed.add(p)
577 let pending = [...plan.paths].sort()
578 if (pending.length === 0) pending = [[...plan.window].sort()[0]]
579 const free = pending.filter((p) => changed.has(p))
580 const rest = pending.filter((p) => !changed.has(p))
581 const budget = this._maxCommitOps - changed.size
582 if (budget < 0) return null // ordinary changes alone fill this commit; the heal continues on the next
583 if (rest.length <= budget) return { paths: new Set(pending), final: true }
584 return { paths: new Set(free.concat(rest.slice(0, budget))), final: false }
585 }
586
587 _healBefore (paths, records) {
550588 const merged = new Map()
551589 for (const r of records) merged.set(r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString())
552590 const before = new Map()
553591 for (const r of this._lastRecords) before.set(r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString())
554 for (const p of plan.paths) {
592 for (const p of paths) {
555593 const m = merged.get(p)
556594 const was = before.get(p)
557595 before.set(p, (was !== undefined && was !== m) ? was : HEAL_PLACEHOLDER)
package.json +1 -1
……@@ -1,6 +1,6 @@
11{
22 "name": "hyper-proto-peer",
3 "version": "0.3.126",
3 "version": "0.3.127",
44 "description": "Self-custody, P2P-native atproto peer \u2014 holds the user's key + content as Hypercores and authors+signs its own repo. The source of truth in the peer/bridge split (Model B); atproto-bridge projects it to bsky.",
55 "type": "module",
66 "main": "lib/peer.js",
test/heal-op-limit.test.mjs +95 -0
……@@ -0,0 +1,95 @@
1// HEAL × RELAY OP LIMIT — a heal must never exceed 200 ops in one commit.
2//
3// indigo rejects a #commit with more than 200 ops (verify.go MaxCommitOps = 200; an ERROR in strict
4// mode, the default). A heal restates every path touched since the fork point, so a fork window that
5// touched more than 200 paths produced one commit the relay would drop — and the consumers it exists for
6// would never converge. The heal is split: partial commits restate the window in slices (`restates`),
7// the last carries `asserts`, and healPlan does not restate again what a partial already did unless a
8// fork could have made a consumer skip it.
9
10import '../lib/bare-globals.js'
11import test from 'brittle'
12import b4a from 'b4a'
13import * as dagCbor from '@ipld/dag-cbor'
14import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
15import { SwarmDevice } from '../lib/swarm-device.js'
16
17const DID = 'did:plc:healoplimit'
18const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })()
19const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-10-13T00:00:00.000Z' })
20const MAX_OPS = 200
21
22class MemKV {
23 constructor (m) { this.m = new Map(m || []) }
24 async set (k, v) { this.m.set(k, v) }
25 async get (k) { return this.m.has(k) ? this.m.get(k) : null }
26 async getNow (k) { return this.get(k) }
27 async del (k) { this.m.delete(k) }
28 async list (prefix = '') { const o = {}; for (const [k, v] of this.m) if (k.startsWith(prefix)) o[k] = v; return o }
29 async listNow (prefix = '') { return this.list(prefix) }
30}
31
32function consumer () {
33 const records = new Map()
34 let rev = ''
35 return {
36 records,
37 offer (f) {
38 if (f.rev <= rev) return
39 if ((f.ops || []).length > MAX_OPS) return // the relay drops it
40 rev = f.rev
41 for (const op of f.ops || []) {
42 if (op.action === 'create') { if (!records.has(op.path)) records.set(op.path, op.cid) } else if (op.action === 'update') records.set(op.path, op.cid)
43 else if (op.action === 'delete') records.delete(op.path)
44 }
45 }
46 }
47}
48
49const cidsOf = async (dev) => new Map((await dev.repo.listRecords(null, { now: true })).map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
50const same = (a, b) => a.size === b.size && [...a].every(([k, v]) => b.get(k) === v)
51const frames = async (sb) => Object.values(await sb.listNow('__commit__/')).sort((a, b) => (a.rev < b.rev ? -1 : 1))
52
53test('a fork window over 200 paths heals in commits of at most 200 ops, and every consumer converges', async (t) => {
54 const sb0 = new MemKV()
55 const S = new SwarmDevice(sb0, { did: DID, privateKey: KEY, self: 'S', firehose: true })
56 await S.putRecord('app.bsky.feed.post', 'base', post('base'))
57 await S.sign({ force: true })
58 const common = await frames(sb0)
59
60 // Partition. A posts 250 records in batches of 100, signing each (so no ordinary commit exceeds the
61 // limit); B posts one.
62 const sbA = new MemKV(sb0.m); const sbB = new MemKV(sb0.m)
63 const A = new SwarmDevice(sbA, { did: DID, privateKey: KEY, self: 'A', firehose: true })
64 const B = new SwarmDevice(sbB, { did: DID, privateKey: KEY, self: 'B', firehose: true })
65 for (let i = 0; i < 250; i++) {
66 await A.putRecord('app.bsky.feed.post', 'a' + String(i).padStart(3, '0'), post('A ' + i))
67 if (i % 100 === 99) await A.sign()
68 }
69 await A.sign()
70 await B.putRecord('app.bsky.feed.post', 'b', post('B'))
71 await B.sign()
72 const forkA = (await frames(sbA)).filter((f) => !common.some((c) => c.commitCid === f.commitCid))
73 const forkB = (await frames(sbB)).filter((f) => !common.some((c) => c.commitCid === f.commitCid))
74
75 // Merge and heal until quiet.
76 const sbM = new MemKV(sb0.m)
77 for (const side of [sbA, sbB]) for (const [k, v] of side.m) if (sb0.m.get(k) !== v) sbM.m.set(k, v)
78 const M = new SwarmDevice(sbM, { did: DID, privateKey: KEY, self: 'M', firehose: true })
79 let signed = 0
80 for (let i = 0; i < 10; i++) { const h = await M.sign(); if (!h) break; signed++ }
81 t.ok(signed >= 2, `the heal was split into ${signed} commits`)
82 t.is(await M.sign(), null, 'then quiet')
83
84 const all = await frames(sbM)
85 const over = all.filter((f) => (f.ops || []).length > MAX_OPS)
86 t.is(over.length, 0, 'no commit exceeds ' + MAX_OPS + ' ops (largest: ' + Math.max(...all.map((f) => (f.ops || []).length)) + ')')
87
88 const merged = await cidsOf(M)
89 const heals = all.filter((f) => !common.concat(forkA, forkB).some((c) => c.commitCid === f.commitCid))
90 for (const [name, fork] of [['A', forkA], ['B', forkB]]) {
91 const c = consumer()
92 for (const f of common.concat(fork, heals).sort((x, y) => (x.rev < y.rev ? -1 : 1))) c.offer(f)
93 t.ok(same(c.records, merged), `consumer of fork ${name} converges (${c.records.size} vs ${merged.size})`)
94 }
95})
test/multi-partition-relay.test.mjs +20 -6
……@@ -21,6 +21,11 @@ import * as dagCbor from '@ipld/dag-cbor'
2121import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
2222import { commitLine } from 'atproto-room/lib/commit-line.js'
2323import { SwarmDevice } from '../lib/swarm-device.js'
24import process from 'bare-process'
25
26// Wider runs without editing the file: MPR_SEEDS=150 MPR_STEPS=24 bare test/multi-partition-relay.test.mjs
27const SEEDS = Number(process.env.MPR_SEEDS || 30)
28const STEPS = Number(process.env.MPR_STEPS || 12)
2429
2530const DID = 'did:plc:multipartition'
2631const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })()
……@@ -61,12 +66,12 @@ class Consumer {
6166const cidsOf = async (dev) => new Map((await dev.repo.listRecords(null, { now: true })).map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
6267const same = (a, b) => a.size === b.size && [...a].every(([k, v]) => b.get(k) === v)
6368
64async function run (seed, steps = 12) {
69async function run (seed, steps = 12, maxCommitOps = 200) {
6570 let s = seed >>> 0
6671 const rnd = () => { s = (s * 1664525 + 1013904223) >>> 0; return s / 4294967296 }
6772 const pick = (arr) => arr[Math.floor(rnd() * arr.length)]
6873
69 const all = ['A', 'B', 'C'].map((name) => { const sb = new MemKV(); return { name, sb, dev: new SwarmDevice(sb, { did: DID, privateKey: KEY, self: name, firehose: true }) } })
74 const all = ['A', 'B', 'C'].map((name) => { const sb = new MemKV(); return { name, sb, dev: new SwarmDevice(sb, { did: DID, privateKey: KEY, self: name, firehose: true, maxCommitOps }) } })
7075 const consumer = new Consumer()
7176 const signAll = async (group) => { for (const n of group) { await n.dev.sign(); for (const f of await n.dev.firehoseLog()) consumer.offer(f) } }
7277
……@@ -89,7 +94,7 @@ async function run (seed, steps = 12) {
8994 for (const g of groups) { await signAll(g); sync(g) }
9095 }
9196 sync(all)
92 for (let i = 0; i < 6; i++) { await signAll(all); sync(all) }
97 for (let i = 0; i < 40; i++) { await signAll(all); sync(all) }
9398
9499 const merged = await cidsOf(all[0].dev)
95100 const heads = new Set()
……@@ -98,11 +103,20 @@ async function run (seed, steps = 12) {
98103 return { nodesAgree, oneHead: heads.size === 1, consumerOk: same(consumer.records, merged) }
99104}
100105
101test('a relay-fed consumer converges after ANY sequence of partitions (seeded, 30 seeds)', async (t) => {
106test('a relay-fed consumer converges after ANY sequence of partitions (seeded)', async (t) => {
102107 const bad = []
103 for (let seed = 1; seed <= 30; seed++) {
104 const r = await run(seed)
108 for (let seed = 1; seed <= SEEDS; seed++) {
109 const r = await run(seed, STEPS)
105110 if (!r.nodesAgree || !r.oneHead || !r.consumerOk) bad.push(seed + ':' + JSON.stringify(r))
106111 }
107112 t.alike(bad, [], 'every seed: nodes agree, one head, consumer == merged records')
108113})
114
115test('the same, with heals SPLIT: a 5-op cap forces partial heals that interleave with partitions ', async (t) => {
116 const bad = []
117 for (let seed = 1; seed <= SEEDS; seed++) {
118 const r = await run(seed, STEPS, 5)
119 if (!r.nodesAgree || !r.oneHead || !r.consumerOk) bad.push(seed + ':' + JSON.stringify(r))
120 }
121 t.alike(bad, [], 'every seed converges with split heals')
122})
test/partition-fork-heal.test.mjs +27 -0
……@@ -22,6 +22,8 @@ import b4a from 'b4a'
2222import { SwarmDevice } from '../lib/swarm-device.js'
2323import * as dagCbor from '@ipld/dag-cbor'
2424import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
25import { CID } from 'multiformats/cid'
26import { commitLine } from 'atproto-room/lib/commit-line.js'
2527
2628const DID = 'did:plc:partitionfork'
2729const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })()
……@@ -182,3 +184,28 @@ test('partition fork: a LOSING branch of two commits still reaches every consume
182184 t.ok(same(av.records, r.merged), `consumer of branch ${name} converges`)
183185 }
184186})
187
188test('partition fork: the device on the LOSING side of an equal-rev tie continues from the chain head', async (t) => {
189 // Two sides sign the same rev; the logs merge with NO heal yet, so the only thing above either fork is
190 // the other fork at the SAME rev. The device whose fork lost the tie must re-anchor to the chain head.
191 const sb0 = new MemKV()
192 const S = new SwarmDevice(sb0, { did: DID, privateKey: KEY, self: 'S', firehose: true })
193 await S.putRecord('app.bsky.feed.post', 'base', post('base'))
194 await S.sign({ force: true })
195 const sbA = new MemKV(sb0.m); const sbB = new MemKV(sb0.m)
196 const A = new SwarmDevice(sbA, { did: DID, privateKey: KEY, self: 'A', firehose: true })
197 const B = new SwarmDevice(sbB, { did: DID, privateKey: KEY, self: 'B', firehose: true })
198 await A.putRecord('app.bsky.feed.post', 'ra', post('from A'))
199 await B.putRecord('app.bsky.feed.post', 'rb', post('from B'))
200 const hA = await A.sign(); const hB = await B.sign()
201 t.is(hA.rev, hB.rev, 'precondition: an equal-rev tie')
202 // Each side receives the other's writes; each device keeps its own head.
203 for (const [from, to] of [[sbA, sbB], [sbB, sbA]]) for (const [k, v] of from.m) if (sb0.m.get(k) !== v && !to.m.has(k)) to.m.set(k, v)
204 const { head } = commitLine(Object.values(await sbA.listNow('__commit__/')))
205 const [loser, loserHead] = head.commitCid === hA.commitCid.toString() ? [B, hB] : [A, hA]
206 await loser.putRecord('app.bsky.feed.post', 'later', post('after the merge'))
207 const h = await loser.sign()
208 t.ok(h, 'the losing side signs')
209 t.not(String(h.signedCommit.prev), loserHead.commitCid.toString(), 'it did not extend its own losing fork')
210 t.is(String(h.signedCommit.prev), head.commitCid, 'it continued from the chain head')
211})