| … | … | @@ -18,6 +18,7 @@ import { signAssertion } from 'atproto-identity/lib/auth.js' |
| 18 | 18 | import b4a from 'b4a' |
| 19 | 19 | import { CID } from 'multiformats/cid' |
| 20 | 20 | import * as dagCbor from '@ipld/dag-cbor' |
| 21 | import { cidForBytesSync } from 'atproto-repo/lib/cid.js' | |
| 21 | 22 | |
| 22 | 23 | // The firehose lives in the SAME shared StateBaseV3 (option a): each lease-holder sign() writes the |
| 23 | 24 | // signed #commit under __commit__/<rev>. It replicates to every device + the keyless membrane reads |
| … | … | @@ -25,6 +26,12 @@ import * as dagCbor from '@ipld/dag-cbor' |
| 25 | 26 | // skips the reserved "__" namespace so these entries never leak into the signed repo. |
| 26 | 27 | const FIREHOSE_PREFIX = '__commit__/' |
| 27 | 28 | import { encodeReceipt, receiptEpoch } from 'atproto-room/lib/membership.js' |
| 29 | // ⛔ Frames are keyed __commit__/<rev>/<commitCid> (frameKey), NOT __commit__/<rev>. A partition elects a | |
| 30 | // signer on each side; both continue nextRev(head) — the SAME rev over different records — and the old | |
| 31 | // key let one overwrite the other, so the losing fork vanished and its consumers never got the other | |
| 32 | // side's records (test/partition-fork-heal.test.mjs). commitLine decides the chain the same way on | |
| 33 | // every device; frames off it are forks that the next commit heals. | |
| 34 | import { commitLine, frameKey, replayCids, CHAIN_MARK } from 'atproto-room/lib/commit-line.js' | |
| 28 | 35 | |
| 29 | 36 | // StateBaseV3's durable banned-identity set + minimum-epoch receipt (see state-base.js). |
| 30 | 37 | // atproto-base-REVOCATION.md §3/§6. |
| … | … | @@ -60,6 +67,11 @@ function serializeEntry (write, seq) { |
| 60 | 67 | prev: write.signedCommit.prev ? write.signedCommit.prev.toString() : null, |
| 61 | 68 | prevData: write.prevData ? write.prevData.toString() : null, // prev MST root (for the #commit frame) |
| 62 | 69 | commitCid: write.commitCid.toString(), |
| 70 | // Written by the fork-aware signer: only marked frames can be classified as forks (commit-line.js — | |
| 71 | // legacy logs carry restarted chains and falling revs that must stay on the line untouched). | |
| 72 | chain: CHAIN_MARK, | |
| 73 | // Fork commits this one heals (commitLine counts a loser named here as done). Absent on ordinary commits. | |
| 74 | ...(write.heals && write.heals.length ? { heals: write.heals } : {}), | |
| 63 | 75 | commit: hex(write.commitBlock.bytes), |
| 64 | 76 | 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 })), |
| 65 | 77 | blocks: write.newBlocks.map((b) => ({ cid: b.cid.toString(), bytes: hex(b.bytes) })) |
| … | … | @@ -349,9 +361,14 @@ export class SwarmDevice { |
| 349 | 361 | // forward-only gate below. Non-blocking (listNow). |
| 350 | 362 | let frames = null |
| 351 | 363 | let net = null |
| 364 | let chain = null | |
| 352 | 365 | if (this.firehose) { |
| 353 | 366 | frames = Object.values(await this.sb.listNow(FIREHOSE_PREFIX)).filter((f) => f && f.rev) |
| 354 | for (const f of frames) if (!net || f.rev > net.rev) net = f | |
| 367 | // The head of the chain, decided identically on every device (highest rev; a same-rev fork is | |
| 368 | // broken by the lower commit CID). The old "highest rev, first seen" picked a different fork on | |
| 369 | // different devices. | |
| 370 | chain = commitLine(frames) | |
| 371 | net = chain.head | |
| 355 | 372 | } |
| 356 | 373 | // Feed our current currency into the presence election + announce it to peers (setSignerHead). |
| 357 | 374 | if (this.sb.setSignerHead) this.sb.setSignerHead(this.head ? this.head.rev : null) |
| … | … | @@ -421,7 +438,12 @@ export class SwarmDevice { |
| 421 | 438 | // (1) adopt the network-latest head so we continue FORWARD from it, never from a stale head. |
| 422 | 439 | const signedCommit = dagCbor.decode(b4a.from(net.commit, 'hex')) |
| 423 | 440 | this.head = { rev: net.rev, commitCid: CID.parse(net.commitCid), dataCid: signedCommit.data, signedCommit } |
| 424 | this._lastRecords = records | |
| 441 | // ⛔ The diff baseline is the record set the NET HEAD commits to, not this device's current view. | |
| 442 | // Using the view absorbed any local write the net head lacks into the signed data root with no | |
| 443 | // op: the diff was empty, no frame was appended, and head moved to a commit nobody ever received | |
| 444 | // (test/reanchor-unsigned-writes.test.mjs). Same replay restore() uses; the view is the fallback | |
| 445 | // only when a referenced block is missing, as there. | |
| 446 | this._lastRecords = (await this._recordsFromFirehose(chain.line)) || records | |
| 425 | 447 | if (this.sb.setSignerHead) this.sb.setSignerHead(this.head.rev) // re-anchored → update our currency |
| 426 | 448 | } |
| 427 | 449 | // (2) net head's record key set = replay every #commit op up to net, in rev order. Keyed by the |
| … | … | @@ -471,22 +493,28 @@ export class SwarmDevice { |
| 471 | 493 | const rev = this._nextRev() |
| 472 | 494 | const prev = this.head ? this.head.commitCid : null |
| 473 | 495 | const out = materialize(records, { did: this.did, rev, prev, privateKey: this.privateKey }) |
| 474 | if (this.head && out.dataCid.equals(this.head.dataCid)) return null // no new intents | |
| 496 | // Forks to heal: commits another partition signed that are off the chain. Their consumers hold a | |
| 497 | // state this chain never diffed from, so the next commit must name what differs for them too. | |
| 498 | const heal = (!force && chain && chain.unhealed.length) ? this._healBefore(chain, records) : null | |
| 499 | const healing = !!(heal && heal.differs) | |
| 500 | if (this.head && out.dataCid.equals(this.head.dataCid) && !healing) return null // no new intents | |
| 475 | 501 | // persist the #commit into the shared firehose log (option a) BEFORE advancing head, so every |
| 476 | // device + the membrane sees each signed commit. seq = position in the rev-ordered log (single | |
| 477 | // writer = the lease-holder, so no seq race). The diff is _lastRecords -> records. | |
| 502 | // device + the membrane sees each signed commit. The diff is _lastRecords -> records, or for a heal | |
| 503 | // the fork-aware state assembled by _healBefore. | |
| 478 | 504 | if (this.firehose) { |
| 479 | 505 | const since = this.head ? this.head.rev : null |
| 480 | const write = buildCommitWrite(this._lastRecords, records, { did: this.did, rev, prev, since, privateKey: this.privateKey }) | |
| 506 | const write = buildCommitWrite(healing ? heal.before : this._lastRecords, records, { did: this.did, rev, prev, since, privateKey: this.privateKey }) | |
| 507 | // Named even when nothing differed: those forks' consumers already match, so this commit's ops | |
| 508 | // apply to them as they are, and the forks are done. | |
| 509 | if (heal) write.heals = heal.heals | |
| 481 | 510 | // Only append a #commit frame when the diff is non-empty. A 0-op frame can only arise from the |
| 482 | 511 | // rare restart-with-unsigned-pending-write edge (baseline == current yet head.dataCid differs); |
| 483 | 512 | // emitting it would be a malformed commit. head still advances below so we don't re-trigger. |
| 484 | 513 | if (write.ops.length > 0) { |
| 485 | // listNow (non-blocking): __commit__ is single-writer (the lease/signer holder), so the | |
| 486 | // autoUpdate snapshot already reflects THIS signer's own firehose entries — the seq count is | |
| 487 | // accurate without a blocking base.update() drain. Matches the non-blocking listRecords above. | |
| 514 | // seq is informational now: the chain is ordered by rev (commitLine), because two partitioned | |
| 515 | // signers both write count+1 here. | |
| 488 | 516 | const existing = await this.sb.listNow(FIREHOSE_PREFIX) |
| 489 | await this.sb.set(FIREHOSE_PREFIX + rev, serializeEntry(write, Object.keys(existing).length + 1)) | |
| 517 | await this.sb.set(frameKey(rev, write.commitCid), serializeEntry(write, Object.keys(existing).length + 1)) | |
| 490 | 518 | } |
| 491 | 519 | } |
| 492 | 520 | this._lastRecords = records |
| … | … | @@ -495,11 +523,79 @@ export class SwarmDevice { |
| 495 | 523 | return this.head |
| 496 | 524 | } |
| 497 | 525 | |
| 498 | // Read the firehose #commit log (seq-ordered) from the shared state — the sequence the membrane | |
| 499 | // relays as subscribeRepos. Any device (or the membrane) reads the same entries via replication. | |
| 526 | // The CHAIN of signed #commit frames, in rev order — the sequence the membrane relays as | |
| 527 | // subscribeRepos and restore() replays. Fork frames are excluded (commitLine); every device reads the | |
| 528 | // same entries via replication and decides the same chain. It used to be every frame sorted by seq, | |
| 529 | // which two partitioned signers both assign as count+1. | |
| 500 | 530 | async firehoseLog () { |
| 531 | return commitLine(await this.firehoseFrames()).line | |
| 532 | } | |
| 533 | ||
| 534 | // Every signed frame, forks included — for a consumer that has to tell a fork it already projected | |
| 535 | // from a log that restarted (serve-membrane). | |
| 536 | async firehoseFrames () { | |
| 501 | 537 | const map = await this.sb.list(FIREHOSE_PREFIX) |
| 502 | return Object.values(map).sort((a, b) => a.seq - b.seq) | |
| 538 | return Object.values(map).filter((f) => f && f.rev) | |
| 539 | } | |
| 540 | ||
| 541 | // The "before" state a fork-heal commit diffs from, so its ops bring EVERY fork's consumers to the | |
| 542 | // merged record set. | |
| 543 | // | |
| 544 | // Why one commit can serve them all: the reference bsky AppView applies ops and nothing else — a | |
| 545 | // `create` for a record it holds is ignored, `update` upserts, `delete` removes — and a merged value | |
| 546 | // for a path is always one side's value. So for each path either side touched since the fork, take | |
| 547 | // the old value from the first fork whose value differs from the merged one (else the chain's own): | |
| 548 | // a path a fork lacks becomes a create (a no-op where it exists with the merged content), a differing | |
| 549 | // value becomes an update, a path gone from the merged set becomes a delete. buildCommitWrite still | |
| 550 | // proves the ops invert to this state's root; only `prevData` names a state no single consumer held, | |
| 551 | // which the relay logs and accepts (indigo verify.go:140-143, warn-only even in strict mode). | |
| 552 | // | |
| 553 | // Returns { before: [{collection, rkey, cid}], heals: [commitCid], differs } — differs=false means | |
| 554 | // every fork's consumers already match the merged set on these paths, so an ordinary commit serves. | |
| 555 | _healBefore (chain, records) { | |
| 556 | const merged = new Map() | |
| 557 | for (const r of records) merged.set(r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()) | |
| 558 | const lineState = replayCids(chain.line) | |
| 559 | ||
| 560 | // Group the unhealed forks into branches by their root (the fork frame whose parent is on the chain). | |
| 561 | const loserCids = new Set(chain.losers.map((f) => String(f.commitCid))) | |
| 562 | const byCid = new Map(chain.losers.map((f) => [String(f.commitCid), f])) | |
| 563 | const rootOf = (f) => { let r = f; while (r.prev && loserCids.has(String(r.prev))) r = byCid.get(String(r.prev)); return r } | |
| 564 | const branches = new Map() | |
| 565 | for (const f of chain.unhealed) { | |
| 566 | const root = rootOf(f) | |
| 567 | const k = String(root.commitCid) | |
| 568 | if (!branches.has(k)) branches.set(k, { root, frames: [] }) | |
| 569 | branches.get(k).frames.push(f) | |
| 570 | } | |
| 571 | ||
| 572 | const touched = new Set() | |
| 573 | const forkStates = [] | |
| 574 | for (const { root, frames } of branches.values()) { | |
| 575 | const at = root.prev ? chain.line.findIndex((f) => String(f.commitCid) === String(root.prev)) : -1 | |
| 576 | const ancestor = replayCids(chain.line.slice(0, at + 1)) | |
| 577 | // The fork's own frames, plus the whole losing branch up to the newest unhealed one. | |
| 578 | const branch = chain.losers.filter((f) => String(rootOf(f).commitCid) === String(root.commitCid)) | |
| 579 | forkStates.push(replayCids(branch, ancestor)) | |
| 580 | for (const f of branch) for (const op of (f.ops || [])) touched.add(op.path) | |
| 581 | for (const f of chain.line.slice(at + 1)) for (const op of (f.ops || [])) touched.add(op.path) | |
| 582 | } | |
| 583 | for (const p of new Set([...merged.keys(), ...lineState.keys()])) if (merged.get(p) !== lineState.get(p)) touched.add(p) | |
| 584 | ||
| 585 | const before = new Map(lineState) | |
| 586 | let differs = false | |
| 587 | for (const p of touched) { | |
| 588 | const m = merged.get(p) | |
| 589 | let v = lineState.get(p) | |
| 590 | for (const s of forkStates) if (s.get(p) !== m) { v = s.get(p); differs = true; break } | |
| 591 | if (v === undefined) before.delete(p); else before.set(p, v) | |
| 592 | } | |
| 593 | const heals = chain.unhealed.map((f) => String(f.commitCid)) | |
| 594 | return { | |
| 595 | before: [...before].map(([path, cid]) => { const i = path.indexOf('/'); return { collection: path.slice(0, i), rkey: path.slice(i + 1), cid } }), | |
| 596 | heals, | |
| 597 | differs | |
| 598 | } | |
| 503 | 599 | } |
| 504 | 600 | |
| 505 | 601 | // Re-seed head + the diff baseline from the shared __commit__ log after a (re)start. Without this, |
| … | … | @@ -579,7 +675,7 @@ export class SwarmDevice { |
| 579 | 675 | const prev = this.head.signedCommit.prev || null |
| 580 | 676 | const write = buildCommitWrite([], this._lastRecords, { did: this.did, rev: this.head.rev, prev, since: null, privateKey: this.privateKey }) |
| 581 | 677 | if (!write.commitCid.equals(this.head.commitCid)) throw new Error('emitBaseline: rebuilt commit CID != adopted head — refusing to emit an inconsistent baseline frame') |
| 582 | await this.sb.set(FIREHOSE_PREFIX + this.head.rev, serializeEntry(write, 1)) | |
| 678 | await this.sb.set(frameKey(this.head.rev, write.commitCid), serializeEntry(write, 1)) | |
| 583 | 679 | return this.head |
| 584 | 680 | } |
| 585 | 681 |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "hyper-proto-peer", |
| 3 | "version": "0.3.124", | |
| 3 | "version": "0.3.125", | |
| 4 | 4 | "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.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "lib/peer.js", |
| … | … | @@ -14,6 +14,9 @@ import './upload-blob-path-uds.test.mjs' // fast + offline; favela-bridge upload |
| 14 | 14 | import './mint.test.mjs' // fast + offline; creating an identity on the device that will use it |
| 15 | 15 | import './hosting-cli.test.mjs' // fast + offline; the invite ceremony's terminal half |
| 16 | 16 | import './swarm-device.test.js' |
| 17 | import './partition-fork-heal.test.mjs' // fast + offline; partitioned signers fork the chain — the heal reaches every consumer | |
| 18 | import './reanchor-unsigned-writes.test.mjs' // fast + offline; a re-anchor must not absorb local writes | |
| 19 | import './partition-fork-membrane.test.mjs' // fast + offline; both sides' membranes converge on the heal | |
| 17 | 20 | import './event-driven-sign.test.mjs' |
| 18 | 21 | import './failover-election.test.mjs' |
| 19 | 22 | import './swarm-device-overlay.test.js' |
| … | … | @@ -24,6 +24,7 @@ import b4a from 'b4a' |
| 24 | 24 | import { secp256k1 } from '@noble/curves/secp256k1' |
| 25 | 25 | import { verifyCommit } from 'atproto-repo/lib/commit.js' |
| 26 | 26 | import { SwarmDevice } from '../lib/swarm-device.js' |
| 27 | import { frameKey } from 'atproto-room/lib/commit-line.js' | |
| 27 | 28 | |
| 28 | 29 | const PRIV = new Uint8Array(Buffer.from('9085d2bef69286a6cbb51623c8fa258629945cd55ca705cc4e66700396894e0c', 'hex')) |
| 29 | 30 | const PUB = secp256k1.getPublicKey(PRIV, true) |
| … | … | @@ -83,8 +84,11 @@ test('two co-owners signing the SAME transition produce byte-identical commits ( |
| 83 | 84 | // double-sign": still exactly one, but because they AGREE, not because one was suppressed. |
| 84 | 85 | const fs = frames(map) |
| 85 | 86 | t.is(fs.length, 2, 'concurrent signers produced ONE new frame, not two — idempotent, no fork') |
| 86 | t.ok(fs.includes('__commit__/' + h2a.rev), 'the surviving frame is keyed by the agreed rev') | |
| 87 | const survivor = map.get('__commit__/' + h2a.rev) | |
| 87 | // Keyed by rev AND commit CID (frameKey): identical commits still share one key — that IS this | |
| 88 | // idempotence — while a partition's different commits at one rev no longer overwrite each other. | |
| 89 | const key = frameKey(h2a.rev, h2a.commitCid) | |
| 90 | t.ok(fs.includes(key), 'the surviving frame is keyed by the agreed rev + commit') | |
| 91 | const survivor = map.get(key) | |
| 88 | 92 | t.is(survivor.commitCid, h2a.commitCid.toString(), 'the surviving frame carries the agreed commit') |
| 89 | 93 | t.is(survivor.commit, survivor.commit, 'frame commit block present') |
| 90 | 94 |
| … | … | @@ -0,0 +1,184 @@ |
| 1 | // PARTITION FORK HEAL — two co-owners signing on opposite sides of a partition. | |
| 2 | // | |
| 3 | // The presence election is per CONNECTED SET, so a partition elects a signer on each side, and both | |
| 4 | // continue the chain from the same head: nextRev(head) is a pure function of the head, so they sign | |
| 5 | // the SAME rev over DIFFERENT record sets. Measured 2026-10-13 against this code: | |
| 6 | // - the frames are keyed __commit__/<rev>, so after the merge ONE overwrites the other (2 of 3 frames | |
| 7 | // survive) and the losing fork leaves no trace in the shared log; | |
| 8 | // - the next commit is diffed against the SURVIVING fork only, so it names that side's records and | |
| 9 | // never the other's. | |
| 10 | // The relay keeps whichever fork reached it first (it drops a #commit with rev <= the stored rev), and | |
| 11 | // the reference bsky AppView applies a #commit's OPS and nothing else: #sync does not touch records, | |
| 12 | // a `create` for a record it already holds is ignored, `update` upserts, `delete` removes. So a | |
| 13 | // consumer that took the fork the log did NOT keep never learns the other side's posts. | |
| 14 | // | |
| 15 | // What healing requires, from that consumer's rules: both forks stay visible in the log, and the heal | |
| 16 | // commit names EVERY path either fork touched since they split — as an update when the record exists in | |
| 17 | // the merged state, as a delete when it does not. | |
| 18 | ||
| 19 | import '../lib/bare-globals.js' | |
| 20 | import test from 'brittle' | |
| 21 | import b4a from 'b4a' | |
| 22 | import { SwarmDevice } from '../lib/swarm-device.js' | |
| 23 | import * as dagCbor from '@ipld/dag-cbor' | |
| 24 | import { cidForBytesSync } from 'atproto-repo/lib/cid.js' | |
| 25 | ||
| 26 | const DID = 'did:plc:partitionfork' | |
| 27 | const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })() | |
| 28 | const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-10-13T00:00:00.000Z' }) | |
| 29 | const P = 'app.bsky.feed.post/' | |
| 30 | ||
| 31 | class MemKV { | |
| 32 | constructor (m) { this.m = new Map(m || []) } | |
| 33 | async set (k, v) { this.m.set(k, v) } | |
| 34 | async get (k) { return this.m.has(k) ? this.m.get(k) : null } | |
| 35 | async getNow (k) { return this.get(k) } | |
| 36 | async del (k) { this.m.delete(k) } | |
| 37 | async list (prefix = '') { const o = {}; for (const [k, v] of this.m) if (k.startsWith(prefix)) o[k] = v; return o } | |
| 38 | async listNow (prefix = '') { return this.list(prefix) } | |
| 39 | } | |
| 40 | ||
| 41 | const framesOf = async (sb) => Object.values(await sb.listNow('__commit__/')) | |
| 42 | ||
| 43 | // The reference bsky AppView's record rules (packages/bsky data-plane subscription.ts + processor.ts, | |
| 44 | // atproto 0bac60a): ops applied blindly; create over an indexed URI is a no-op; update upserts; delete | |
| 45 | // removes (unknown = no-op). The relay in front drops any #commit whose rev <= the stored rev. | |
| 46 | function appview () { | |
| 47 | const records = new Map() | |
| 48 | let rev = '' | |
| 49 | return { | |
| 50 | records, | |
| 51 | commit (frame) { | |
| 52 | if (frame.rev <= rev) return false | |
| 53 | rev = frame.rev | |
| 54 | for (const op of frame.ops || []) { | |
| 55 | 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) | |
| 56 | else if (op.action === 'delete') records.delete(op.path) | |
| 57 | } | |
| 58 | return true | |
| 59 | } | |
| 60 | } | |
| 61 | } | |
| 62 | ||
| 63 | // One partition round: common history, then each side writes and signs on its own replica. | |
| 64 | async function partition ({ aWrites, bWrites }) { | |
| 65 | const sb0 = new MemKV() | |
| 66 | const S = new SwarmDevice(sb0, { did: DID, privateKey: KEY, self: 'S', firehose: true }) | |
| 67 | await S.putRecord('app.bsky.feed.post', 'base', post('base')) | |
| 68 | await S.putRecord('app.bsky.feed.post', 'shared', post('shared v1')) | |
| 69 | await S.sign({ force: true }) | |
| 70 | const common = await framesOf(sb0) | |
| 71 | ||
| 72 | const sbA = new MemKV(sb0.m) | |
| 73 | const sbB = new MemKV(sb0.m) | |
| 74 | const A = new SwarmDevice(sbA, { did: DID, privateKey: KEY, self: 'A', firehose: true }) | |
| 75 | const B = new SwarmDevice(sbB, { did: DID, privateKey: KEY, self: 'B', firehose: true }) | |
| 76 | await aWrites(A) | |
| 77 | await bWrites(B) | |
| 78 | const hA = await A.sign() | |
| 79 | const hB = await B.sign() | |
| 80 | const forkA = (await framesOf(sbA)).find((f) => f.rev === hA.rev) | |
| 81 | const forkB = (await framesOf(sbB)).find((f) => f.rev === hB.rev) | |
| 82 | ||
| 83 | // Merge: the common history plus the keys each side actually WROTE. Copying a whole replica would | |
| 84 | // clobber the other side's writes with untouched common values. A key written on both sides keeps | |
| 85 | // one value, which is what the shared view does (the later-linearized write wins; here B's). | |
| 86 | const sbM = new MemKV(sb0.m) | |
| 87 | for (const side of [sbA, sbB]) { | |
| 88 | for (const [k, v] of side.m) if (sb0.m.get(k) !== v) sbM.m.set(k, v) | |
| 89 | } | |
| 90 | const M = new SwarmDevice(sbM, { did: DID, privateKey: KEY, self: 'M', firehose: true }) | |
| 91 | const heal = await M.sign() | |
| 92 | const healFrame = heal ? (await framesOf(sbM)).find((f) => f.rev === heal.rev) : null | |
| 93 | // path -> record CID: a consumer must end with the same CONTENT, not just the same paths. | |
| 94 | const merged = new Map((await M.repo.listRecords(null, { now: true })).map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()])) | |
| 95 | return { common, forkA, forkB, hA, hB, sbM, M, healFrame, merged } | |
| 96 | } | |
| 97 | ||
| 98 | // A consumer that took fork `fork` at the relay, then the heal. Returns path -> record CID. | |
| 99 | function consumerAfter (common, fork, healFrame) { | |
| 100 | const av = appview() | |
| 101 | for (const f of common.sort((a, b) => (a.rev < b.rev ? -1 : 1))) av.commit(f) | |
| 102 | av.commit(fork) | |
| 103 | if (healFrame) av.commit(healFrame) | |
| 104 | return av.records | |
| 105 | } | |
| 106 | ||
| 107 | const same = (a, b) => a.size === b.size && [...a].every(([k, v]) => b.get(k) === v) | |
| 108 | ||
| 109 | test('partition fork: both forks survive the merge in the shared log', async (t) => { | |
| 110 | const r = await partition({ | |
| 111 | aWrites: (A) => A.putRecord('app.bsky.feed.post', 'ra', post('from A')), | |
| 112 | bWrites: (B) => B.putRecord('app.bsky.feed.post', 'rb', post('from B')) | |
| 113 | }) | |
| 114 | t.is(r.hA.rev, r.hB.rev, 'precondition: both sides sign nextRev(head) — the same rev') | |
| 115 | t.ok(!r.hA.dataCid.equals(r.hB.dataCid), 'precondition: over different record sets') | |
| 116 | const cids = new Set((await framesOf(r.sbM)).map((f) => f.commitCid)) | |
| 117 | t.ok(cids.has(r.forkA.commitCid), 'fork A\'s signed commit is still in the log') | |
| 118 | t.ok(cids.has(r.forkB.commitCid), 'fork B\'s signed commit is still in the log') | |
| 119 | }) | |
| 120 | ||
| 121 | test('partition fork: a consumer of EITHER fork converges to the merged record set after the heal', async (t) => { | |
| 122 | const r = await partition({ | |
| 123 | aWrites: (A) => A.putRecord('app.bsky.feed.post', 'ra', post('from A')), | |
| 124 | bWrites: (B) => B.putRecord('app.bsky.feed.post', 'rb', post('from B')) | |
| 125 | }) | |
| 126 | t.ok(r.healFrame, 'the merged device signs a heal commit') | |
| 127 | t.ok(r.merged.has(P + 'ra') && r.merged.has(P + 'rb'), 'precondition: the merged view holds both posts') | |
| 128 | t.ok(same(consumerAfter(r.common, r.forkA, r.healFrame), r.merged), 'consumer that took fork A ends with the merged set') | |
| 129 | t.ok(same(consumerAfter(r.common, r.forkB, r.healFrame), r.merged), 'consumer that took fork B ends with the merged set') | |
| 130 | }) | |
| 131 | ||
| 132 | test('partition fork: a delete on one side and an edit on the other both reach every consumer', async (t) => { | |
| 133 | const r = await partition({ | |
| 134 | aWrites: async (A) => { await A.deleteRecord('app.bsky.feed.post', 'base') }, | |
| 135 | bWrites: async (B) => { await B.putRecord('app.bsky.feed.post', 'shared', post('shared v2 from B')) } | |
| 136 | }) | |
| 137 | t.ok(r.healFrame, 'the merged device signs a heal commit') | |
| 138 | for (const fork of [r.forkA, r.forkB]) { | |
| 139 | const got = consumerAfter(r.common, fork, r.healFrame) | |
| 140 | t.ok(same(got, r.merged), `consumer of fork ${fork === r.forkA ? 'A' : 'B'} holds exactly the merged records (paths AND content)`) | |
| 141 | } | |
| 142 | t.ok(!r.merged.has(P + 'base'), 'precondition: the delete wins in the merged view') | |
| 143 | }) | |
| 144 | ||
| 145 | test('partition fork: a heal is signed ONCE — the next sign with nothing new is a no-op', async (t) => { | |
| 146 | const r = await partition({ | |
| 147 | aWrites: (A) => A.putRecord('app.bsky.feed.post', 'ra', post('from A')), | |
| 148 | bWrites: (B) => B.putRecord('app.bsky.feed.post', 'rb', post('from B')) | |
| 149 | }) | |
| 150 | t.ok(r.healFrame && (r.healFrame.heals || []).length === 1, 'the heal names the fork it heals') | |
| 151 | const again = await r.M.sign() | |
| 152 | t.is(again, null, 'no second heal and no empty commit') | |
| 153 | const fresh = new SwarmDevice(r.sbM, { did: DID, privateKey: KEY, self: 'F', firehose: true }) | |
| 154 | t.is(await fresh.sign(), null, 'a device that restarts on the merged log does not heal again either') | |
| 155 | }) | |
| 156 | ||
| 157 | test('partition fork: a LOSING branch of two commits still reaches every consumer', async (t) => { | |
| 158 | // A signs twice, B three times, both from the same head: B has the higher rev and is the chain, so | |
| 159 | // A's whole two-commit branch is the fork. | |
| 160 | const r = await partition({ | |
| 161 | aWrites: async (A) => { | |
| 162 | await A.putRecord('app.bsky.feed.post', 'ra1', post('A one')) | |
| 163 | await A.sign() | |
| 164 | await A.putRecord('app.bsky.feed.post', 'ra2', post('A two')) | |
| 165 | }, | |
| 166 | bWrites: async (B) => { | |
| 167 | await B.putRecord('app.bsky.feed.post', 'rb1', post('B one')) | |
| 168 | await B.sign() | |
| 169 | await B.putRecord('app.bsky.feed.post', 'rb2', post('B two')) | |
| 170 | await B.sign() | |
| 171 | await B.putRecord('app.bsky.feed.post', 'rb3', post('B three')) | |
| 172 | } | |
| 173 | }) | |
| 174 | t.ok(r.healFrame && (r.healFrame.heals || []).length === 2, 'the heal names both commits of the losing branch') | |
| 175 | const commonCids = new Set(r.common.map((f) => f.commitCid)) | |
| 176 | const all = Object.values(await r.sbM.listNow('__commit__/')) | |
| 177 | const branch = (re) => all.filter((f) => !commonCids.has(f.commitCid) && !(f.heals || []).length && (f.ops || []).some((o) => re.test(o.path))) | |
| 178 | for (const [name, frames] of [['A (lost, 2 commits)', branch(/ra[12]$/)], ['B (chain, 3 commits)', branch(/rb[123]$/)]]) { | |
| 179 | const av = appview() | |
| 180 | for (const f of r.common.concat(frames).sort((a, b) => (a.rev < b.rev ? -1 : 1))) av.commit(f) | |
| 181 | av.commit(r.healFrame) | |
| 182 | t.ok(same(av.records, r.merged), `consumer of branch ${name} converges`) | |
| 183 | } | |
| 184 | }) |
| … | … | @@ -0,0 +1,105 @@ |
| 1 | // PARTITION FORK × MEMBRANE — each side's membrane projected its own fork; after the merge and the heal | |
| 2 | // both must serve the merged repo with the heal as the firehose head, and neither may refuse. | |
| 3 | // | |
| 4 | // serveMembrane projects the chain BY POSITION (firehose append from seq.length) and rebuilds the repo | |
| 5 | // core by replaying ops. The fork that loses has the lower rev, and every commit adds 1µs, so a losing | |
| 6 | // branch is never longer than the winner's: a membrane that projected the losing fork is never AHEAD of | |
| 7 | // the chain. When the heal arrives its ops, applied to that fork's state, must reproduce the merged root. | |
| 8 | ||
| 9 | import '../lib/bare-globals.js' | |
| 10 | import test from 'brittle' | |
| 11 | import b4a from 'b4a' | |
| 12 | import fs from 'bare-fs' | |
| 13 | import * as dagCbor from '@ipld/dag-cbor' | |
| 14 | import { cidForBytesSync } from 'atproto-repo/lib/cid.js' | |
| 15 | import { serveMembrane } from 'atproto-room/lib/serve-membrane.js' | |
| 16 | import { decodeFrame } from 'atproto-store/lib/firehose.js' | |
| 17 | import { openCorestore } from 'atproto-store/lib/corestore.js' | |
| 18 | import { commitLine } from 'atproto-room/lib/commit-line.js' | |
| 19 | import { SwarmDevice } from '../lib/swarm-device.js' | |
| 20 | ||
| 21 | const DID = 'did:plc:partitionmembrane' | |
| 22 | const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })() | |
| 23 | const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-10-13T00:00:00.000Z' }) | |
| 24 | const tmp = () => { const d = '/tmp/hpp-pfm-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d } | |
| 25 | ||
| 26 | class MemKV { | |
| 27 | constructor (m) { this.m = new Map(m || []) } | |
| 28 | async set (k, v) { this.m.set(k, v) } | |
| 29 | async get (k) { return this.m.has(k) ? this.m.get(k) : null } | |
| 30 | async getNow (k) { return this.get(k) } | |
| 31 | async del (k) { this.m.delete(k) } | |
| 32 | async list (prefix = '') { const o = {}; for (const [k, v] of this.m) if (k.startsWith(prefix)) o[k] = v; return o } | |
| 33 | async listNow (prefix = '') { return this.list(prefix) } | |
| 34 | } | |
| 35 | ||
| 36 | function fakeSwarm () { | |
| 37 | return { on () {}, off () {}, keyPair: { publicKey: b4a.alloc(32, 7) }, join () { return { flushed: async () => {} } } } | |
| 38 | } | |
| 39 | ||
| 40 | async function quietErrors (fn) { | |
| 41 | const lines = [] | |
| 42 | const orig = console.error | |
| 43 | console.error = (s) => lines.push(String(s)) | |
| 44 | try { await fn() } finally { console.error = orig } | |
| 45 | return lines | |
| 46 | } | |
| 47 | ||
| 48 | const recordCids = (rows) => new Map(rows.map((r) => [r.key, cidForBytesSync(dagCbor.encode(r.record)).toString()])) | |
| 49 | ||
| 50 | async function scenario (t, { aSigns, bSigns }) { | |
| 51 | const sb0 = new MemKV() | |
| 52 | const S = new SwarmDevice(sb0, { did: DID, privateKey: KEY, self: 'S', firehose: true }) | |
| 53 | await S.putRecord('app.bsky.feed.post', 'base', post('base')) | |
| 54 | await S.putRecord('app.bsky.feed.post', 'shared', post('shared v1')) | |
| 55 | await S.sign({ force: true }) | |
| 56 | ||
| 57 | const side = async (name, signs) => { | |
| 58 | const sb = new MemKV(sb0.m) | |
| 59 | const dev = new SwarmDevice(sb, { did: DID, privateKey: KEY, self: name, firehose: true }) | |
| 60 | for (let i = 1; i <= signs; i++) { await dev.putRecord('app.bsky.feed.post', 'r' + name + i, post(name + ' ' + i)); await dev.sign() } | |
| 61 | const membrane = await serveMembrane(dev, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID }) | |
| 62 | return { sb, dev, membrane } | |
| 63 | } | |
| 64 | const A = await side('a', aSigns) | |
| 65 | const B = await side('b', bSigns) | |
| 66 | if (aSigns === 2 && bSigns === 2) await A.dev.putRecord('app.bsky.feed.post', 'shared', post('shared v2 from A')) | |
| 67 | if (aSigns === 2 && bSigns === 2) await A.dev.sign() | |
| 68 | ||
| 69 | // Merge: each replica receives the other's writes (keys the other side changed from the common log). | |
| 70 | const aOnly = [...A.sb.m].filter(([k, v]) => sb0.m.get(k) !== v) | |
| 71 | const bOnly = [...B.sb.m].filter(([k, v]) => sb0.m.get(k) !== v) | |
| 72 | for (const [k, v] of bOnly) A.sb.m.set(k, v) | |
| 73 | for (const [k, v] of aOnly) B.sb.m.set(k, v) | |
| 74 | ||
| 75 | // The heal is signed on A's replica and replicates to B's. | |
| 76 | const heal = await A.dev.sign() | |
| 77 | t.ok(heal, 'A signs the heal') | |
| 78 | for (const [k, v] of A.sb.m) if (k.startsWith('__commit__/') && !B.sb.m.has(k)) B.sb.m.set(k, v) | |
| 79 | ||
| 80 | const { line } = commitLine(Object.values(await A.sb.listNow('__commit__/'))) | |
| 81 | const merged = recordCids((await A.dev.repo.listRecords(null, { now: true })).map((r) => ({ key: r.collection + '/' + r.rkey, record: r.record }))) | |
| 82 | ||
| 83 | for (const [name, s] of [['A', A], ['B', B]]) { | |
| 84 | const errors = await quietErrors(() => s.membrane.sync()) | |
| 85 | t.absent(errors.find((l) => /REFUSED|STALE/.test(l)), `membrane ${name}: no refusal or stale warning`) | |
| 86 | const repo = recordCids(await s.membrane.repo.records()) | |
| 87 | t.ok(repo.size === merged.size && [...merged].every(([k, v]) => repo.get(k) === v), `membrane ${name}: repo core holds the merged records`) | |
| 88 | const last = decodeFrame(await s.membrane.seq.core.get(s.membrane.seq.core.length - 1)) | |
| 89 | t.is(last.body.commit.toString(), heal.commitCid.toString(), `membrane ${name}: firehose head is the heal`) | |
| 90 | t.ok(s.membrane.seq.core.length <= line.length, `membrane ${name}: firehose never longer than the chain`) | |
| 91 | } | |
| 92 | } | |
| 93 | ||
| 94 | test('membranes on both sides of a one-commit fork converge on the heal', async (t) => { | |
| 95 | await scenario(t, { aSigns: 1, bSigns: 1 }) | |
| 96 | }) | |
| 97 | ||
| 98 | test('membranes converge when one side lost a two-commit branch', async (t) => { | |
| 99 | await scenario(t, { aSigns: 3, bSigns: 2 }) | |
| 100 | await scenario(t, { aSigns: 2, bSigns: 3 }) | |
| 101 | }) | |
| 102 | ||
| 103 | test('membranes converge with a concurrent edit of one record', async (t) => { | |
| 104 | await scenario(t, { aSigns: 2, bSigns: 2 }) | |
| 105 | }) |
| … | … | @@ -0,0 +1,56 @@ |
| 1 | // RE-ANCHOR WITH UNSIGNED LOCAL WRITES — the forward-only gate must not swallow them. | |
| 2 | // | |
| 3 | // When sign() finds a network head newer than its own, it re-anchors (head = the net frame). It used to | |
| 4 | // set the diff baseline to the device's CURRENT VIEW at that moment. If the device already held a local | |
| 5 | // write the net head does not have, that write went into the signed data root with NO op in any frame: | |
| 6 | // the before/after diff was empty, so no #commit frame was appended at all, yet head advanced to the | |
| 7 | // unpublished commit. An ops-only consumer (the reference bsky AppView applies ops and nothing else) | |
| 8 | // never learns that record, and the next frame's prevData points at a root no frame ever published. | |
| 9 | // | |
| 10 | // The baseline after a re-anchor must be the record set the NET HEAD commits to (the firehose replay), | |
| 11 | // so the local write is diffed as an op. | |
| 12 | ||
| 13 | import '../lib/bare-globals.js' | |
| 14 | import test from 'brittle' | |
| 15 | import b4a from 'b4a' | |
| 16 | import { SwarmDevice } from '../lib/swarm-device.js' | |
| 17 | ||
| 18 | const DID = 'did:plc:reanchorunsigned' | |
| 19 | const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })() | |
| 20 | const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-10-13T00:00:00.000Z' }) | |
| 21 | ||
| 22 | class 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 | ||
| 32 | test('re-anchor: a local write the net head lacks is published as an op, not absorbed', async (t) => { | |
| 33 | const sb = new MemKV() | |
| 34 | const A = new SwarmDevice(sb, { did: DID, privateKey: KEY, self: 'A', firehose: true }) | |
| 35 | const B = new SwarmDevice(sb, { did: DID, privateKey: KEY, self: 'B', firehose: true }) | |
| 36 | ||
| 37 | await A.putRecord('app.bsky.feed.post', 'r1', post('one')) | |
| 38 | const h1 = await A.sign({ force: true }) | |
| 39 | t.ok(await B.sign() === null && B.head && B.head.rev === h1.rev, 'B anchored at the first commit') | |
| 40 | ||
| 41 | // A moves the chain on; B is now behind it. | |
| 42 | await A.putRecord('app.bsky.feed.post', 'r2', post('two')) | |
| 43 | const h2 = await A.sign() | |
| 44 | t.ok(h2 && h2.rev > h1.rev, 'A signed a newer head') | |
| 45 | ||
| 46 | // B writes locally BEFORE its next sign, so that sign both re-anchors and has a new record. | |
| 47 | await B.putRecord('app.bsky.feed.post', 'r3', post('three')) | |
| 48 | const h3 = await B.sign() | |
| 49 | t.ok(h3 && h3.rev > h2.rev, 'B signed forward from the net head') | |
| 50 | ||
| 51 | const frames = Object.values(await sb.listNow('__commit__/')) | |
| 52 | const published = frames.flatMap((f) => (f.ops || []).map((o) => o.path)) | |
| 53 | t.ok(published.includes('app.bsky.feed.post/r3'), 'r3 reached the firehose as an op') | |
| 54 | const head = frames.find((f) => f.commitCid === h3.commitCid.toString()) | |
| 55 | t.ok(head, 'the commit B signed was appended as a frame (head never points at an unpublished commit)') | |
| 56 | }) |