#!/usr/bin/env bare
/**
* Integration test: reproduce Autobase corruption from poisoned addWriter
* and verify the drain loop patch prevents infinite hang.
*
* Uses autobase-test-helpers for replication (same as upstream tests).
*/
import test from 'brittle'
import Corestore from 'corestore'
import Autobase from 'autobase'
import Hyperbee from 'hyperbee'
import crypto from 'hypercore-crypto'
import os from 'bare-os'
import path from 'bare-path'
import fs from 'bare-fs'
import { replicate, replicateAndSync } from 'autobase-test-helpers'
const tmpdir = os.tmpdir()
let idx = 0
function tmpPath (name) {
return path.join(tmpdir, 'ab-poison-' + name + '-' + (idx++) + '-' + Date.now())
}
function open (store) {
return new Hyperbee(store.get('view', { valueEncoding: 'json' }), {
keyEncoding: 'utf-8',
valueEncoding: 'json'
})
}
async function cleanup (...dirs) {
for (const d of dirs) await fs.promises.rm(d, { recursive: true }).catch(() => {})
}
// Apply that validates before calling host.addWriter (our fix)
async function safeApply (nodes, view, host) {
const batch = view.batch()
for (const node of nodes) {
const op = node.value
if (op.type === 'put') {
await batch.put(op.key, op.value)
}
if (op.type === 'add-writer' && op.writerKey) {
const writerKey = Buffer.from(op.writerKey, 'hex')
try {
const probe = host.base.store.get({ key: writerKey, active: false })
await probe.ready()
const valid = probe.manifest !== null || probe.length > 0
await probe.close()
if (valid) {
await host.addWriter(writerKey, { indexer: false })
}
} catch {}
}
}
await batch.flush()
}
// Apply that blindly calls host.addWriter (the old broken behavior)
async function unsafeApply (nodes, view, host) {
const batch = view.batch()
for (const node of nodes) {
const op = node.value
if (op.type === 'put') {
await batch.put(op.key, op.value)
}
if (op.type === 'add-writer' && op.writerKey) {
await host.addWriter(Buffer.from(op.writerKey, 'hex'), { indexer: false })
}
}
await batch.flush()
}
function createBase (store, key, apply) {
return new Autobase(store.session(), key, {
open,
apply,
valueEncoding: 'json',
ackInterval: 0,
ackThreshold: 0,
fastForward: false
})
}
// ─────────────────────────────────────────────────────────────
test('baseline: two peers replicate and sync', async (t) => {
t.timeout(15000)
const dirA = tmpPath('a')
const dirB = tmpPath('b')
const storeA = new Corestore(dirA, { primaryKey: Buffer.alloc(32, 0), unsafe: true })
const storeB = new Corestore(dirB, { primaryKey: Buffer.alloc(32, 1), unsafe: true })
const a = createBase(storeA, null, safeApply)
await a.ready()
const b = createBase(storeB, a.local.key, safeApply)
await b.ready()
await a.append({ type: 'put', key: 'msg:1', value: { text: 'hello' } })
await replicateAndSync([a, b])
const entry = await b.view.get('msg:1')
t.ok(entry, 'replicated')
t.is(entry.value.text, 'hello')
await a.close()
await b.close()
await storeA.close()
await storeB.close()
await cleanup(dirA, dirB)
})
// ─────────────────────────────────────────────────────────────
test('safe apply: poisoned add-writer does not hang on replication', async (t) => {
t.timeout(15000)
const dirA = tmpPath('a')
const dirB = tmpPath('b')
const storeA = new Corestore(dirA, { primaryKey: Buffer.alloc(32, 0), unsafe: true })
const storeB = new Corestore(dirB, { primaryKey: Buffer.alloc(32, 1), unsafe: true })
const a = createBase(storeA, null, safeApply)
await a.ready()
const b = createBase(storeB, a.local.key, safeApply)
await b.ready()
// Write data + poisoned add-writer + more data
await a.append({ type: 'put', key: 'msg:1', value: { text: 'before' } })
const bogusKey = crypto.randomBytes(32).toString('hex')
await a.append({ type: 'add-writer', writerKey: bogusKey })
await a.append({ type: 'put', key: 'msg:2', value: { text: 'after' } })
await replicateAndSync([a, b])
const e1 = await b.view.get('msg:1')
const e2 = await b.view.get('msg:2')
t.ok(e1, 'msg:1 replicated')
t.ok(e2, 'msg:2 replicated past poison')
t.is(e1.value.text, 'before')
t.is(e2.value.text, 'after')
await a.close()
await b.close()
await storeA.close()
await storeB.close()
await cleanup(dirA, dirB)
})
// ─────────────────────────────────────────────────────────────
test('unsafe apply: poisoned add-writer — reopen must not hang', async (t) => {
t.timeout(15000)
const dirA = tmpPath('a')
const storeA = new Corestore(dirA, { primaryKey: Buffer.alloc(32, 0), unsafe: true })
const a = createBase(storeA, null, unsafeApply)
await a.ready()
const key = a.local.key
// Poison it
await a.append({ type: 'put', key: 'msg:1', value: { text: 'before' } })
const bogusKey = crypto.randomBytes(32).toString('hex')
await a.append({ type: 'add-writer', writerKey: bogusKey })
await a.update()
await a.close()
await storeA.close()
// Reopen — this is where the drain loop used to hang
const storeA2 = new Corestore(dirA, { primaryKey: Buffer.alloc(32, 0), unsafe: true })
const a2 = createBase(storeA2, key, unsafeApply)
await a2.ready()
t.comment('writable: ' + a2.writable)
const timer = setTimeout(() => {
t.fail('update() hung for 10 seconds')
t.end()
}, 10000)
await a2.update()
clearTimeout(timer)
t.pass('update() did not hang on reopen')
await a2.close()
await storeA2.close()
await cleanup(dirA)
})
// ─────────────────────────────────────────────────────────────
test('unsafe apply: poisoned writer replicated to peer — peer must not hang', async (t) => {
t.timeout(15000)
const dirA = tmpPath('a')
const dirB = tmpPath('b')
const storeA = new Corestore(dirA, { primaryKey: Buffer.alloc(32, 0), unsafe: true })
const storeB = new Corestore(dirB, { primaryKey: Buffer.alloc(32, 1), unsafe: true })
const a = createBase(storeA, null, unsafeApply)
await a.ready()
const b = createBase(storeB, a.local.key, unsafeApply)
await b.ready()
// Write data + poison
await a.append({ type: 'put', key: 'msg:1', value: { text: 'hello' } })
const bogusKey = crypto.randomBytes(32).toString('hex')
await a.append({ type: 'add-writer', writerKey: bogusKey })
// Replicate — peer B receives the poison
const done = replicate([a, b])
const timer = setTimeout(() => {
t.fail('sync hung for 10 seconds')
t.end()
}, 10000)
// Try to update B — should not hang with the drain loop patch
for (let i = 0; i < 5; i++) {
await b.update()
await new Promise(r => setTimeout(r, 200))
}
clearTimeout(timer)
t.pass('peer B update() did not hang despite receiving poisoned writer')
await done()
await a.close()
await b.close()
await storeA.close()
await storeB.close()
await cleanup(dirA, dirB)
})