/**
* hyper-deploy Phase 1: Local release distribution test
*
* Proves hyper-overlay's primitives can distribute build artifacts
* between two peers over Protomux. Uses SecretStream pairs (no network).
*
* This is a consumer of hyper-overlay — no modifications to hyper-overlay.
* The orchestration logic lives here in hyper-deploy.
*/
import test from 'brittle'
import Corestore from 'corestore'
import Protomux from 'protomux'
import c from 'compact-encoding'
import SecretStream from '@hyperswarm/secret-stream'
import fs from 'bare-fs'
import path from 'bare-path'
import os from 'bare-os'
import crypto from 'hypercore-crypto'
// hyper-overlay as dependency (not modified)
import { FileIndex } from 'hyper-overlay/file-index'
import { TransferManager } from 'hyper-overlay/transfer'
import { chunk, hashChunk } from 'hyper-overlay/chunker'
const tmpdir = os.tmpdir()
const PARTIAL_SUFFIX = '.overlay-partial'
// --- Helpers ---
function makePair () {
const a = new SecretStream(true)
const b = new SecretStream(false)
a.rawStream.pipe(b.rawStream).pipe(a.rawStream)
return [a, b]
}
function mktemp (prefix) {
const dir = path.join(tmpdir, `hyper-deploy-${prefix}-${Date.now()}`)
fs.mkdirSync(dir, { recursive: true })
return dir
}
/**
* Walk a directory recursively, return relative paths
*/
function walkDir (dir, prefix) {
prefix = prefix || ''
const results = []
for (const entry of fs.readdirSync(dir)) {
const full = path.join(dir, entry)
const rel = prefix ? prefix + '/' + entry : entry
const stat = fs.statSync(full)
if (stat.isDirectory()) {
results.push(...walkDir(full, rel))
} else {
results.push({ path: rel, size: stat.size, mtime: stat.mtimeMs })
}
}
return results
}
/**
* Compute content hash for a file
*/
function hashFile (filePath) {
const data = fs.readFileSync(filePath)
return crypto.data(data).toString('hex')
}
/**
* Index a directory: walk, hash, store in FileIndex + prepare chunk maps
*/
async function indexDirectory (dir, fileIndex, transferManager) {
const files = walkDir(dir)
let totalSize = 0
for (const file of files) {
const fullPath = path.join(dir, file.path)
const contentHash = hashFile(fullPath)
await fileIndex.putFile(file.path, {
contentHash,
size: file.size,
mtime: file.mtime
})
totalSize += file.size
}
return { fileCount: files.length, totalSize }
}
/**
* Release distribution over a single Protomux channel.
*
* Protocol (custom, minimal):
* Message 0 (fileList): server → client, JSON array of { path, contentHash, size }
* Message 1 (fileReq): client → server, JSON { path }
* Message 2 (chunkInfo): server → client, JSON { path, tier, chunks: [{hash, offset, length}] }
* Message 3 (chunkData): server → client, raw bytes prefixed with path+index header
* Message 4 (control): bidirectional, JSON { type, ... }
*/
const DEPLOY_PROTOCOL = 'hyper-deploy'
// === Tests ===
test('release distribution: small files', async (t) => {
t.timeout(30000)
// Create test release directory
const srcDir = mktemp('src')
for (let i = 0; i < 10; i++) {
fs.writeFileSync(path.join(srcDir, `file_${i}.txt`), `content for file ${i}\n`)
}
const dstDir = mktemp('dst')
// Setup stores
const storeA = new Corestore(mktemp('store-a'))
const storeB = new Corestore(mktemp('store-b'))
const fileIndexA = new FileIndex(storeA)
const fileIndexB = new FileIndex(storeB)
const transferA = new TransferManager(fileIndexA)
const transferB = new TransferManager(fileIndexB)
await fileIndexA.ready()
await fileIndexB.ready()
// Index source directory
const { fileCount, totalSize } = await indexDirectory(srcDir, fileIndexA, transferA)
t.is(fileCount, 10, 'indexed 10 files')
// Connect via SecretStream
const [sa, sb] = makePair()
storeA.replicate(sa)
storeB.replicate(sb)
const muxA = Protomux.from(sa)
const muxB = Protomux.from(sb)
// Transfer all files through Protomux
const transferDone = new Promise(async (resolve, reject) => {
const chA = muxA.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('test') })
const chB = muxB.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('test') })
const fileListMsg = chA.addMessage({ encoding: c.raw })
const fileReqMsg = chA.addMessage({ encoding: c.raw })
const chunkInfoMsg = chA.addMessage({ encoding: c.raw })
const chunkDataMsg = chA.addMessage({ encoding: c.raw })
const fileListMsgB = chB.addMessage({ encoding: c.raw })
const fileReqMsgB = chB.addMessage({ encoding: c.raw })
const chunkInfoMsgB = chB.addMessage({ encoding: c.raw })
const chunkDataMsgB = chB.addMessage({ encoding: c.raw })
let filesCompleted = 0
let bytesTransferred = 0
let pendingFiles = []
// CLIENT (B): receive file list, request each file
fileListMsgB.onmessage = (data) => {
const files = JSON.parse(data.toString())
pendingFiles = files
// Request first file
if (files.length > 0) {
fileReqMsgB.send(Buffer.from(JSON.stringify({ path: files[0].path })))
}
}
// CLIENT (B): receive chunk info, then receive chunks
let currentChunks = null
let currentPath = null
chunkInfoMsgB.onmessage = (data) => {
const info = JSON.parse(data.toString())
currentPath = info.path
currentChunks = info.chunks
const targetPath = path.join(dstDir, currentPath)
transferB.startReceive(targetPath, { size: info.size, chunks: currentChunks })
}
// CLIENT (B): receive chunk data
chunkDataMsgB.onmessage = (data) => {
// Header: 4 bytes index (uint32 LE) + rest is chunk data
const index = data.readUInt32LE(0)
const chunkData = data.subarray(4)
const targetPath = path.join(dstDir, currentPath)
const result = transferB.writeChunk(targetPath, index, chunkData)
if (!result.ok) {
reject(new Error(`writeChunk failed: ${result.error}`))
return
}
bytesTransferred += chunkData.length
if (transferB.isComplete(targetPath)) {
const fin = transferB.finalize(targetPath)
if (!fin.ok) {
reject(new Error(`finalize failed: ${fin.error}`))
return
}
filesCompleted++
// Request next file
if (filesCompleted < pendingFiles.length) {
fileReqMsgB.send(Buffer.from(JSON.stringify({ path: pendingFiles[filesCompleted].path })))
} else {
resolve({ filesCompleted, bytesTransferred })
}
}
}
// SERVER (A): handle file requests
fileReqMsg.onmessage = async (data) => {
const req = JSON.parse(data.toString())
const filePath = path.join(srcDir, req.path)
const prepared = await transferA.prepareFile(filePath, req.path)
if (!prepared) {
reject(new Error(`prepareFile failed for ${req.path}`))
return
}
// Send chunk info
chunkInfoMsg.send(Buffer.from(JSON.stringify({
path: req.path,
tier: prepared.tier,
size: prepared.size,
chunks: prepared.chunks
})))
// Send all chunks
for (let i = 0; i < prepared.chunks.length; i++) {
const chunkBytes = transferA.readChunk(filePath, prepared.chunks[i].offset, prepared.chunks[i].length)
if (!chunkBytes) {
reject(new Error(`readChunk failed for ${req.path} index ${i}`))
return
}
// Header: 4 bytes index + chunk data
const header = Buffer.alloc(4)
header.writeUInt32LE(i)
chunkDataMsg.send(Buffer.concat([header, chunkBytes]))
}
}
chA.open()
chB.open()
// Server sends file list
const files = await fileIndexA.listFiles()
const fileList = files.map(f => ({ path: f.path, contentHash: f.contentHash, size: f.size }))
fileListMsg.send(Buffer.from(JSON.stringify(fileList)))
})
const result = await transferDone
t.is(result.filesCompleted, 10, 'all 10 files transferred')
t.ok(result.bytesTransferred > 0, `transferred ${result.bytesTransferred} bytes`)
// Verify each file matches
for (let i = 0; i < 10; i++) {
const srcHash = hashFile(path.join(srcDir, `file_${i}.txt`))
const dstHash = hashFile(path.join(dstDir, `file_${i}.txt`))
t.is(srcHash, dstHash, `file_${i}.txt hash matches`)
}
t.teardown(async () => {
sa.destroy()
sb.destroy()
await storeA.close()
await storeB.close()
})
})
test('release distribution: large binary (100MB)', async (t) => {
t.timeout(60000)
const srcDir = mktemp('lg-src')
const dstDir = mktemp('lg-dst')
// Create 100MB file with random-ish data
const bigBuf = Buffer.alloc(100 * 1024 * 1024)
for (let i = 0; i < bigBuf.length; i += 4096) {
bigBuf.writeUInt32LE(i, i) // non-zero pattern so it actually chunks
}
fs.writeFileSync(path.join(srcDir, 'large.bin'), bigBuf)
const storeA = new Corestore(mktemp('lg-store-a'))
const storeB = new Corestore(mktemp('lg-store-b'))
const fileIndexA = new FileIndex(storeA)
const fileIndexB = new FileIndex(storeB)
const transferA = new TransferManager(fileIndexA)
const transferB = new TransferManager(fileIndexB)
await fileIndexA.ready()
await fileIndexB.ready()
const { fileCount } = await indexDirectory(srcDir, fileIndexA, transferA)
t.is(fileCount, 1)
const [sa, sb] = makePair()
storeA.replicate(sa)
storeB.replicate(sb)
const muxA = Protomux.from(sa)
const muxB = Protomux.from(sb)
const startTime = Date.now()
// Simpler single-file transfer
const transferDone = new Promise(async (resolve, reject) => {
const chA = muxA.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('lg') })
const chB = muxB.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('lg') })
const infoMsg = chA.addMessage({ encoding: c.raw })
const dataMsg = chA.addMessage({ encoding: c.raw })
const infoMsgB = chB.addMessage({ encoding: c.raw })
const dataMsgB = chB.addMessage({ encoding: c.raw })
let chunks = null
let totalChunks = 0
let receivedChunks = 0
let bytesRx = 0
infoMsgB.onmessage = (data) => {
const info = JSON.parse(data.toString())
chunks = info.chunks
totalChunks = chunks.length
const targetPath = path.join(dstDir, 'large.bin')
transferB.startReceive(targetPath, { size: info.size, chunks })
}
dataMsgB.onmessage = (data) => {
const index = data.readUInt32LE(0)
const chunkData = data.subarray(4)
const targetPath = path.join(dstDir, 'large.bin')
const result = transferB.writeChunk(targetPath, index, chunkData)
if (!result.ok) { reject(new Error(result.error)); return }
receivedChunks++
bytesRx += chunkData.length
if (transferB.isComplete(targetPath)) {
transferB.finalize(targetPath)
resolve({ totalChunks, bytesRx })
}
}
chA.open()
chB.open()
// Server prepares and sends
const filePath = path.join(srcDir, 'large.bin')
const prepared = await transferA.prepareFile(filePath, 'large.bin')
infoMsg.send(Buffer.from(JSON.stringify({
size: prepared.size,
chunks: prepared.chunks
})))
for (let i = 0; i < prepared.chunks.length; i++) {
const bytes = transferA.readChunk(filePath, prepared.chunks[i].offset, prepared.chunks[i].length)
const header = Buffer.alloc(4)
header.writeUInt32LE(i)
dataMsg.send(Buffer.concat([header, bytes]))
}
})
const result = await transferDone
const elapsed = Date.now() - startTime
t.ok(result.totalChunks > 0, `${result.totalChunks} chunks`)
t.is(result.bytesRx, 100 * 1024 * 1024, '100MB received')
// Verify hash
const srcHash = hashFile(path.join(srcDir, 'large.bin'))
const dstHash = hashFile(path.join(dstDir, 'large.bin'))
t.is(srcHash, dstHash, 'large binary hash matches')
t.comment(`100MB in ${elapsed}ms (${result.totalChunks} chunks, ${(result.bytesRx / elapsed * 1000 / 1024 / 1024).toFixed(0)} MB/s)`)
t.teardown(async () => {
sa.destroy()
sb.destroy()
await storeA.close()
await storeB.close()
})
})
test('release distribution: resumable after disconnect', async (t) => {
t.timeout(60000)
const srcDir = mktemp('res-src')
const dstDir = mktemp('res-dst')
// Create 50MB file — enough chunks to split meaningfully
const bigBuf = Buffer.alloc(50 * 1024 * 1024)
for (let i = 0; i < bigBuf.length; i += 4096) {
bigBuf.writeUInt32LE(i, i)
}
fs.writeFileSync(path.join(srcDir, 'resumable.bin'), bigBuf)
// Shared stores (persist across reconnects)
const storeA = new Corestore(mktemp('res-store-a'))
const storeB = new Corestore(mktemp('res-store-b'))
const fileIndexA = new FileIndex(storeA)
const fileIndexB = new FileIndex(storeB)
const transferA = new TransferManager(fileIndexA)
await fileIndexA.ready()
await fileIndexB.ready()
await indexDirectory(srcDir, fileIndexA, transferA)
// Prepare file once (get chunk info)
const filePath = path.join(srcDir, 'resumable.bin')
const prepared = await transferA.prepareFile(filePath, 'resumable.bin')
const totalChunks = prepared.chunks.length
const cutoff = Math.floor(totalChunks * 0.5)
t.comment(`total chunks: ${totalChunks}, will disconnect after ${cutoff}`)
// --- FIRST ATTEMPT: transfer ~50%, then disconnect ---
const targetPath = path.join(dstDir, 'resumable.bin')
const partialPath = path.join(dstDir, '.resumable.bin' + PARTIAL_SUFFIX)
let firstAttemptChunks = 0
{
const [sa, sb] = makePair()
storeA.replicate(sa)
storeB.replicate(sb)
const transferB1 = new TransferManager(fileIndexB)
await new Promise((resolve) => {
const muxA = Protomux.from(sa)
const muxB = Protomux.from(sb)
const chA = muxA.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('res1') })
const chB = muxB.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('res1') })
const infoMsg = chA.addMessage({ encoding: c.raw })
const dataMsg = chA.addMessage({ encoding: c.raw })
const infoMsgB = chB.addMessage({ encoding: c.raw })
const dataMsgB = chB.addMessage({ encoding: c.raw })
infoMsgB.onmessage = (data) => {
const info = JSON.parse(data.toString())
transferB1.startReceive(targetPath, { size: info.size, chunks: info.chunks })
}
dataMsgB.onmessage = (data) => {
const index = data.readUInt32LE(0)
const chunkData = data.subarray(4)
transferB1.writeChunk(targetPath, index, chunkData)
firstAttemptChunks++
if (firstAttemptChunks >= cutoff) {
sa.destroy()
sb.destroy()
resolve()
}
}
chA.open()
chB.open()
infoMsg.send(Buffer.from(JSON.stringify({ size: prepared.size, chunks: prepared.chunks })))
for (let i = 0; i < prepared.chunks.length; i++) {
const bytes = transferA.readChunk(filePath, prepared.chunks[i].offset, prepared.chunks[i].length)
const header = Buffer.alloc(4)
header.writeUInt32LE(i)
dataMsg.send(Buffer.concat([header, bytes]))
}
})
t.ok(firstAttemptChunks >= cutoff, `first attempt: ${firstAttemptChunks}/${totalChunks} chunks`)
t.ok(fs.existsSync(partialPath), 'partial file exists after disconnect')
}
// --- SECOND ATTEMPT: scan partial, send only missing ---
{
const [sa2, sb2] = makePair()
storeA.replicate(sa2)
storeB.replicate(sb2)
const transferB2 = new TransferManager(fileIndexB)
// Scan partial to find which chunks are already good
const haveSet = new Set()
const partialData = fs.readFileSync(partialPath)
for (let i = 0; i < prepared.chunks.length; i++) {
const ch = prepared.chunks[i]
const slice = partialData.subarray(ch.offset, ch.offset + ch.length)
if (slice.length === ch.length && hashChunk(slice) === ch.hash) {
haveSet.add(i)
}
}
const needed = []
for (let i = 0; i < prepared.chunks.length; i++) {
if (!haveSet.has(i)) needed.push(i)
}
t.comment(`resume: have ${haveSet.size}/${totalChunks}, need ${needed.length}`)
t.ok(needed.length < totalChunks, 'should need fewer chunks than total')
// Resume transfer — DON'T call startReceive (it truncates the partial).
// Instead, manually populate the _active state pointing at existing partial.
const state = {
partialPath,
targetPath,
chunks: prepared.chunks,
received: new Set(haveSet),
total: prepared.chunks.length,
size: prepared.size
}
transferB2._active.set(targetPath, state)
let secondAttemptChunks = 0
await new Promise((resolve, reject) => {
const muxA2 = Protomux.from(sa2)
const muxB2 = Protomux.from(sb2)
const chA = muxA2.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('res2') })
const chB = muxB2.createChannel({ protocol: DEPLOY_PROTOCOL, id: Buffer.from('res2') })
const dataMsg = chA.addMessage({ encoding: c.raw })
const dataMsgB = chB.addMessage({ encoding: c.raw })
dataMsgB.onmessage = (data) => {
const index = data.readUInt32LE(0)
const chunkData = data.subarray(4)
const result = transferB2.writeChunk(targetPath, index, chunkData)
if (!result.ok) { reject(new Error(result.error)); return }
secondAttemptChunks++
if (transferB2.isComplete(targetPath)) {
transferB2.finalize(targetPath)
resolve()
}
}
chA.open()
chB.open()
for (const i of needed) {
const bytes = transferA.readChunk(filePath, prepared.chunks[i].offset, prepared.chunks[i].length)
const header = Buffer.alloc(4)
header.writeUInt32LE(i)
dataMsg.send(Buffer.concat([header, bytes]))
}
})
t.is(secondAttemptChunks, needed.length, `resumed: only ${secondAttemptChunks} chunks sent (not ${totalChunks})`)
const srcHash = hashFile(path.join(srcDir, 'resumable.bin'))
const dstHash = hashFile(path.join(dstDir, 'resumable.bin'))
t.is(srcHash, dstHash, 'resumed file hash matches original')
t.teardown(async () => {
sa2.destroy()
sb2.destroy()
await storeA.close()
await storeB.close()
})
}
})