/**
* Peer-to-peer tests for the 0.5a tree layer.
*
* Wires two OverlayProtocol instances together via a Protomux over a
* duplex pipe, then exercises requestTree end-to-end.
*/
import test from 'brittle'
import Protomux from 'protomux'
import { Duplex } from 'streamx'
import { tmpStore } from './helpers.js'
import { FileIndex } from '../lib/file-index.js'
import { OverlayProtocol } from '../lib/protocol.js'
import { computeTreeHash, KIND_FILE, KIND_DIR } from '../lib/tree.js'
function makeDuplex () {
let aWrite, bWrite
const a = new Duplex({ write (data, cb) { bWrite(data); cb() }, read () {} })
const b = new Duplex({ write (data, cb) { aWrite(data); cb() }, read () {} })
aWrite = (data) => a.push(data)
bWrite = (data) => b.push(data)
return [a, b]
}
/**
* Minimal stub syncEngine — OverlayProtocol only needs _fileIndex,
* feedKey, feed (for _onOpen sync-state exchange), markSynced, logPut.
*/
function stubSyncEngine (fileIndex) {
return {
_fileIndex: fileIndex,
feedKey: '0'.repeat(64),
feed: { length: 0, async get () { return null } },
async markSynced () {},
async logPut () {}
}
}
/**
* Minimal stub transferManager — not used for tree-only tests.
*/
function stubTransferManager (fileIndex) {
return {
_fileIndex: fileIndex,
async prepareFile () { return null },
readChunk () { return null },
computeNeeded () { return [] },
startReceive () { return { partialPath: '', received: new Set(), total: 0 } },
writeChunk () { return { ok: false } },
isComplete () { return false },
finalize () { return { ok: false } },
cancel () {}
}
}
async function makePeer (label) {
const fileIndex = new FileIndex(tmpStore(label))
await fileIndex.ready()
const protocol = new OverlayProtocol(
stubSyncEngine(fileIndex),
stubTransferManager(fileIndex),
{ key: Buffer.alloc(32, 0xab), autoSync: false }
)
return { fileIndex, protocol }
}
function connect (peerA, peerB) {
const [sA, sB] = makeDuplex()
const muxA = Protomux.from(sA)
const muxB = Protomux.from(sB)
const pA = peerA.protocol.attach(muxA)
const pB = peerB.protocol.attach(muxB)
return { pA, pB }
}
// Give the channel time to open before sending (Protomux needs a tick).
const tick = () => new Promise((resolve) => setTimeout(resolve, 50))
// ── Tests ─────────────────────────────────────────────────────
test('requestTree — peer returns stored tree entries', async (t) => {
const A = await makePeer('tree-peer-a')
const B = await makePeer('tree-peer-b')
const entries = [
{ kind: KIND_FILE, exec: 0, name: 'index.js', childHash: 'a'.repeat(64), size: 100 },
{ kind: KIND_FILE, exec: 0, name: 'readme.md', childHash: 'b'.repeat(64), size: 500 }
]
const hash = computeTreeHash(entries)
await A.fileIndex.putTree(hash, { entries, size: 600 })
const { pB } = connect(A, B)
await tick()
const got = await B.protocol.requestTree(pB, hash, { timeout: 2000 })
t.is(got.length, 2)
t.is(got[0].name, 'index.js')
t.is(got[0].size, 100)
t.is(got[1].name, 'readme.md')
A.protocol.destroy()
B.protocol.destroy()
await A.fileIndex.close()
await B.fileIndex.close()
})
test('requestTree — peer responds with empty entries for unknown hash', async (t) => {
const A = await makePeer('tree-peer-miss-a')
const B = await makePeer('tree-peer-miss-b')
const { pB } = connect(A, B)
await tick()
const unknownHash = 'f'.repeat(64)
const got = await B.protocol.requestTree(pB, unknownHash, { timeout: 2000 })
t.alike(got, [], 'empty entries for unknown hash')
A.protocol.destroy()
B.protocol.destroy()
await A.fileIndex.close()
await B.fileIndex.close()
})
test('requestTree — times out when peer never responds', async (t) => {
const A = await makePeer('tree-peer-to-a')
const B = await makePeer('tree-peer-to-b')
const { pB } = connect(A, B)
await tick()
// Simulate an unresponsive peer: replace A's handler with a no-op.
const peerInA = [...A.protocol._peers.values()][0]
A.protocol._onTreeRequest = () => {}
// Clear the registered onmessage to silence requests at A.
peerInA.msgs.treeRequest.onmessage = () => {}
const hash = 'c'.repeat(64)
let err = null
try {
await B.protocol.requestTree(pB, hash, { timeout: 150 })
} catch (e) {
err = e
}
t.ok(err, 'request rejected on timeout')
t.ok(err.message.includes('timed out'), 'clear timeout message')
A.protocol.destroy()
B.protocol.destroy()
await A.fileIndex.close()
await B.fileIndex.close()
})
test('requestTree — concurrent requests for the same hash JOIN one in-flight request', async (t) => {
// Was: the second caller was rejected ("already pending"). A tree hash is
// content-addressed, so both callers are owed the same entries. pgh's
// materializer fetches up to 16 subtrees at once, and a repo with two
// identical directories (hyper-tps, 2026-10-11) failed the WHOLE install on
// that rejection.
const A = await makePeer('tree-peer-dup-a')
const B = await makePeer('tree-peer-dup-b')
const entries = [{ kind: KIND_FILE, exec: false, name: 'x.png', childHash: 'e'.repeat(64), size: 3 }]
const hash = computeTreeHash(entries)
await A.fileIndex.putTree(hash, { entries, size: 3 })
const { pB } = connect(A, B)
await tick()
let served = 0
const orig = A.protocol._onTreeRequest.bind(A.protocol)
A.protocol._onTreeRequest = (peer, msg) => { served++; return orig(peer, msg) }
const [r1, r2] = await Promise.all([
B.protocol.requestTree(pB, hash, { timeout: 2000 }),
B.protocol.requestTree(pB, hash, { timeout: 2000 })
])
t.alike(r1, r2, 'both callers get the same entries')
t.is(r1.length, 1)
t.is(served, 1, 'one request on the wire')
A.protocol.destroy()
B.protocol.destroy()
await A.fileIndex.close()
await B.fileIndex.close()
})
test('requestTree — handles nested directory structures', async (t) => {
const A = await makePeer('tree-peer-nest-a')
const B = await makePeer('tree-peer-nest-b')
// Leaf subtree
const libEntries = [
{ kind: KIND_FILE, exec: 0, name: 'util.js', childHash: '1'.repeat(64), size: 200 }
]
const libHash = computeTreeHash(libEntries)
await A.fileIndex.putTree(libHash, { entries: libEntries, size: 200 })
// Root tree pointing to the leaf
const rootEntries = [
{ kind: KIND_DIR, exec: 0, name: 'lib', childHash: libHash, size: 200 },
{ kind: KIND_FILE, exec: 0, name: 'index.js', childHash: '2'.repeat(64), size: 150 }
]
const rootHash = computeTreeHash(rootEntries)
await A.fileIndex.putTree(rootHash, { entries: rootEntries, size: 350 })
const { pB } = connect(A, B)
await tick()
const root = await B.protocol.requestTree(pB, rootHash, { timeout: 2000 })
t.is(root.length, 2)
const libEntry = root.find(e => e.name === 'lib')
t.ok(libEntry, 'root has lib entry')
t.is(libEntry.kind, KIND_DIR)
t.is(libEntry.childHash, libHash)
// Walk down: fetch the subtree
const lib = await B.protocol.requestTree(pB, libHash, { timeout: 2000 })
t.is(lib.length, 1)
t.is(lib[0].name, 'util.js')
A.protocol.destroy()
B.protocol.destroy()
await A.fileIndex.close()
await B.fileIndex.close()
})