/**
* Multi-source (torrent-style) content distribution via
* DeployServer.serveOnConnection — topology tests for the pgh multi-source
* deploy feature (increment 1 + increment 3).
*
* Two properties, proven separately because they have conflicting in-memory
* transport requirements (see note on the secret-stream test):
*
* TEST A — MULTI-SOURCE: two DeployServers, each seeding the SAME
* content-addressed tree, both attached to one requester via
* serveOnConnection. The requester's ChunkScheduler pulls chunks from
* BOTH seeders in parallel (peerCount >= 2). This proves serveOnConnection
* produces a valid torrent-style seeder — the whole point of seeding
* content on the shared registry topic (a sidecar that replicates the
* registry now also serves its content). Modeled on
* hyper-overlay/test/overlay-v2-fetch.test.js lines 63-97.
*
* TEST B — SHARED-MUX ORDERING: serveOnConnection attached onto a
* secret-stream connection that corestore.replicate ALREADY claimed
* (the exact order the sidecar's _registrySwarm 'connection' handler
* runs: registryStore.replicate(conn) FIRST, then serveOnConnection).
* The content channel still pairs and a fetch completes. This is the
* keet many-protocols-one-connection pattern, same ordering proven by
* hyper-overlay/test/overlay-v2-corestore-mux.test.js.
*
* Why split: TEST A needs zero-latency synchronous in-memory framing so the
* scheduler observes both seeders before either drains the file — only a
* BARE duplex (Protomux.from(duplex)) gives that. corestore.replicate, on
* the other hand, REQUIRES the keyed secret-stream handshake (a bare duplex
* throws "Invalid Hypercore key"), but secret-stream's async flushing
* serializes the two seeders so multi-peer distribution isn't observable
* in-memory. So TEST A drops the (separately-proven) registry replication to
* observe distribution, and TEST B keeps the registry-mux ordering to prove
* pairing. On a real network both hold at once — latency interleaves the
* seeders AND secret-stream carries replication.
*/
import test from 'brittle'
import Protomux from 'protomux'
import SecretStream from '@hyperswarm/secret-stream'
import { Duplex } from 'streamx'
import crypto from 'hypercore-crypto'
import Corestore from 'corestore'
import fs from '#node:fs'
import path from '#node:path'
import os from '#node:os'
import { DeployServer } from 'hyper-deploy'
import { FileIndex } from 'hyper-overlay/file-index'
import { TransferManager } from 'hyper-overlay/transfer'
// hyper-overlay 0.3.0 renamed lib/protocol-v2.js -> lib/protocol.js and OverlayProtocolV2 ->
// OverlayProtocol (v1 was dropped; the current class IS the v2 wire protocol). The old subpath is not
// in the package's exports map, so this import threw at MODULE LOAD and — because npm test globs
// test/*.test.js — took the whole pgh suite with it. Aliased so the rest of the file is unchanged.
import { OverlayProtocol as OverlayProtocolV2 } from 'hyper-overlay/protocol'
function mkdtemp (label) {
const dir = path.join(os.tmpdir(), label + '-' + crypto.randomBytes(6).toString('hex'))
fs.mkdirSync(dir, { recursive: true })
return dir
}
// In-memory duplex pair (same pattern as the overlay-v2 tests). Synchronous
// framing — both ends' messages queue together, no transport-level latency.
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]
}
// Keyed secret-stream pair over a duplex pair — exactly what Hyperswarm
// hands a 'connection' handler, and what corestore.replicate requires.
async function secretPair () {
const [da, db] = makeDuplex()
const sa = new SecretStream(true, da)
const sb = new SecretStream(false, db)
await Promise.all([
new Promise(r => sa.on('connect', r)),
new Promise(r => sb.on('connect', r))
])
return [sa, sb]
}
const settle = (ms = 300) => new Promise(r => setTimeout(r, ms))
// A DeployServer that serves `dir` but never opens its own swarm — we drive
// connections by hand to control the exact mux ordering. Its overlay channel
// id is re-keyed to `sharedTopic` so a requester keyed to the same topic
// pairs with it (stands in for the shared registry discoveryKey).
async function makeSeeder (dir, sharedTopic, { serveLatency = 0 } = {}) {
const store = new Corestore(mkdtemp('ms-srvstore'))
const swarmStub = { on () {}, join () { return { flushed: async () => {}, destroy: async () => {} } }, destroy: async () => {} }
const srv = new DeployServer(dir, { corestore: store, swarm: swarmStub })
await srv.ready()
srv._overlayProtocol._overlayKey = sharedTopic
// Optional artificial per-chunk-request latency. On a zero-latency
// in-memory link the seeder whose chunk-list lands first can drain the
// whole (content-addressed) file before the second seeder is added to the
// scheduler's peer set — so multi-source distribution isn't observable. A
// few ms of latency on the chunk-serve path interleaves the two seeders
// exactly as a real network would, making the multi-source assignment
// deterministic. Latency is applied to the SERVE path only (this is a
// property of the transport, not of serveOnConnection).
if (serveLatency > 0) {
const proto = srv._overlayProtocol
const orig = proto._onChunkNeed.bind(proto)
proto._onChunkNeed = async (peer, msg) => {
await new Promise(r => setTimeout(r, serveLatency))
return orig(peer, msg)
}
}
return srv
}
// A requester protocol mirroring the shape NetworkFetcher builds in _doReady:
// stub sync engine + FileIndex + TransferManager + OverlayProtocolV2.
async function makeRequester (sharedTopic) {
const store = new Corestore(mkdtemp('ms-req'))
await store.ready()
const index = new FileIndex(store)
await index.ready()
const tm = new TransferManager(index)
const stubSync = {
_fileIndex: index,
feedKey: '0'.repeat(64),
feed: { length: 0, async get () { return null } },
async markSynced () {},
async logPut () {}
}
const proto = new OverlayProtocolV2(stubSync, tm, { key: sharedTopic, autoSync: false })
return { store, index, proto }
}
test('A — multi-source: two serveOnConnection seeders; fetcher pulls chunks from BOTH', async (t) => {
const content = crypto.randomBytes(900 * 1024) // ~56 chunks at tier 0
const oid = crypto.data(content).toString('hex')
const dir1 = mkdtemp('ms-seedA1')
const dir2 = mkdtemp('ms-seedA2')
fs.writeFileSync(path.join(dir1, 'blob.bin'), content)
fs.writeFileSync(path.join(dir2, 'blob.bin'), content)
const sharedTopic = crypto.randomBytes(32)
// 5ms serve latency on each seeder so the two interleave deterministically
// (see makeSeeder note). Without it, in-memory zero-latency framing lets
// one seeder drain the whole file before the other engages.
const s1 = await makeSeeder(dir1, sharedTopic, { serveLatency: 5 })
const s2 = await makeSeeder(dir2, sharedTopic, { serveLatency: 5 })
const req = await makeRequester(sharedTopic)
const conns = []
t.teardown(async () => {
for (const c of conns) { try { c.destroy() } catch {} }
try { req.proto.destroy() } catch {}
try { await s1.close() } catch {}
try { await s2.close() } catch {}
try { await req.index.close() } catch {}
try { await req.store.close() } catch {}
})
// Bare duplex per seeder (no corestore replication — see file header).
// serveOnConnection attaches the overlay content channel onto Protomux.from
// (conn); the requester attaches its overlay onto the other end.
for (const srv of [s1, s2]) {
const [srvConn, reqConn] = makeDuplex()
conns.push(srvConn, reqConn)
const served = srv.serveOnConnection(srvConn)
t.ok(served, 'serveOnConnection attached the content channel')
req.proto.attach(Protomux.from(reqConn))
}
await settle()
const peers = [...req.proto._peers.values()]
t.is(peers.length, 2, 'requester is connected to BOTH seeders')
const destPath = path.join(mkdtemp('ms-destA'), 'out.bin')
const stats = await req.proto.fetchContent(oid, peers, { destPath, timeout: 15000 })
const got = fs.readFileSync(destPath)
t.is(Buffer.compare(got, content), 0, 'assembled file matches the seeded content')
t.ok(stats.peerCount >= 2,
'chunks were pulled from BOTH seeders (multi-source) — peerCount=' + stats.peerCount)
})
test('B — shared-mux ordering: corestore.replicate FIRST, then serveOnConnection; fetch completes', async (t) => {
const content = crypto.randomBytes(300 * 1024)
const oid = crypto.data(content).toString('hex')
const dir = mkdtemp('ms-seedB')
fs.writeFileSync(path.join(dir, 'blob.bin'), content)
const sharedTopic = crypto.randomBytes(32)
const srv = await makeSeeder(dir, sharedTopic)
const req = await makeRequester(sharedTopic)
// The registry store the sidecar replicates on its registry connections.
const regStore = new Corestore(mkdtemp('ms-regB'))
await regStore.ready()
const reqRegStore = new Corestore(mkdtemp('ms-reqregB'))
await reqRegStore.ready()
const conns = []
t.teardown(async () => {
for (const c of conns) { try { c.destroy() } catch {} }
try { req.proto.destroy() } catch {}
try { await srv.close() } catch {}
try { await req.index.close() } catch {}
try { await req.store.close() } catch {}
try { await regStore.close() } catch {}
try { await reqRegStore.close() } catch {}
})
const [srvConn, reqConn] = await secretPair()
conns.push(srvConn, reqConn)
// EXACTLY the sidecar's _registrySwarm 'connection' handler order:
// registryStore.replicate(conn) — claims the Protomux first
// deployServer.serveOnConnection(conn) — content channel onto same mux
regStore.replicate(srvConn)
const served = srv.serveOnConnection(srvConn)
t.ok(served, 'serveOnConnection attached under corestore-replicate-first ordering')
reqRegStore.replicate(reqConn)
req.proto.attach(Protomux.from(reqConn))
await settle(500)
t.is(req.proto.peerCount, 1, 'content channel paired despite registry replication sharing the mux')
const peers = [...req.proto._peers.values()]
const destPath = path.join(mkdtemp('ms-destB'), 'out.bin')
await req.proto.fetchContent(oid, peers, { destPath, timeout: 15000 })
const got = fs.readFileSync(destPath)
t.is(Buffer.compare(got, content), 0,
'fetched the file over the shared (replicate-first) mux — keet many-protocols-one-connection works')
})