| … | … | @@ -31,7 +31,9 @@ import { encodeReceipt, receiptEpoch } from 'atproto-room/lib/membership.js' |
| 31 | 31 | // key let one overwrite the other, so the losing fork vanished and its consumers never got the other |
| 32 | 32 | // side's records (test/partition-fork-heal.test.mjs). commitLine decides the chain the same way on |
| 33 | 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' |
| 34 | import { commitLine, frameKey, healPlan, CHAIN_MARK } from 'atproto-room/lib/commit-line.js' |
| 35 | // The stand-in old value that turns a restated path into an `update` op. Never a real record's CID. |
| 36 | const HEAL_PLACEHOLDER = cidForBytesSync(dagCbor.encode({ $type: 'hyper-proto-peer.healPlaceholder' })).toString() |
| 35 | 37 | |
| 36 | 38 | // StateBaseV3's durable banned-identity set + minimum-epoch receipt (see state-base.js). |
| 37 | 39 | // atproto-base-REVOCATION.md §3/§6. |
| … | … | @@ -70,8 +72,9 @@ function serializeEntry (write, seq) { |
| 70 | 72 | // Written by the fork-aware signer: only marked frames can be classified as forks (commit-line.js — |
| 71 | 73 | // legacy logs carry restarted chains and falling revs that must stay on the line untouched). |
| 72 | 74 | 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 } : {}), |
| 75 | // A heal: the fork commits it knew (heals) and the fork-point rev from which it restated every touched |
| 76 | // path (asserts). atproto-room healPlan decides from these whether a heal can have been skipped. |
| 77 | ...(write.asserts !== undefined ? { heals: write.heals || [], asserts: write.asserts } : {}), |
| 75 | 78 | commit: hex(write.commitBlock.bytes), |
| 76 | 79 | 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 })), |
| 77 | 80 | blocks: write.newBlocks.map((b) => ({ cid: b.cid.toString(), bytes: hex(b.bytes) })) |
| … | … | @@ -493,26 +496,22 @@ export class SwarmDevice { |
| 493 | 496 | const rev = this._nextRev() |
| 494 | 497 | const prev = this.head ? this.head.commitCid : null |
| 495 | 498 | const out = materialize(records, { did: this.did, rev, prev, privateKey: this.privateKey }) |
| 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 |
| 499 | // Forks to heal (atproto-room healPlan): a consumer behind a relay may hold ANY mix of branches, so the |
| 500 | // heal restates every path any frame touched since the earliest uncovered fork point. |
| 501 | const plan = (!force && chain) ? healPlan(chain) : null |
| 502 | if (this.head && out.dataCid.equals(this.head.dataCid) && !plan) return null // no new intents |
| 501 | 503 | // persist the #commit into the shared firehose log (option a) BEFORE advancing head, so every |
| 502 | 504 | // device + the membrane sees each signed commit. The diff is _lastRecords -> records, or for a heal |
| 503 | | // the fork-aware state assembled by _healBefore. |
| 505 | // the asserting state built by _healBefore. |
| 504 | 506 | if (this.firehose) { |
| 505 | 507 | const since = this.head ? this.head.rev : null |
| 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 |
| 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 } |
| 510 | 510 | // Only append a #commit frame when the diff is non-empty. A 0-op frame can only arise from the |
| 511 | 511 | // rare restart-with-unsigned-pending-write edge (baseline == current yet head.dataCid differs); |
| 512 | 512 | // emitting it would be a malformed commit. head still advances below so we don't re-trigger. |
| 513 | 513 | if (write.ops.length > 0) { |
| 514 | | // seq is informational now: the chain is ordered by rev (commitLine), because two partitioned |
| 515 | | // signers both write count+1 here. |
| 514 | // seq is informational now: two partitioned signers both write count+1 here. |
| 516 | 515 | const existing = await this.sb.listNow(FIREHOSE_PREFIX) |
| 517 | 516 | await this.sb.set(frameKey(rev, write.commitCid), serializeEntry(write, Object.keys(existing).length + 1)) |
| 518 | 517 | } |
| … | … | @@ -538,64 +537,26 @@ export class SwarmDevice { |
| 538 | 537 | return Object.values(map).filter((f) => f && f.rev) |
| 539 | 538 | } |
| 540 | 539 | |
| 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, |
| 540 | // The "before" state a heal commit diffs from, so that its ops RESTATE every path in the plan: |
| 541 | // a record in the merged set -> `update` (never `create`: the reference bsky AppView ignores a create |
| 542 | // for a record it already holds, so a consumer holding a stale copy would keep it). Where the chain's |
| 543 | // own previous value differs it is used; otherwise a fixed placeholder CID stands in, so the diff |
| 544 | // still yields an update. |
| 545 | // a path absent from the merged set -> `delete` (a no-op for a consumer that never had it). |
| 546 | // Every other path keeps the chain's previous value, so ordinary changes diff as usual. buildCommitWrite |
| 547 | // still proves the ops invert to this state's root; only `prevData` names a state no consumer held, |
| 551 | 548 | // 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) { |
| 549 | _healBefore (plan, records) { |
| 556 | 550 | const merged = new Map() |
| 557 | 551 | 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) { |
| 552 | const before = new Map() |
| 553 | 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) { |
| 588 | 555 | 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 |
| 556 | const was = before.get(p) |
| 557 | before.set(p, (was !== undefined && was !== m) ? was : HEAL_PLACEHOLDER) |
| 598 | 558 | } |
| 559 | return [...before].map(([path, cid]) => { const i = path.indexOf('/'); return { collection: path.slice(0, i), rkey: path.slice(i + 1), cid } }) |
| 599 | 560 | } |
| 600 | 561 | |
| 601 | 562 | // Re-seed head + the diff baseline from the shared __commit__ log after a (re)start. Without this, |