// Cross-machine e2e proving the keet-pattern Autobase design.
//
// This is the design pgh-v3 should follow. Three phases, all in one script:
//
// PHASE A — non-indexer writer canon (proves optimistic + ackInterval works)
// primary creates autobase; mirror joins as plain writer (indexer:false);
// both append; both see linearized view; indexerCount stays 1; signedLength
// advances via primary's auto-acks.
//
// PHASE B — start-indexing promotion (proves preferFastForward unstucks
// manifest migration that my earlier in-process Scenario 2 hung on)
// primary appends start-indexing op for mirror; apply calls
// host.addWriter({indexer:true}) + host.preferFastForward(); within ~5s
// both peers see indexerCount==2 and signedLength continues advancing.
//
// PHASE C — move-indexer atomic swap (proves keet's capacity-management
// trick that keeps indexer count constant)
// primary spawns an in-process tertiary peer (3rd corestore, same
// autobase key, stream-piped to primary). primary appends move-indexer
// op (demote primary, promote tertiary). After: primary is no longer in
// linearizer.indexers, tertiary is. indexerCount stays at 2 throughout
// (no manifest race).
//
// Driver — same 3-step setup as cross-machine-autobase.js:
//
// tabertron$ node test/keet-pattern-e2e.js init-primary --storage /tmp/kd-tb
// stortron$ node test/keet-pattern-e2e.js init-mirror --storage /tmp/kd-st --bootstrap <hex>
// tabertron$ node test/keet-pattern-e2e.js register-writer --storage /tmp/kd-tb --writer <mirror-hex>
// tabertron$ node test/keet-pattern-e2e.js run --storage /tmp/kd-tb --role primary
// stortron$ node test/keet-pattern-e2e.js run --storage /tmp/kd-st --role mirror
//
// Each `run` drives all three phases (mirror is passive during B's ack period
// and entirely passive during C). PASS/FAIL printed at the end.
import Autobase from 'autobase'
import Corestore from 'corestore'
import Hyperswarm from 'hyperswarm'
import b4a from 'b4a'
import fs from 'fs'
const args = parseArgs(process.argv.slice(2))
const cmd = args._[0]
async function main () {
if (cmd === 'init-primary') return initPrimary()
if (cmd === 'init-mirror') return initMirror()
if (cmd === 'register-writer') return registerWriter()
if (cmd === 'run') return runPeer()
console.error('usage: keet-pattern-e2e.js {init-primary|init-mirror|register-writer|run} [opts]')
process.exit(2)
}
function parseArgs (argv) {
const out = { _: [] }
for (let i = 0; i < argv.length; i++) {
const a = argv[i]
if (a.startsWith('--')) {
const key = a.slice(2)
const val = argv[i + 1] && !argv[i + 1].startsWith('--') ? argv[++i] : true
out[key] = val
} else out._.push(a)
}
return out
}
// Apply handles every op type the keet pattern uses.
async function apply (nodes, view, host) {
for (const node of nodes) {
const v = node.value
if (!v || !v.type) continue
try {
switch (v.type) {
case 'register-writer':
// Plain writer — indexer:false per keet's default addDevice path.
await host.addWriter(b4a.from(v.writerKey, 'hex'), { indexer: false })
break
case 'start-indexing':
await host.addWriter(b4a.from(v.writerKey, 'hex'), { indexer: true })
host.preferFastForward() // <— the missing piece that pgh-v3 lacks
break
case 'stop-indexing':
await host.addWriter(b4a.from(v.writerKey, 'hex'), { indexer: false })
host.preferFastForward()
break
case 'move-indexer':
// Atomic swap (keet's handleMoveIndexer pattern).
await host.addWriter(b4a.from(v.promoteKey, 'hex'), { indexer: true })
await host.addWriter(b4a.from(v.demoteKey, 'hex'), { indexer: false })
host.preferFastForward()
break
case 'msg':
await view.append(v)
break
}
} catch (err) {
console.error('[apply] op failed:', v.type, err.message)
}
}
}
async function openBase (storage, bootstrapKey = null) {
const store = new Corestore(storage)
await store.ready()
const base = new Autobase(store, bootstrapKey, {
apply,
open: (s) => s.get('view', { valueEncoding: 'json' }),
valueEncoding: 'json',
ackInterval: 1000, // <— keet pattern: auto-ack on a timer
ackThreshold: 4, // <— keet pattern: per-indexer churn threshold
optimistic: true // <— keet pattern: non-indexer writer optimism
})
await base.ready()
return { store, base }
}
async function initPrimary () {
if (!args.storage) throw new Error('--storage required')
if (fs.existsSync(args.storage)) {
console.error(`refusing to clobber existing storage at ${args.storage}`)
process.exit(1)
}
const { store, base } = await openBase(args.storage)
console.log(`BOOTSTRAP_KEY=${b4a.toString(base.key, 'hex')}`)
console.log(`PRIMARY_LOCAL=${b4a.toString(base.local.key, 'hex')}`)
await base.close(); await store.close()
}
async function initMirror () {
if (!args.storage) throw new Error('--storage required')
if (!args.bootstrap) throw new Error('--bootstrap <hex> required')
if (fs.existsSync(args.storage)) {
console.error(`refusing to clobber existing storage at ${args.storage}`)
process.exit(1)
}
const { store, base } = await openBase(args.storage, b4a.from(args.bootstrap, 'hex'))
console.log(`MIRROR_LOCAL=${b4a.toString(base.local.key, 'hex')}`)
await base.close(); await store.close()
}
async function registerWriter () {
if (!args.storage) throw new Error('--storage required')
if (!args.writer) throw new Error('--writer <hex> required')
const { store, base } = await openBase(args.storage)
await base.append({ type: 'register-writer', writerKey: args.writer })
await base.update()
console.log(`appended register-writer op for ${args.writer}`)
await base.close(); await store.close()
}
function snapshot (label, base) {
return {
label,
local: b4a.toString(base.local.key, 'hex').slice(0, 12),
indexerCount: base.linearizer.indexers.length,
signedLength: base.signedLength,
viewLength: base.view ? base.view.length : 0,
activeWriters: base.activeWriters.size,
writable: base.writable
}
}
function logSnap (s) {
console.log(` [${s.label}] local=${s.local} writable=${s.writable} indexers=${s.indexerCount} signed=${s.signedLength} viewLen=${s.viewLength} activeWriters=${s.activeWriters}`)
}
async function waitFor (pred, timeoutMs, pollMs = 500) {
const t0 = Date.now()
while (Date.now() - t0 < timeoutMs) {
if (await pred()) return true
await new Promise(r => setTimeout(r, pollMs))
}
return false
}
async function countMsgsByRole (base, role) {
let n = 0
const len = base.view ? base.view.length : 0
for (let i = 0; i < len; i++) {
const v = await base.view.get(i)
if (v && v.type === 'msg' && v.from === role) n++
}
return n
}
async function runPeer () {
if (!args.storage) throw new Error('--storage required')
if (!args.role) throw new Error('--role primary|mirror required')
const role = args.role
const { store, base } = await openBase(args.storage)
console.log(`[${role}] booted`)
console.log(`[${role}] autobase = ${b4a.toString(base.key, 'hex').slice(0, 16)}…`)
console.log(`[${role}] local = ${b4a.toString(base.local.key, 'hex').slice(0, 16)}…`)
logSnap(snapshot(role, base))
// Hyperswarm replication on autobase.discoveryKey.
let swarmConnected = false
const swarm = new Hyperswarm()
swarm.on('connection', (conn) => {
console.log(`[${role}] swarm connection`)
swarmConnected = true
store.replicate(conn)
})
await swarm.join(base.discoveryKey, { client: true, server: true }).flushed()
console.log(`[${role}] swarm flushed`)
const failures = []
// Wait for at least one swarm connection AND for mirror to be writable
// (mirror must see register-writer op replicated from primary).
console.log(`[${role}] waiting for swarm connection + writability…`)
if (!(await waitFor(() => swarmConnected && base.writable, 90000, 500))) {
console.error(`[${role}] FAIL setup: connected=${swarmConnected} writable=${base.writable}`)
failures.push('setup: never connected/writable')
await swarm.destroy(); await base.close(); await store.close()
process.exit(1)
}
console.log(`[${role}] connected + writable, both peers in sync`)
// ── PHASE A — non-indexer writer canon ─────────────────────────
console.log(`\n[${role}] === PHASE A: plain-writer linearization ===`)
console.log(`[${role}] appending 5 msgs`)
for (let i = 0; i < 5; i++) {
await base.append({ type: 'msg', from: role, seq: i, ts: Date.now() })
}
await base.update()
const otherRole = role === 'primary' ? 'mirror' : 'primary'
console.log(`[${role}] waiting for ${otherRole}'s 5 msgs in view…`)
const sawA = await waitFor(async () => (await countMsgsByRole(base, otherRole)) >= 5, 60000, 1000)
const finalA = snapshot(role + '@A-end', base)
logSnap(finalA)
if (!sawA) {
console.error(`[${role}] PHASE A FAIL: did not see 5 ${otherRole} msgs`)
failures.push('A: missing other-role msgs')
} else if (finalA.indexerCount !== 1) {
console.error(`[${role}] PHASE A FAIL: expected indexerCount==1, got ${finalA.indexerCount}`)
failures.push(`A: indexerCount=${finalA.indexerCount}`)
} else {
console.log(`[${role}] PHASE A PASS — non-indexer writer linearizes via primary auto-ack`)
}
// ── PHASE B — promote mirror to indexer (start-indexing) ───────
console.log(`\n[${role}] === PHASE B: start-indexing promotion ===`)
if (role === 'primary') {
if (!args['mirror-key']) {
console.error('[primary] PHASE B FAIL: --mirror-key not given')
failures.push('B: --mirror-key missing')
} else {
const mirrorKey = args['mirror-key']
console.log(`[primary] promoting mirror ${mirrorKey.slice(0, 12)} to indexer…`)
await base.append({ type: 'start-indexing', writerKey: mirrorKey })
await base.update()
}
}
console.log(`[${role}] waiting for indexerCount to become 2…`)
const sawB = await waitFor(() => base.linearizer.indexers.length === 2, 30000, 500)
const finalB = snapshot(role + '@B-end', base)
logSnap(finalB)
if (!sawB) {
console.error(`[${role}] PHASE B FAIL: indexerCount stayed at ${finalB.indexerCount}`)
failures.push(`B: indexerCount=${finalB.indexerCount}`)
} else {
// Drive a few more ops so we can observe signedLength advancing AFTER promotion.
if (base.writable) {
for (let i = 0; i < 3; i++) await base.append({ type: 'msg', from: role + '-postB', seq: i, ts: Date.now() })
await base.update()
}
await new Promise(r => setTimeout(r, 3000))
const postB = snapshot(role + '@B-after-acks', base)
logSnap(postB)
if (postB.signedLength <= finalB.signedLength) {
console.error(`[${role}] PHASE B FAIL: signedLength did not advance after promotion (was ${finalB.signedLength}, now ${postB.signedLength})`)
failures.push(`B: no signedLength advance`)
} else {
console.log(`[${role}] PHASE B PASS — indexerCount==2, signedLength advancing`)
}
}
// ── PHASE C — move-indexer (only when --phase-c flag passed) ───
// Phase C requires a 3rd peer. It hit a hypercore checkout race in
// earlier runs; deferring until A+B are solid. Run with --phase-c to
// attempt it; otherwise skip cleanly.
if (args['phase-c']) {
if (role === 'primary') {
console.log(`\n[primary] === PHASE C: move-indexer atomic swap ===`)
const cResult = await runPhaseC(base, store)
if (cResult.pass) console.log(`[primary] PHASE C PASS — atomic swap, indexerCount stayed 2`)
else { console.error(`[primary] PHASE C FAIL: ${cResult.reason}`); failures.push(`C: ${cResult.reason}`) }
} else {
console.log(`\n[mirror] === PHASE C: passive observer ===`)
await new Promise(r => setTimeout(r, 30000))
const finalC = snapshot('mirror@C-end', base)
logSnap(finalC)
if (finalC.indexerCount !== 2) failures.push(`C-mirror: indexerCount=${finalC.indexerCount}`)
}
} else {
console.log(`\n[${role}] PHASE C skipped (pass --phase-c to attempt)`)
}
console.log('')
if (failures.length === 0) console.log(`[${role}] ALL PHASES PASS`)
else console.log(`[${role}] FAILURES: ${failures.join('; ')}`)
await swarm.destroy()
await base.close()
await store.close()
process.exit(failures.length === 0 ? 0 : 1)
}
async function findMirrorWriterKey (base) {
// Mirror = writer added via register-writer that is NOT in indexers.
const sys = base._applyState && base._applyState.system
if (!sys) return null
const indexerHexSet = new Set((sys.indexers || []).map(i => i.key.toString('hex')))
// activeWriters has Writer objects with .core.key
for (const w of base.activeWriters) {
const hex = b4a.toString(w.core.key, 'hex')
if (indexerHexSet.has(hex)) continue
if (b4a.equals(w.core.key, base.local.key)) continue
return hex
}
return null
}
async function runPhaseC (primaryBase, primaryStore) {
// Spawn tertiary in-process. CRITICAL: connect the corestore stream BEFORE
// opening the autobase, so tertiary can fast-forward during ready() instead
// of crashing on "Invalid checkout N for batch, length is 0" when it tries
// to read system view state that hasn't replicated yet.
const tertiaryStorage = `${args.storage}-tertiary`
if (fs.existsSync(tertiaryStorage)) fs.rmSync(tertiaryStorage, { recursive: true, force: true })
const tertiaryStore = new Corestore(tertiaryStorage)
await tertiaryStore.ready()
console.log('[phaseC] connecting tertiary corestore stream BEFORE opening autobase…')
const s1 = primaryStore.replicate(true)
const s2 = tertiaryStore.replicate(false)
s1.on('error', () => {}); s2.on('error', () => {})
s1.pipe(s2).pipe(s1)
// Let initial corestore handshake + manifest discovery flow.
await new Promise(r => setTimeout(r, 3000))
console.log('[phaseC] opening tertiary autobase (will fast-forward over the pipe)…')
const tertiaryBase = new Autobase(tertiaryStore, primaryBase.key, {
apply,
open: (s) => s.get('view', { valueEncoding: 'json' }),
valueEncoding: 'json',
ackInterval: 1000,
ackThreshold: 4,
optimistic: true
})
await tertiaryBase.ready()
console.log(`[tertiary] ready, local=${b4a.toString(tertiaryBase.local.key, 'hex').slice(0, 12)}`)
// Wait for tertiary to fully catch up to primary's signed length.
console.log(`[phaseC] waiting for tertiary to catch up (target signedLength ≥ ${primaryBase.signedLength})…`)
if (!(await waitFor(() => tertiaryBase.signedLength >= primaryBase.signedLength, 30000, 500))) {
await tertiaryBase.close(); await tertiaryStore.close()
return { pass: false, reason: `tertiary did not catch up (got signedLength=${tertiaryBase.signedLength}, primary at ${primaryBase.signedLength})` }
}
logSnap(snapshot('tertiary@catchup', tertiaryBase))
// Register tertiary as plain writer.
console.log(`[phaseC] registering tertiary as plain writer…`)
await primaryBase.append({ type: 'register-writer', writerKey: b4a.toString(tertiaryBase.local.key, 'hex') })
await primaryBase.update()
if (!(await waitFor(() => tertiaryBase.writable, 30000, 500))) {
await tertiaryBase.close(); await tertiaryStore.close()
return { pass: false, reason: 'tertiary never became writable after register-writer' }
}
console.log('[tertiary] writable; appending 1 msg before swap')
await tertiaryBase.append({ type: 'msg', from: 'tertiary', seq: 0, ts: Date.now() })
await tertiaryBase.update()
// Snapshot pre-swap.
await primaryBase.update()
const preIC = primaryBase.linearizer.indexers.length
const preSigned = primaryBase.signedLength
console.log(`[primary] pre-swap: indexerCount=${preIC}, signedLength=${preSigned}`)
// Move-indexer atomic op: demote primary, promote tertiary.
const primaryHex = b4a.toString(primaryBase.local.key, 'hex')
const tertiaryHex = b4a.toString(tertiaryBase.local.key, 'hex')
console.log(`[primary] move-indexer: demote ${primaryHex.slice(0, 12)} promote ${tertiaryHex.slice(0, 12)}`)
await primaryBase.append({ type: 'move-indexer', demoteKey: primaryHex, promoteKey: tertiaryHex })
await primaryBase.update()
// Watch the swap. Track indexerCount sequence + ensure it never dips below 2.
let dippedBelow2 = false
const observedSeq = []
let primaryStillIndexer = true, tertiaryNowIndexer = false
const t0 = Date.now()
while (Date.now() - t0 < 60000) {
await primaryBase.update()
const ic = primaryBase.linearizer.indexers.length
if (observedSeq[observedSeq.length - 1] !== ic) observedSeq.push(ic)
if (ic < 2) dippedBelow2 = true
const hexes = primaryBase.linearizer.indexers.map(i => i.core.key.toString('hex'))
primaryStillIndexer = hexes.includes(primaryHex)
tertiaryNowIndexer = hexes.includes(tertiaryHex)
if (!primaryStillIndexer && tertiaryNowIndexer) break
await new Promise(r => setTimeout(r, 250))
}
// Drive a few more ops so signedLength can advance via the new indexer set.
if (tertiaryBase.writable) {
for (let i = 1; i <= 3; i++) await tertiaryBase.append({ type: 'msg', from: 'tertiary-postSwap', seq: i, ts: Date.now() })
await tertiaryBase.update()
}
await new Promise(r => setTimeout(r, 5000))
await primaryBase.update()
await tertiaryBase.update()
const postIC = primaryBase.linearizer.indexers.length
const postSigned = primaryBase.signedLength
const postHexes = primaryBase.linearizer.indexers.map(i => i.core.key.toString('hex'))
primaryStillIndexer = postHexes.includes(primaryHex)
tertiaryNowIndexer = postHexes.includes(tertiaryHex)
console.log(`[primary] post-swap: indexerCount=${postIC}, signedLength=${postSigned}`)
console.log(`[primary] swap sequence: ${observedSeq.join(' → ')}`)
console.log(`[primary] primaryIsIndexer=${primaryStillIndexer} tertiaryIsIndexer=${tertiaryNowIndexer}`)
await tertiaryBase.close()
await tertiaryStore.close()
if (dippedBelow2) return { pass: false, reason: `indexerCount dropped below 2 during swap (seq: ${observedSeq.join(' → ')})` }
if (primaryStillIndexer) return { pass: false, reason: 'primary still in indexers post-swap' }
if (!tertiaryNowIndexer) return { pass: false, reason: 'tertiary not in indexers post-swap' }
if (postIC !== 2) return { pass: false, reason: `final indexerCount=${postIC}, expected 2` }
if (postSigned <= preSigned) return { pass: false, reason: `signedLength did not advance after swap (pre=${preSigned}, post=${postSigned})` }
return { pass: true }
}
main().catch(err => { console.error(err); process.exit(1) })