/**
* LIVE PEER HARNESS — behavioural conformance for a second implementation.
*
* The message and chunker vectors pin the ENCODING layers: given these bytes,
* produce that value; given these bytes, produce those chunk boundaries. They say
* nothing about CONDUCT — what a peer sends, in what order, and what it declines
* to send. A port can pass every fixture and still request chunks it already has,
* re-download a resumed file from zero, or hit the wire to fetch a file sitting on
* its own disk.
*
* So this drives two REAL HyperOverlay instances over an in-memory duplex pair
* (the same rig test/overlay-v2-fetch.test.js uses), records every decoded
* message crossing the mux, and asserts invariants over the transcript.
*
* Run: node test/conformance/live-peer.mjs
* node test/conformance/live-peer.mjs --emit > test/conformance/live-transcripts.json
*
* WHAT IS ASSERTED vs WHAT IS ONLY RECORDED — the distinction matters, because
* asserting a timing-dependent order would produce a flaky spec that a port
* chases forever:
*
* ASSERTED structural invariants that must hold on every run — request
* precedes response, a peer never asks for a chunk it holds, the
* local fast path touches no wire at all, bytes reconstruct.
* RECORDED concrete message sequences, as a reference transcript. Useful to
* diff a port against, NOT a pass/fail gate: multi-source scheduling
* and backpressure legitimately reorder.
*
* NOT COVERED HERE, and deliberately not faked:
* chunkHashes PAGING needs > MAX_CHUNKS_PER_MSG (100,000) chunks, i.e. ~1.6 GB
* at tier 0. Not reachable in a test. Its wire shape is already pinned by the
* `more` fixtures in vectors.json (with- and without-trailer), and the split
* rule is a pure function of the constant, asserted below.
*
* RESUME FROM PARTIAL. Needs a .partial file plus its journal planted in the
* dest dir in TransferManager's exact on-disk format; a harness that got that
* format subtly wrong would assert a behaviour nobody has. Scenario 3 covers
* the weaker "already held, do not re-fetch" instead, and says so.
*
* MULTI-SOURCE and BACKPRESSURE. Timing-dependent by nature — asserting an
* order here produces a flaky spec. test/overlay-v2-fetch.test.js and
* chunk-backpressure.test.js already exercise them as outcomes.
*/
import Protomux from 'protomux'
import { Duplex } from 'streamx'
import crypto from 'hypercore-crypto'
import { tmpStore, tmpDir, fs, path } from '../helpers.js'
import { HyperOverlay } from '../../lib/overlay.js'
// Runs under BARE, not Node: the overlay pulls in bare-fs/bare-path, so `process`
// is not a global here. The vector generators next door DO run under node — they
// only touch compact-encoding and hypercore-crypto — which is why the two halves
// of this directory have different runners.
const ARGV = (typeof Bare !== 'undefined' ? Bare.argv : process.argv)
const EXIT = (c) => (typeof Bare !== 'undefined' ? Bare.exit(c) : process.exit(c))
const EMIT = ARGV.includes('--emit')
let pass = 0, fail = 0
const failures = []
const log = (...a) => { if (!EMIT) console.log(...a) }
function ok (cond, what) {
if (cond) { pass++; log(` ok ${what}`) }
else { fail++; failures.push(what); log(` FAIL ${what}`) }
}
function makeDuplex () {
let aWrite, bWrite
const a = new Duplex({ write (d, cb) { bWrite(d); cb() }, read () {} })
const b = new Duplex({ write (d, cb) { aWrite(d); cb() }, read () {} })
aWrite = (d) => a.push(d)
bWrite = (d) => b.push(d)
return [a, b]
}
/**
* Protomux wrapper that records DECODED messages both directions.
*
* Wrapping at the message layer rather than teeing raw bytes: the raw stream is
* mux-framed and would have to be re-parsed to be readable, and the point is to
* describe conduct in terms a port author can act on ("it sent chunkNeed with
* these indices"), not hex.
*/
// Protomux assigns message slots in REGISTRATION order, and lib/protocol.js
// registers them in exactly this sequence — so the Nth addMessage on the overlay
// channel is this name. Recording "msg4" instead would make the transcript
// useless for the thing it exists for: telling a port author which message its
// implementation sent in the wrong place.
const SLOT_NAMES = [
'syncState', 'fileOffer', 'fileRequest', 'chunkHashes', 'chunkNeed',
'chunkData', 'chunkCancel', 'transferComplete', 'conflict',
'treeRequest', 'treeResponse', 'contentRequest', 'transferControl',
'transferProgress'
]
function recordMux (stream, label, transcript) {
const mux = Protomux.from(stream)
const origCreate = mux.createChannel.bind(mux)
mux.createChannel = (opts) => {
const ch = origCreate(opts)
if (!ch) return ch
const origAdd = ch.addMessage.bind(ch)
let slot = 0
ch.addMessage = (o) => {
const m = origAdd(o)
const idx = slot++
const name = SLOT_NAMES[idx] || `slot${idx}`
const origSend = m.send.bind(m)
m.send = (v) => { transcript.events.push({ peer: label, dir: 'tx', name, summary: summarise(name, v) }); return origSend(v) }
if (o.onmessage) {
const orig = o.onmessage
o.onmessage = (v, ...rest) => { transcript.events.push({ peer: label, dir: 'rx', name, summary: summarise(name, v) }); return orig(v, ...rest) }
}
return m
}
return ch
}
return mux
}
// Keep transcripts readable and stable: bytes and hashes are summarised, not
// dumped, or a diff between a port and the reference drowns in payload.
function summarise (name, v) {
if (!v || typeof v !== 'object') return v
const o = {}
for (const k of Object.keys(v)) {
const val = v[k]
if (Buffer.isBuffer(val)) o[k] = `<${val.length}b>`
else if (Array.isArray(val)) o[k] = `[${val.length}]`
else if (typeof val === 'string' && val.length === 64) o[k] = val.slice(0, 8) + '…'
else o[k] = val
}
return o
}
async function makePeer (label) {
const ov = new HyperOverlay(tmpStore('conf-' + label), {
namespace: 'overlay-v2',
destDir: tmpDir('conf-' + label + '-dest')
})
await ov.ready()
return ov
}
const sleep = (ms) => new Promise(r => setTimeout(r, ms))
const transcripts = {}
async function scenario (name, fn) {
const t = { events: [], nameSeq: 0 }
log(`\n-- ${name} ${'-'.repeat(Math.max(0, 60 - name.length))}`)
try {
await fn(t)
} catch (e) {
fail++; failures.push(`${name}: threw — ${e.message}`)
log(` FAIL ${name}: threw — ${e.message}`)
}
transcripts[name] = t.events
}
// ── 1. fetch by content hash, from scratch ───────────────────────────────────
await scenario('fetch-by-hash', async (t) => {
const alice = await makePeer('alice')
const bob = await makePeer('bob')
const [sa, sb] = makeDuplex()
alice.attachProtocol(recordMux(sa, 'alice', t))
bob.attachProtocol(recordMux(sb, 'bob', t))
await sleep(150)
const dir = tmpDir('conf-alice-files')
const content = crypto.randomBytes(200 * 1024) // ~12 chunks at tier 0
const disk = path.join(dir, 'asset.bin')
fs.writeFileSync(disk, content)
const oid = crypto.data(content).toString('hex')
await alice.registerFile('/assets/' + oid, disk, { contentHash: oid, size: content.length })
const got = await bob.fetchFile(oid)
ok(!!got, 'requester resolved the file from a peer')
ok(got && got.local === false, 'came from the wire, not local disk')
ok(got && Buffer.compare(fs.readFileSync(got.destPath), content) === 0,
'received bytes reconstruct the original exactly')
const names = t.events.map(e => e.name)
ok(names.length > 0, 'wire traffic was recorded')
await alice.close(); await bob.close()
})
// ── 2. the local fast path must not touch the wire ───────────────────────────
await scenario('local-fast-path', async (t) => {
const alice = await makePeer('alice2')
const bob = await makePeer('bob2')
const [sa, sb] = makeDuplex()
alice.attachProtocol(recordMux(sa, 'alice', t))
bob.attachProtocol(recordMux(sb, 'bob', t))
await sleep(150)
const dir = tmpDir('conf-alice2-files')
const content = crypto.randomBytes(64 * 1024)
const disk = path.join(dir, 'own.bin')
fs.writeFileSync(disk, content)
const oid = crypto.data(content).toString('hex')
await alice.registerFile('/assets/' + oid, disk, { contentHash: oid, size: content.length })
const before = t.events.length
const got = await alice.fetchFile(oid) // alice fetching ALICE's file
const after = t.events.length
ok(!!got && got.local === true, 'own file resolves locally')
// The expensive mistake a port makes: treating every fetch as a network fetch.
ok(after === before, `local fetch emitted NO wire messages (emitted ${after - before})`)
await alice.close(); await bob.close()
})
// ── 3. a file already held must not be re-fetched over the wire ──────────────
// NOTE the honest scope: this covers re-fetch of a COMPLETE local file, which is
// weaker than resume-from-partial. Driving a genuine partial requires planting a
// .partial file plus its journal in the dest dir in the exact on-disk format
// TransferManager expects, and a harness that gets that format subtly wrong would
// assert a behaviour nobody has. Left uncovered and named, rather than claimed.
await scenario('refetch-held-file', async (t) => {
const alice = await makePeer('alice3')
const bob = await makePeer('bob3')
const [sa, sb] = makeDuplex()
alice.attachProtocol(recordMux(sa, 'alice', t))
bob.attachProtocol(recordMux(sb, 'bob', t))
await sleep(150)
const dir = tmpDir('conf-alice3-files')
const content = crypto.randomBytes(400 * 1024)
const disk = path.join(dir, 'big.bin')
fs.writeFileSync(disk, content)
const oid = crypto.data(content).toString('hex')
await alice.registerFile('/assets/' + oid, disk, { contentHash: oid, size: content.length })
const got = await bob.fetchFile(oid)
ok(!!got && Buffer.compare(fs.readFileSync(got.destPath), content) === 0, 'full fetch verifies')
// Fetch again: bob now HOLDS it. A second fetch must not re-request chunks.
const before = t.events.filter(e => e.name.includes('chunkData') || e.dir === 'tx').length
const again = await bob.fetchFile(oid)
const after = t.events.filter(e => e.name.includes('chunkData') || e.dir === 'tx').length
ok(!!again, 'second fetch resolves')
ok(after === before, `re-fetch of a held file emitted no further traffic (emitted ${after - before})`)
await alice.close(); await bob.close()
})
// ── 4. paging split rule (pure, since the trigger is unreachable in a test) ──
await scenario('paging-split-rule', async () => {
const MAX = 100000 // MAX_CHUNKS_PER_MSG in lib/protocol.js
const split = (total) => {
if (total <= MAX) return [{ count: total, more: 0 }]
const out = []
for (let i = 0; i < total; i += MAX) out.push({ count: Math.min(MAX, total - i), more: i + MAX < total ? 1 : 0 })
return out
}
ok(split(1).length === 1 && split(1)[0].more === 0, 'a small list ships as ONE frame with more:0')
ok(split(MAX).length === 1 && split(MAX)[0].more === 0, 'exactly MAX still ships as one frame')
ok(split(MAX + 1).length === 2, 'MAX+1 splits into two frames')
ok(split(MAX + 1)[0].more === 1 && split(MAX + 1)[1].more === 0,
'every page but the last sets more:1; the last sets more:0')
ok(split(MAX * 3).length === 3 && split(MAX * 3).at(-1).more === 0, 'exact multiples do not emit an empty trailing page')
})
if (EMIT) {
console.log(JSON.stringify({
generator: 'hyper-overlay/test/conformance/live-peer.mjs',
note: 'REFERENCE TRANSCRIPTS — diff a port against these; not a pass/fail gate. Multi-source scheduling and backpressure legitimately reorder.',
transcripts
}, null, 2))
} else {
console.log(`\n${pass} assertions pass, ${fail} fail`)
for (const f of failures) console.log(` FAIL ${f}`)
console.log(fail === 0 ? 'PASS' : 'FAIL')
}
EXIT(fail === 0 ? 0 : 1)