// HEAL × RELAY OP LIMIT — a heal must never exceed 200 ops in one commit.
//
// indigo rejects a #commit with more than 200 ops (verify.go MaxCommitOps = 200; an ERROR in strict
// mode, the default). A heal restates every path touched since the fork point, so a fork window that
// touched more than 200 paths produced one commit the relay would drop — and the consumers it exists for
// would never converge. The heal is split: partial commits restate the window in slices (`restates`),
// the last carries `asserts`, and healPlan does not restate again what a partial already did unless a
// fork could have made a consumer skip it.
import '../lib/bare-globals.js'
import test from 'brittle'
import b4a from 'b4a'
import * as dagCbor from '@ipld/dag-cbor'
import { cidForBytesSync } from 'atproto-repo/lib/cid.js'
import { SwarmDevice } from '../lib/swarm-device.js'
const DID = 'did:plc:healoplimit'
const KEY = (() => { const k = b4a.alloc(32); k[31] = 7; return k })()
const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-10-13T00:00:00.000Z' })
const MAX_OPS = 200
class MemKV {
constructor (m) { this.m = new Map(m || []) }
async set (k, v) { this.m.set(k, v) }
async get (k) { return this.m.has(k) ? this.m.get(k) : null }
async getNow (k) { return this.get(k) }
async del (k) { this.m.delete(k) }
async list (prefix = '') { const o = {}; for (const [k, v] of this.m) if (k.startsWith(prefix)) o[k] = v; return o }
async listNow (prefix = '') { return this.list(prefix) }
}
function consumer () {
const records = new Map()
let rev = ''
return {
records,
offer (f) {
if (f.rev <= rev) return
if ((f.ops || []).length > MAX_OPS) return // the relay drops it
rev = f.rev
for (const op of f.ops || []) {
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)
else if (op.action === 'delete') records.delete(op.path)
}
}
}
}
const cidsOf = async (dev) => new Map((await dev.repo.listRecords(null, { now: true })).map((r) => [r.collection + '/' + r.rkey, cidForBytesSync(dagCbor.encode(r.record)).toString()]))
const same = (a, b) => a.size === b.size && [...a].every(([k, v]) => b.get(k) === v)
const frames = async (sb) => Object.values(await sb.listNow('__commit__/')).sort((a, b) => (a.rev < b.rev ? -1 : 1))
test('a fork window over 200 paths heals in commits of at most 200 ops, and every consumer converges', async (t) => {
const sb0 = new MemKV()
const S = new SwarmDevice(sb0, { did: DID, privateKey: KEY, self: 'S', firehose: true })
await S.putRecord('app.bsky.feed.post', 'base', post('base'))
await S.sign({ force: true })
const common = await frames(sb0)
// Partition. A posts 250 records in batches of 100, signing each (so no ordinary commit exceeds the
// limit); B posts one.
const sbA = new MemKV(sb0.m); const sbB = new MemKV(sb0.m)
const A = new SwarmDevice(sbA, { did: DID, privateKey: KEY, self: 'A', firehose: true })
const B = new SwarmDevice(sbB, { did: DID, privateKey: KEY, self: 'B', firehose: true })
for (let i = 0; i < 250; i++) {
await A.putRecord('app.bsky.feed.post', 'a' + String(i).padStart(3, '0'), post('A ' + i))
if (i % 100 === 99) await A.sign()
}
await A.sign()
await B.putRecord('app.bsky.feed.post', 'b', post('B'))
await B.sign()
const forkA = (await frames(sbA)).filter((f) => !common.some((c) => c.commitCid === f.commitCid))
const forkB = (await frames(sbB)).filter((f) => !common.some((c) => c.commitCid === f.commitCid))
// Merge and heal until quiet.
const sbM = new MemKV(sb0.m)
for (const side of [sbA, sbB]) for (const [k, v] of side.m) if (sb0.m.get(k) !== v) sbM.m.set(k, v)
const M = new SwarmDevice(sbM, { did: DID, privateKey: KEY, self: 'M', firehose: true })
let signed = 0
for (let i = 0; i < 10; i++) { const h = await M.sign(); if (!h) break; signed++ }
t.ok(signed >= 2, `the heal was split into ${signed} commits`)
t.is(await M.sign(), null, 'then quiet')
const all = await frames(sbM)
const over = all.filter((f) => (f.ops || []).length > MAX_OPS)
t.is(over.length, 0, 'no commit exceeds ' + MAX_OPS + ' ops (largest: ' + Math.max(...all.map((f) => (f.ops || []).length)) + ')')
const merged = await cidsOf(M)
const heals = all.filter((f) => !common.concat(forkA, forkB).some((c) => c.commitCid === f.commitCid))
for (const [name, fork] of [['A', forkA], ['B', forkB]]) {
const c = consumer()
for (const f of common.concat(fork, heals).sort((x, y) => (x.rev < y.rev ? -1 : 1))) c.offer(f)
t.ok(same(c.records, merged), `consumer of fork ${name} converges (${c.records.size} vs ${merged.size})`)
}
})