#!/usr/bin/env bare
/**
* E2E Test: Two-peer file sync over in-memory Protomux streams
*
* Creates two overlay instances (alice + bob) connected via Duplex streams.
* Tests: file transfer, multi-chunk, dedup detection, conflict resolution.
*
* Usage: bare test/e2e-two-peers.js
*/
import Corestore from 'corestore'
import Protomux from 'protomux'
import { Duplex } from 'streamx'
import crypto from 'hypercore-crypto'
import fs from 'bare-fs'
import path from 'bare-path'
import os from 'bare-os'
import process from 'bare-process'
import { FileIndex } from '../lib/file-index.js'
import { SyncEngine } from '../lib/sync-engine.js'
import { TransferManager } from '../lib/transfer.js'
import { OverlayProtocolV2 } from '../lib/protocol-v2.js'
import { hashChunk } from '../lib/chunker.js'
const tmpBase = path.join(os.tmpdir(), 'overlay-e2e-' + Date.now())
function log (label, ...args) {
console.log(`[${label}]`, ...args)
}
function mkTmp (name) {
const dir = path.join(tmpBase, name)
fs.mkdirSync(dir, { recursive: true })
return dir
}
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]
}
async function createPeer (name) {
const storageDir = mkTmp(name + '-store')
const filesDir = mkTmp(name + '-files')
const store = new Corestore(storageDir)
const ns = store.namespace(name)
const index = new FileIndex(ns)
await index.ready()
const syncEngine = new SyncEngine(index, store)
await syncEngine.ready()
const transfer = new TransferManager(index)
const filePaths = new Map()
const protocol = new OverlayProtocolV2(syncEngine, transfer, { filePaths })
return { name, store, index, syncEngine, transfer, protocol, filePaths, filesDir }
}
async function cleanup (peers) {
for (const peer of peers) {
peer.protocol.destroy()
await peer.syncEngine.close()
await peer.index.close()
await peer.store.close()
}
}
let passed = 0
let failed = 0
function pass (msg) {
passed++
log('test', 'PASS:', msg)
}
function fail (msg) {
failed++
console.error('[test] FAIL:', msg)
}
// ── Main ──────────────────────────────────────────────────────
async function main () {
console.log('=== hyper-overlay e2e: two-peer sync ===\n')
const alice = await createPeer('alice')
const bob = await createPeer('bob')
// Connect via in-memory Protomux streams
const [streamA, streamB] = makeDuplex()
const muxA = Protomux.from(streamA)
const muxB = Protomux.from(streamB)
alice.protocol.attach(muxA)
bob.protocol.attach(muxB)
// Wait for protocol handshake
await new Promise(resolve => setTimeout(resolve, 100))
log('test', 'peers connected via Protomux\n')
// ── Test 1: Small file transfer ────────────────────────────
console.log('--- Test 1: Small file transfer ---')
const testContent = Buffer.from('Hello from Alice! P2P transfer test ' + Date.now())
const testFile = path.join(alice.filesDir, 'hello.txt')
fs.writeFileSync(testFile, testContent)
const prepared = await alice.transfer.prepareFile(testFile, '/hello.txt')
if (!prepared) { fail('prepareFile returned null'); return }
log('alice', prepared.chunks.length, 'chunks,', prepared.size, 'bytes')
await alice.index.putFile('/hello.txt', {
contentHash: hashChunk(testContent),
size: testContent.length,
mtime: fs.statSync(testFile).mtimeMs
})
alice.filePaths.set('/hello.txt', testFile)
// Bob receives
const bobTarget = path.join(bob.filesDir, 'hello.txt')
bob.transfer.startReceive(bobTarget, { size: prepared.size, chunks: prepared.chunks })
for (let i = 0; i < prepared.chunks.length; i++) {
const c = prepared.chunks[i]
const data = alice.transfer.readChunk(testFile, c.offset, c.length)
const result = bob.transfer.writeChunk(bobTarget, i, data)
if (!result.ok) { fail('chunk write: ' + result.error); return }
}
const fin = bob.transfer.finalize(bobTarget)
if (!fin.ok) { fail('finalize: ' + fin.error); return }
const received = fs.readFileSync(bobTarget)
if (Buffer.compare(received, testContent) === 0) {
pass('small file content matches (' + received.length + ' bytes)')
} else {
fail('content mismatch: expected ' + testContent.length + ', got ' + received.length)
}
// ── Test 2: Multi-chunk file ───────────────────────────────
console.log('\n--- Test 2: Multi-chunk file (128KB) ---')
const largeContent = crypto.randomBytes(128 * 1024)
const largeFile = path.join(alice.filesDir, 'large.bin')
fs.writeFileSync(largeFile, largeContent)
const largePrepared = await alice.transfer.prepareFile(largeFile, '/large.bin')
log('alice', largePrepared.chunks.length, 'chunks,', largePrepared.size, 'bytes')
const bobLargeTarget = path.join(bob.filesDir, 'large.bin')
bob.transfer.startReceive(bobLargeTarget, { size: largePrepared.size, chunks: largePrepared.chunks })
for (let i = 0; i < largePrepared.chunks.length; i++) {
const c = largePrepared.chunks[i]
const data = alice.transfer.readChunk(largeFile, c.offset, c.length)
const result = bob.transfer.writeChunk(bobLargeTarget, i, data)
if (!result.ok) { fail('chunk ' + i + ': ' + result.error); return }
}
bob.transfer.finalize(bobLargeTarget)
const receivedLarge = fs.readFileSync(bobLargeTarget)
if (Buffer.compare(receivedLarge, largeContent) === 0) {
pass('multi-chunk file matches (' + largePrepared.chunks.length + ' chunks)')
} else {
fail('large file mismatch')
}
// ── Test 3: Out-of-order chunk delivery ────────────────────
console.log('\n--- Test 3: Out-of-order chunks ---')
const oooContent = crypto.randomBytes(96 * 1024)
const oooFile = path.join(alice.filesDir, 'ooo.bin')
fs.writeFileSync(oooFile, oooContent)
const oooPrepared = await alice.transfer.prepareFile(oooFile, '/ooo.bin')
const bobOooTarget = path.join(bob.filesDir, 'ooo.bin')
bob.transfer.startReceive(bobOooTarget, { size: oooPrepared.size, chunks: oooPrepared.chunks })
// Send in reverse order
for (let i = oooPrepared.chunks.length - 1; i >= 0; i--) {
const c = oooPrepared.chunks[i]
const data = alice.transfer.readChunk(oooFile, c.offset, c.length)
bob.transfer.writeChunk(bobOooTarget, i, data)
}
bob.transfer.finalize(bobOooTarget)
const receivedOoo = fs.readFileSync(bobOooTarget)
if (Buffer.compare(receivedOoo, oooContent) === 0) {
pass('out-of-order delivery works (' + oooPrepared.chunks.length + ' chunks)')
} else {
fail('out-of-order mismatch')
}
// ── Test 4: Dedup detection ────────────────────────────────
console.log('\n--- Test 4: Dedup detection ---')
const dupFile = path.join(alice.filesDir, 'dup.bin')
fs.writeFileSync(dupFile, largeContent) // same content as large.bin
const dupPrepared = await alice.transfer.prepareFile(dupFile, '/dup.bin')
const origHashes = new Set(largePrepared.chunks.map(c => c.hash))
const needed = alice.transfer.computeNeeded(dupPrepared.chunks, origHashes)
if (needed.length === 0) {
pass('dedup detected — 0 new chunks for identical content (' + dupPrepared.chunks.length + ' matched)')
} else {
fail('dedup missed ' + needed.length + ' chunks')
}
// ── Test 5: Hash verification rejects bad data ─────────────
console.log('\n--- Test 5: Hash verification ---')
const verifyTarget = path.join(bob.filesDir, 'verify.bin')
bob.transfer.startReceive(verifyTarget, {
size: 10,
chunks: [{ hash: 'a'.repeat(64), offset: 0, length: 10 }]
})
const badResult = bob.transfer.writeChunk(verifyTarget, 0, Buffer.from('wrong data'))
if (!badResult.ok && badResult.error.includes('hash mismatch')) {
pass('tampered chunk rejected')
} else {
fail('tampered chunk was accepted')
}
bob.transfer.cancel(verifyTarget)
// ── Test 6: Sync engine conflict detection ─────────────────
console.log('\n--- Test 6: Conflict detection ---')
await alice.syncEngine.logPut('/shared.txt', { contentHash: 'alice-v', size: 100, mtime: 1000 })
await bob.syncEngine.logPut('/shared.txt', { contentHash: 'bob-v', size: 200, mtime: 2000 })
const aliceKey = alice.syncEngine.feedKey
const bobKey = bob.syncEngine.feedKey
// Cross-reference feeds
alice.syncEngine._peerFeeds.set(bobKey, bob.syncEngine.feed)
bob.syncEngine._peerFeeds.set(aliceKey, alice.syncEngine.feed)
// Set ancestor
await alice.index.putSyncState(bobKey, '/shared.txt', { lastSeq: 0, lastHash: 'original' })
const diff = await alice.syncEngine.diff(bobKey)
const conflicts = diff.filter(a => a.action === 'conflict')
if (conflicts.find(c => c.path === '/shared.txt')) {
pass('conflict detected for /shared.txt')
} else {
fail('no conflict detected, got: ' + diff.map(a => a.path + ':' + a.action).join(', '))
}
// ── Test 7: Conflict resolution strategies ─────────────────
console.log('\n--- Test 7: Conflict resolution ---')
const conflict = conflicts.find(c => c.path === '/shared.txt')
const mineWins = alice.syncEngine.resolveConflict(conflict, 'mine-wins')
if (mineWins.action === 'push' && mineWins.winner === 'local') {
pass('mine-wins → push local')
} else {
fail('mine-wins wrong: ' + JSON.stringify(mineWins))
}
const theirsWins = alice.syncEngine.resolveConflict(conflict, 'theirs-wins')
if (theirsWins.action === 'pull' && theirsWins.winner === 'remote') {
pass('theirs-wins → pull remote')
} else {
fail('theirs-wins wrong')
}
const keepBoth = alice.syncEngine.resolveConflict(conflict, 'keep-both')
if (keepBoth.action === 'keep-both') {
pass('keep-both works')
} else {
fail('keep-both wrong')
}
// ── Test 8: Partial file safety ────────────────────────────
console.log('\n--- Test 8: Partial file safety ---')
const partialTarget = path.join(bob.filesDir, 'partial.bin')
const partialState = bob.transfer.startReceive(partialTarget, {
size: 200,
chunks: [
{ hash: 'a'.repeat(64), offset: 0, length: 100 },
{ hash: 'b'.repeat(64), offset: 100, length: 100 }
]
})
if (fs.existsSync(partialState.partialPath)) {
pass('partial file created at ' + path.basename(partialState.partialPath))
} else {
fail('no partial file')
}
// Finalize should reject incomplete
const incompleteFin = bob.transfer.finalize(partialTarget)
if (!incompleteFin.ok) {
pass('incomplete transfer rejected')
} else {
fail('incomplete transfer was finalized')
}
// Cancel should clean up
bob.transfer.cancel(partialTarget)
if (!fs.existsSync(partialState.partialPath)) {
pass('partial file cleaned up on cancel')
} else {
fail('partial file not cleaned up')
}
// ── Test 9: No blob storage anywhere ───────────────────────
console.log('\n--- Test 9: Zero blob storage ---')
const aliceFile = await alice.index.getFile('/hello.txt')
const aliceChunkMap = await alice.index.getChunkMap('/hello.txt')
let blobFound = false
if (aliceFile && aliceFile.data) blobFound = true
if (aliceChunkMap) {
for (const c of aliceChunkMap) {
if (c.data) blobFound = true
}
}
if (!blobFound) {
pass('no blob data in file index')
} else {
fail('blob data found in index')
}
// ── Summary ────────────────────────────────────────────────
console.log('\n=== Results: ' + passed + ' passed, ' + failed + ' failed ===')
await cleanup([alice, bob])
if (failed > 0) process.exit(1)
process.exit(0)
}
main().catch(err => {
console.error('E2E FAILED:', err)
process.exit(1)
})