/**
* NetworkFetcher — Hyperswarm-backed { getTree, getContent, getLatestRelease }
*
* Phase 0.5 follow-on. Lives between pgh-runtime / installTree (which
* expect a fetcher shape) and the wire (Hyperswarm + OverlayProtocol
* tree-request / content-request messages, plus chunk flow for content).
*
* Server side: hyper-deploy's DeployServer (0.5b + enableTrees) attaches
* OverlayProtocol to each incoming connection and has trees +
* contentHashPaths pre-populated. The client just needs to connect on
* the same topic and open a protomux channel.
*
* Usage:
* const fetcher = new NetworkFetcher({ topic: '...hex...', timeout: 15000 })
* await fetcher.ready()
* const tree = await fetcher.getTree(rootHash)
* const content = await fetcher.getContent(contentHash)
* await fetcher.close()
*
* Limitations (scope boundary — further refinement tracked separately):
* - getLatestRelease currently reads from the LOCAL ReleaseRegistry.
* Replication of the publisher's remote registry is a follow-on.
* - Content transfer uses the existing chunk flow triggered by the
* server's _onContentRequest handler — getContent awaits the next
* transferComplete that lands locally.
*/
import fs from '#node:fs'
import path from '#node:path'
import { homedir, tmpdir as osTmpdir } from '#node:os'
import b4a from 'b4a'
import Hyperswarm from 'hyperswarm'
import Protomux from 'protomux'
import Corestore from 'corestore'
import crypto from 'hypercore-crypto'
import { FileIndex } from 'hyper-overlay/file-index'
import { TransferManager } from 'hyper-overlay/transfer'
import { OverlayProtocol } from 'hyper-overlay/protocol'
import { computeTreeHash } from 'hyper-overlay/tree'
import { ReleaseRegistry } from './release-registry.js'
const DEFAULT_TIMEOUT = 30000
const DEFAULT_LOCAL_REGISTRY = path.join(homedir(), '.config', 'pgh', 'release-registry')
export class NetworkFetcher {
/**
* @param {object} opts
* @param {string|Buffer} opts.topic — 32-byte discovery key
* @param {number} [opts.timeout] — per-request timeout (ms)
* @param {string} [opts.storeDir] — temp corestore path
* @param {Hyperswarm} [opts.swarm] — reuse an existing swarm
* @param {string} [opts.localRegistryDir] — for getLatestRelease (local lookup)
* @param {string} [opts.destDir] — where content chunks land; default tmp
*/
constructor (opts = {}) {
if (!opts.topic) throw new Error('NetworkFetcher requires opts.topic')
this.topic = typeof opts.topic === 'string' ? b4a.from(opts.topic, 'hex') : opts.topic
this.timeout = opts.timeout || DEFAULT_TIMEOUT
this.localRegistryDir = opts.localRegistryDir || DEFAULT_LOCAL_REGISTRY
this.swarm = opts.swarm || null
this._ownSwarm = !this.swarm
this._storeDir = opts.storeDir || null
this._destDir = opts.destDir || null
this._store = null
this._fileIndex = null
this._transferManager = null
this._protocol = null
this._peers = [] // all peers discovered on the topic (in connection order)
this._peerResolve = null
this._pendingContent = new Map() // contentHash → { resolve, reject, timer, diskPath }
this._ready = false
// Per-peer timeouts for individual requests. Tree lookups are small
// RPCs — if a peer doesn't respond in a few seconds it doesn't have
// the tree. Content is an actual chunk stream, so it legitimately
// takes longer; we keep that timeout at the full request budget.
// The big speedup comes from `_markPeerGood` promoting responsive
// peers to the front of the list after their first success.
this._perPeerTreeTimeout = Math.min(opts.perPeerTreeTimeout || 5000, this.timeout)
this._perPeerContentTimeout = opts.perPeerContentTimeout || this.timeout
// Silent-peer watchdog: how long an in-flight content request can go
// without ANY chunk arrival before we declare the peer silent and
// fail the request. Companion to fail-fast-on-disconnect (which only
// catches outright connection death). 30s is enough to ride out
// brief mux contention; longer than that and the peer is stuck.
this._idleTimeout = opts.idleTimeout || 30000
}
// Back-compat: _peer was the first peer; expose the first peer for callers
// that reference it (tests, diagnostics). Real tree/content requests walk
// _peers so they survive stale DHT advertisements.
get _peer () { return this._peers[0] || null }
/**
* Promote a peer to the front of our peer list after a successful
* request. This is the key to fast multi-peer handling: once we know
* a peer responds with real data, subsequent calls try it FIRST
* instead of burning per-peer timeouts on dead peers every call.
*/
_markPeerGood (peer) {
const idx = this._peers.indexOf(peer)
if (idx > 0) {
this._peers.splice(idx, 1)
this._peers.unshift(peer)
}
}
async ready () {
if (this._ready) return
// Memoize the in-flight promise. Without this, concurrent callers
// (e.g. server.js's fire-and-forget `fetcher.ready().then(...)` AND
// pgh-runtime's `installTree` → `getTree` → `await this.ready()`)
// each run `new Corestore(this._storeDir)` on the same path. The
// device-file FDLock admits only one; the loser throws "File
// descriptor could not be locked" and leaves `_store` in a closed
// state, so subsequent reads fail with "Corestore is closed".
if (this._readyPromise) return this._readyPromise
this._readyPromise = this._doReady()
try {
await this._readyPromise
} catch (err) {
this._readyPromise = null
throw err
}
}
async _doReady () {
if (!this._storeDir) {
this._storeDir = fs.mkdtempSync(path.join(osTmpdir(), 'pgh-fetcher-'))
}
if (!this._destDir) {
this._destDir = fs.mkdtempSync(path.join(osTmpdir(), 'pgh-fetcher-dest-'))
}
this._store = new Corestore(this._storeDir)
await this._store.ready()
this._fileIndex = new FileIndex(this._store)
await this._fileIndex.ready()
this._transferManager = new TransferManager(this._fileIndex)
const stubSyncEngine = {
_fileIndex: this._fileIndex,
feedKey: '0'.repeat(64),
feed: { length: 0, async get () { return null } },
async markSynced () {},
async logPut () {}
}
this._protocol = new OverlayProtocol(
stubSyncEngine,
this._transferManager,
{
key: this.topic,
autoSync: false,
// Fires when the client-side chunk flow finalizes a file.
// We use it to resolve getContent promises keyed by synthetic path.
onSynced: (info) => this._onSynced(info),
// Fires for EVERY chunk arrival. We reset the per-pending idle
// timer so a peer that connects then goes silent mid-stream is
// detected within `_idleTimeout` ms (not the 5-min global cap).
onChunkProgress: (info) => this._onChunkProgress(info)
}
)
if (!this.swarm) this.swarm = new Hyperswarm()
this.swarm.on('connection', (conn) => this._onConnection(conn))
this.swarm.join(this.topic, { server: false, client: true })
await this.swarm.flush()
// Wait for at least one peer before resolving
if (this._peers.length === 0) {
await Promise.race([
new Promise((resolve) => { this._peerResolve = resolve }),
new Promise((_, reject) => setTimeout(
() => reject(new Error('NetworkFetcher: no peer within ' + this.timeout + 'ms')),
this.timeout
))
])
}
this._ready = true
}
_onConnection (conn) {
const mux = Protomux.from(conn)
const peer = this._protocol.attach(mux)
if (!peer) return
// Track ALL peers on the topic. The DHT may return multiple peers
// (stale advertisements, other seeds, etc.) and only SOME have the
// tree/content we're looking for. requestTree + requestContent
// iterate this list until a peer returns non-empty data.
this._peers.push(peer)
// Connection-death handling. Without this, a peer that disconnects
// mid-chunk-stream leaves _pendingContent entries waiting on the
// 5-min global timeout — the user sees this as a "stall" (chunks
// streaming, then nothing). 2026-05-13 observed:
// tabertron logs `deploy-server error (non-fatal) [ECONNRESET]`
// stortron's getContent just sits there waiting for transferComplete
// Listen for the underlying conn close + fail any pending requests
// that were tied to this peer. Caller retries fast on next peer.
const handleClose = () => {
this._peers = this._peers.filter(p => p !== peer)
for (const pending of this._pendingContent.values()) {
if (pending.peer === peer) {
pending.finish(new Error('peer disconnected mid-stream (' +
pending.contentHash.slice(0, 12) + ')'))
}
}
}
// Hyperswarm streams emit 'close' on disconnect; protomux mux also
// emits 'close' when the underlying stream dies. Attach to both for
// belt-and-suspenders.
conn.on('close', handleClose)
conn.on('error', (err) => {
// Errors here are typically wire-level (ECONNRESET, ETIMEDOUT).
// We don't fail-fast on error alone — wait for the matching
// 'close' event so we don't double-fire. The on-error listener
// exists just to keep Node from crashing.
})
if (this._peerResolve) { this._peerResolve(peer); this._peerResolve = null }
}
async close () {
for (const pending of this._pendingContent.values()) {
clearTimeout(pending.timer)
pending.reject(new Error('NetworkFetcher closing'))
}
this._pendingContent.clear()
if (this._protocol) {
try { this._protocol.destroy() } catch {}
this._protocol = null
}
if (this._fileIndex) { try { await this._fileIndex.close() } catch {} }
if (this._store) { try { await this._store.close() } catch {} }
if (this._ownSwarm && this.swarm) {
try { await this.swarm.destroy() } catch {}
this.swarm = null
}
this._ready = false
}
/**
* Fetch a tree by its hash, walking all known peers until one returns
* a non-empty response.
*
* hyper-overlay's `_onTreeRequest` (server side) replies with
* `entries: []` on lookup miss. Combined with DHT promiscuity — the
* same topic may have multiple advertised peers, some stale/unrelated,
* some with data — a single-peer fetch is unreliable: whichever peer
* happens to connect first might not have the tree.
*
* Strategy: try peers in connection order. First peer to return
* non-empty entries wins. If all peers return empty, treat as
* not-found (throw) — the caller knows the tree SHOULD have content
* (it's in a lockfile) so empty is a failure signal, not a genuinely
* empty directory.
*
* @param {string} hash
* @returns {Promise<{ entries, size }>}
*/
async getTree (hash) {
if (!this._ready) await this.ready()
if (this._peers.length === 0) throw new Error('NetworkFetcher: no peer available')
// Walk peers, accept the first VERIFIED response. hyper-overlay's
// protocol returns `entries: []` when a peer doesn't have a tree —
// that's a valid negative response ("try elsewhere"), not "this
// tree is empty." Content-address verification distinguishes:
// computeTreeHash(entries) must equal the requested hash.
let lastErr = null
const tried = new Set()
for (let round = 0; round < 2; round++) {
for (const peer of this._peers) {
if (tried.has(peer)) continue
tried.add(peer)
try {
const entries = await this._protocol.requestTree(peer, hash, { timeout: this._perPeerTreeTimeout })
const actual = computeTreeHash(entries || [])
if (actual === hash) {
this._markPeerGood(peer)
const size = entries.reduce((a, e) => a + (e.size || 0), 0)
return { entries, size }
}
// Hash mismatch → this peer doesn't have it. Try next.
} catch (err) {
lastErr = err
}
}
// Second round: wait for additional peers to arrive, then retry
// any new ones (plus the old ones if they haven't been tried yet).
if (round === 0) {
await new Promise(r => setTimeout(r, 2000))
tried.clear()
}
}
throw new Error('NetworkFetcher: tree ' + hash.slice(0, 12) + '… not found on any of ' + this._peers.length + ' peer(s)' + (lastErr ? ' (last err: ' + lastErr.message + ')' : ''))
}
/**
* Fetch a file's content by its content hash.
*
* Uses the synthetic-path convention in hyper-overlay/v2 content-request:
* server responds with chunkHashes whose `path` is
* `"content:" + contentHash`. Client pre-registers its target disk
* path under that key in `_filePaths` so the chunk flow writes to
* our destination; then resolves when transferComplete arrives with
* the matching synthetic path.
*
* @param {string} contentHash — hex blake2b hash of the file content
* @returns {Promise<Buffer>}
*/
async getContent (contentHash) {
if (!this._ready) await this.ready()
if (this._peers.length === 0) throw new Error('NetworkFetcher: no peer available')
// Multi-source: fan the request out to ALL peers at once via the
// ChunkScheduler (hyper-overlay `fetchContent`). It assigns distinct
// chunks per peer, fails over on disconnect, and hash-verifies every
// chunk — pulling one blob from many seeders in parallel (the torrent
// path) instead of the legacy one-peer-at-a-time failover in
// `_getContentFromPeer`. Proven by hyper-overlay's
// overlay-v2-fetch.test.js ("chunks fetched from MULTIPLE peers").
// Retry once to pick up peers that arrive late on the topic.
const destPath = path.join(this._destDir, 'blob-' + contentHash + '-' + Date.now() + '-' + Math.random().toString(36).slice(2, 8))
let lastErr = null
for (let round = 0; round < 2; round++) {
const peers = this._peers.slice()
if (peers.length > 0) {
try {
await this._protocol.fetchContent(contentHash, peers, {
destPath,
timeout: this._perPeerContentTimeout
})
// The chunk flow verifies each chunk; verify the full
// concatenation against the advertised content hash here so no
// caller of getContent ever sees un-verified bytes.
const data = fs.readFileSync(destPath)
const actual = crypto.data(data).toString('hex')
if (actual !== contentHash) {
throw Object.assign(
new Error('content-hash mismatch: expected ' + contentHash.slice(0, 12) + '…, got ' + actual.slice(0, 12) + '…'),
{ code: 'EHASHMISMATCH', expected: contentHash, actual }
)
}
try { fs.unlinkSync(destPath) } catch {}
return data
} catch (err) {
lastErr = err
try { fs.unlinkSync(destPath) } catch {}
}
}
// Wait briefly for additional peers to arrive on the topic, then retry.
if (round === 0) await new Promise(r => setTimeout(r, 2000))
}
throw new Error('NetworkFetcher: content ' + contentHash.slice(0, 12) + '… not found on any of ' + this._peers.length + ' peer(s)' + (lastErr ? ' (last err: ' + lastErr.message + ')' : ''))
}
async _getContentFromPeer (peer, contentHash) {
const syntheticPath = 'content:' + contentHash
// Per-peer destPath so concurrent attempts don't clobber each other.
const destPath = path.join(this._destDir, 'blob-' + contentHash + '-' + Date.now() + '-' + Math.random().toString(36).slice(2, 8))
// Pre-register destination so _onChunkHashes + chunk-data handlers
// write to the right place when chunks arrive.
this._protocol._filePaths.set(syntheticPath, destPath)
return new Promise((resolve, reject) => {
let settled = false
const finish = (err, result) => {
if (settled) return
settled = true
clearTimeout(timer)
clearTimeout(pending.idleTimer)
this._pendingContent.delete(contentHash)
if (err) reject(err); else resolve(result)
}
const timer = setTimeout(() => {
finish(new Error('content request timed out on peer: ' + contentHash.slice(0, 12)))
}, this._perPeerContentTimeout)
// Silent-peer watchdog: reset on each chunk arrival via
// _onChunkProgress. If no chunk arrives within _idleTimeout the
// request fails fast and getContent moves on to the next peer.
const idleFire = () => {
finish(new Error('peer went silent mid-stream after ' +
(pending.bytesReceived || 0) + ' byte(s): ' + contentHash.slice(0, 12)))
}
const pending = {
syntheticPath, destPath, finish, timer, peer, contentHash,
bytesReceived: 0,
idleTimer: setTimeout(idleFire, this._idleTimeout),
idleFire
}
// Tag the pending entry with the peer it was sent to. _onConnection's
// close handler uses this to fail-fast any pending requests when the
// connection dies (the dominant LAN-stall pattern observed
// 2026-05-13: tabertron pushed 7 chunks, ECONNRESET, stortron sat
// waiting on the 5-min global timeout).
this._pendingContent.set(contentHash, pending)
this._protocol.requestContent(peer, contentHash, null)
})
}
/**
* Reset the idle timer for any pending content request matching the
* synthetic path of the arriving chunk. Companion to fail-fast on
* disconnect: this fires when the peer is alive but silent.
*/
_onChunkProgress (info) {
if (!info || !info.path || !info.path.startsWith('content:')) return
const contentHash = info.path.slice('content:'.length)
const pending = this._pendingContent.get(contentHash)
if (!pending) return
pending.bytesReceived += (info.bytes || 0)
clearTimeout(pending.idleTimer)
pending.idleTimer = setTimeout(pending.idleFire, this._idleTimeout)
}
/**
* Called by OverlayProtocol when a file completes on the receiver
* side (direction === 'pull'). Resolves any pending getContent whose
* synthetic path matches — AFTER content-address verification.
*
* hyper-overlay's chunk flow doesn't verify the final concatenation
* against the advertised hash (that's the caller's responsibility in
* this protocol). We do it here so no caller of getContent ever sees
* un-verified bytes.
*/
_onSynced (info) {
if (info.direction !== 'pull') return
if (!info.path || !info.path.startsWith('content:')) return
const contentHash = info.path.slice('content:'.length)
const pending = this._pendingContent.get(contentHash)
if (!pending) return
try {
const data = fs.readFileSync(pending.destPath)
const actual = crypto.data(data).toString('hex')
if (actual !== contentHash) {
pending.finish(Object.assign(
new Error('content-hash mismatch: expected ' + contentHash.slice(0, 12) + '…, got ' + actual.slice(0, 12) + '…'),
{ code: 'EHASHMISMATCH', expected: contentHash, actual }
))
return
}
pending.finish(null, data)
} catch (err) {
pending.finish(err)
}
}
/**
* Look up the latest release for an app name.
*
* MVP: reads from the LOCAL ReleaseRegistry. Remote replication of
* the publisher's registry is follow-on work — until then, callers
* that want network-side release lookup need to ship the registry
* core key separately (e.g. via peer config).
*
* @param {string} name
* @returns {Promise<object|null>}
*/
async getLatestRelease (name) {
if (!fs.existsSync(this.localRegistryDir)) return null
const store = new Corestore(this.localRegistryDir)
const reg = new ReleaseRegistry(store)
try {
await reg.ready()
return await reg.latest(name)
} finally {
try { await reg.close() } catch {}
try { await store.close() } catch {}
}
}
}