/**
* hyper-deploy — P2P release distribution built on hyper-overlay
*
* Serves build artifacts from a directory to peers over Hyperswarm.
* File-level dedup via content hashing. Files >= 1MB get FastCDC chunking
* for resumable transfers. Smaller files transfer as whole blobs.
*
* Usage:
* import { DeployServer, DeployClient } from 'hyper-deploy'
*
* const server = new DeployServer(dir)
* await server.ready()
*
* const client = new DeployClient(topic, destDir)
* await client.ready()
* const result = await client.pull()
*/
import Hyperswarm from 'hyperswarm'
import Corestore from 'corestore'
import crypto from 'hypercore-crypto'
import fs from '#fs'
import path from '#path'
import os from '#os'
import ReadyResource from 'ready-resource'
import safetyCatch from 'safety-catch'
import Protomux from 'protomux'
import { FileIndex } from 'hyper-overlay/file-index'
import { TransferManager } from 'hyper-overlay/transfer'
import { hashChunk } from 'hyper-overlay/chunker'
import { walkTree } from 'hyper-overlay/tree'
import { OverlayProtocol } from 'hyper-overlay/protocol'
import { attachProtocol } from './lib/protocol.js'
const CHUNK_THRESHOLD = 1048576 // 1 MB — files below this transfer as whole blobs
// ── Helpers ──────────────────────────────────────────────────
// Walks `dir` and returns an array of entries describing every file and
// symlink found. Symlink ENTRIES are returned with the link-target string
// as their served content (not the file they resolve to) — this matches
// hyper-overlay/lib/tree.js:walkTree, which records KIND_SYMLINK entries
// the same way. Without this, the link-target string's content hash is
// never indexed for serving, so any client materializing a tree with
// symlinks hangs on the symlink's content-request (root cause of the
// "install stuck at SunshineClouds2" hang). Uses lstatSync so symlinks
// are NOT followed.
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
try {
const stat = fs.lstatSync(full)
if (stat.isSymbolicLink()) {
const linkTarget = fs.readlinkSync(full)
const targetBytes = Buffer.from(linkTarget, 'utf8')
results.push({ path: rel, size: targetBytes.length, isSymlink: true, linkTarget })
} else if (stat.isDirectory()) {
results.push(...walkDir(full, rel))
} else if (stat.isFile()) {
results.push({ path: rel, size: stat.size })
}
} catch {}
}
return results
}
function hashFile (filePath) {
return hashChunk(fs.readFileSync(filePath))
}
// Bumped when the persisted index gains a field the fast manifest-diff boot
// path relies on. On a version mismatch (or absence) the boot does ONE full
// _scan() to repopulate + re-stamp, then fast-paths thereafter. v1 added
// `linkname` on file entries (so symlinks are reconstructable without lstat).
const SCAN_SCHEMA_VERSION = 1
// ── DeployServer ─────────────────────────────────────────────
export class DeployServer extends ReadyResource {
constructor (dir, opts) {
super()
opts = opts || {}
this.dir = path.resolve(dir)
this.topic = opts.topic || crypto.randomBytes(32)
this.swarm = opts.swarm || null
this._ownSwarm = !this.swarm
this._corestore = opts.corestore || null
this._ownStore = !this._corestore
this._fileIndex = null
this._transferManager = null
this._discovery = null
this._sessions = new Set()
// Connections we've already armed an overlay-release handler on, so a
// repeated serveOnConnection (it is documented idempotent-ish, and
// _onConnection calls it too) adds exactly one 'close' listener. Weak so
// holding it never keeps a dead connection alive. See serveOnConnection.
this._armedConns = new WeakSet()
// 0.5b tree publish + 0.5c tree serving: when trees are populated
// (which we do in _scan), also attach hyper-overlay/v2 to each
// connection so clients can request trees + content-by-hash.
this._enableTrees = opts.enableTrees !== false
this._overlayProtocol = null
this.manifest = null
// Overlay lookup maps + the manifest identity they were built from.
// See _overlayLookupMaps.
this._lookupMaps = null
this._lookupMapsFor = null
this.stats = { totalConnections: 0, filesServed: 0, bytesServed: 0 }
// Symlink-target blob dir. Durability follows the corestore: a
// caller-supplied (persistent) store — e.g. the sidecar — keeps blobs in a
// durable sibling dir so the fast manifest-diff boot path can serve symlink
// content across restarts WITHOUT re-walking to re-materialize them. An OWN
// (ephemeral tmp) store gets no persistent fast-path benefit, so it uses a
// tmp dir that _close() cleans up (no litter). content-addressed either way.
this._symlinkBlobsDir = opts.symlinkBlobsDir || (this._ownStore
? path.join(os.tmpdir(), 'hyper-deploy-symlinks-' + crypto.randomBytes(8).toString('hex'))
: path.join(path.dirname(this.dir), '.hyper-deploy-symlink-blobs'))
this.ready().catch(safetyCatch)
}
async _open () {
if (!fs.existsSync(this.dir)) {
throw new Error('Directory not found: ' + this.dir)
}
// Setup corestore + overlay primitives
if (!this._corestore) {
const tmp = path.join(os.tmpdir(), 'hyper-deploy-srv-' + Date.now())
this._corestore = new Corestore(tmp)
}
this._fileIndex = new FileIndex(this._corestore)
await this._fileIndex.ready()
this._transferManager = new TransferManager(this._fileIndex)
// Overlay-v2 protocol for tree-request / content-request. Shares
// the fileIndex with the file-list flow; no separate storage.
if (this._enableTrees) {
const stubSyncEngine = {
_fileIndex: this._fileIndex,
feedKey: '0'.repeat(64),
feed: { length: 0, async get () { return null } },
async markSynced () {},
async logPut () {}
}
this._overlayProtocol = new OverlayProtocol(
stubSyncEngine,
this._transferManager,
{ key: this.topic, autoSync: false }
)
}
// Index directory. Prefer the fast manifest-diff boot path (reconstruct the
// manifest from the persistent fileIndex + walk only NEW hash-dirs); fall
// back to a full rescan whenever the persisted schema is absent/older or
// anything is inconsistent — correctness over speed.
fs.mkdirSync(this._symlinkBlobsDir, { recursive: true })
await this._scanBoot()
// Start swarm
if (!this.swarm) this.swarm = new Hyperswarm()
this.swarm.on('connection', (conn) => this._onConnection(conn))
this._discovery = this.swarm.join(this.topic, { server: true, client: false })
await this._discovery.flushed()
this.emit('ready')
}
async _close () {
for (const session of this._sessions) {
session.destroyed = true
}
this._sessions.clear()
if (this._discovery) {
await this._discovery.destroy()
this._discovery = null
}
if (this._ownSwarm && this.swarm) {
await this.swarm.destroy()
this.swarm = null
}
if (this._fileIndex) await this._fileIndex.close()
if (this._ownStore && this._corestore) await this._corestore.close()
// Ephemeral (own tmp corestore → no persistent fast-path benefit): clean up
// the tmp symlink-blob dir on close (no litter). Durable (caller-supplied
// corestore, e.g. the sidecar): KEEP the blobs so the fast manifest-diff
// boot path can serve symlink content across restarts — they are GC'd
// alongside the tree store, not on process exit.
if (this._ownStore && this._symlinkBlobsDir) {
try { fs.rmSync(this._symlinkBlobsDir, { recursive: true, force: true }) } catch {}
}
}
get topicHex () {
return this.topic.toString('hex')
}
get connections () {
return this._sessions.size
}
/**
* Refresh the file/tree index from disk.
*
* Default behavior (legacy): full rescan — walks the entire `this.dir`,
* re-hashes every file, rebuilds the manifest from scratch. O(total
* bytes in the served directory). Caller-cost: many minutes for a
* pgh-sidecar tree-store with tens of thousands of files.
*
* Incremental mode: pass `{ newHashes: ['<hash1>', ...] }` to scan
* ONLY the named subdirectories (e.g. `<dir>/<hash1>/`). The result
* is appended to the existing manifest. O(bytes in those subdirs).
* Use this when you know exactly which trees were just staged —
* publish-release does, since the caller is naming the tree.
*
* Removed-on-disk files are NOT reaped in incremental mode (they
* stay in the manifest until the next full rescan). For pgh's
* append-only publish flow this is correct; if you need eviction
* call rescan() with no args to do a full rebuild.
*/
async rescan (opts = {}) {
if (opts && Array.isArray(opts.newHashes) && opts.newHashes.length > 0) {
const added = await this._scanIncremental(opts.newHashes)
return {
files: this.manifest.files.length,
totalSize: this.manifest.totalSize,
incremental: true,
addedFiles: added.addedFiles,
addedBytes: added.addedBytes,
scannedHashes: opts.newHashes.length
}
}
await this._scan()
return { files: this.manifest.files.length, totalSize: this.manifest.totalSize }
}
/**
* Walk only the listed top-level subdirs (typically tree-hash dirs
* staged by `pgh publish`) and add their files + trees to the
* existing fileIndex + manifest. No re-hashing of paths already in
* the manifest.
*
* @param {string[]} subPaths - directory names relative to this.dir
* @returns {{ addedFiles: number, addedBytes: number }}
*/
async _scanIncremental (subPaths) {
// Durable symlink-blob dir (set in the constructor, ensured in _open).
fs.mkdirSync(this._symlinkBlobsDir, { recursive: true })
if (!this.manifest) {
// First-ever scan or after a full reset — fall back to legacy full scan,
// since incremental can't synthesize the manifest skeleton from nothing.
await this._scan()
return { addedFiles: 0, addedBytes: 0 }
}
let addedFiles = 0
let addedBytes = 0
// Track paths already in manifest so we don't double-add on
// overlapping incremental calls.
const existing = new Set(this.manifest.files.map(f => f.path))
for (const sub of subPaths) {
const subDir = path.join(this.dir, sub)
let stat
try { stat = fs.lstatSync(subDir) } catch { continue }
if (!stat.isDirectory()) continue
const files = walkDir(subDir, sub)
for (let i = 0; i < files.length; i++) {
const f = files[i]
if (existing.has(f.path)) continue // already indexed
let diskPath, contentHash, mtime
if (f.isSymlink) {
const linkBytes = Buffer.from(f.linkTarget, 'utf8')
contentHash = hashChunk(linkBytes)
diskPath = path.join(this._symlinkBlobsDir, contentHash)
if (!fs.existsSync(diskPath)) fs.writeFileSync(diskPath, linkBytes)
mtime = Date.now()
} else {
diskPath = path.join(this.dir, f.path)
contentHash = hashFile(diskPath)
mtime = fs.statSync(diskPath).mtimeMs
}
f.diskPath = diskPath
await this._fileIndex.putFile(f.path, {
contentHash,
size: f.size,
mtime,
linkname: f.isSymlink ? f.linkTarget : null
})
if (f.size >= CHUNK_THRESHOLD) {
await this._transferManager.prepareFile(diskPath, f.path)
}
this.manifest.files.push({
path: f.path,
size: f.size,
diskPath: f.diskPath,
hash: contentHash
})
existing.add(f.path)
addedFiles++
addedBytes += f.size
this.manifest.totalSize += f.size
// Also push the new entry into the overlay protocol's lookup
// maps so peers connected BEFORE this rescan can find the
// new bytes via content-request without reconnecting. Without
// this, the maps are stale until the next inbound connection
// overwrites them via `_onConnection`.
if (this._overlayProtocol) {
if (this._overlayProtocol._contentHashPaths) {
this._overlayProtocol._contentHashPaths.set(contentHash, diskPath)
}
if (this._overlayProtocol._filePaths) {
this._overlayProtocol._filePaths.set(f.path, diskPath)
}
}
}
// Tree-level publish for the newly-added subdir. walkTree(subDir)
// produces tree hashes for the new subtree only; we putTree each
// one and putTreePath at sub-relative paths plus '/'.
try {
const trees = walkTree(subDir)
for (const [relPath, tree] of trees) {
await this._fileIndex.putTree(tree.hash, {
entries: tree.entries,
size: tree.size
})
// The relPath here is sub-tree-relative. Tree-by-path lookups
// use this.dir-relative paths — prefix `sub`. The empty
// relPath is the subtree root, which maps to `sub` in our
// top-level naming.
const fullRel = relPath ? sub + '/' + relPath : sub
await this._fileIndex.putTreePath(fullRel, tree.hash)
}
// The new subtree's root is itself a tree advertised at hash =
// tree-by-name. The PUBLISHER calls us with the tree hash so we
// also want path = '<sub>' to map there, which the loop above
// covers for relPath = ''.
this.emit('incremental-indexed', sub, trees.size)
} catch (err) {
this.emit('tree-publish-error', err)
}
}
return { addedFiles, addedBytes }
}
/**
* Boot indexer. Fast manifest-diff path when the persisted index carries the
* current scan schema; otherwise a full _scan() that repopulates + re-stamps
* the schema. ANY failure in the fast path falls back to a full _scan().
*/
async _scanBoot () {
let schema = null
try { schema = await this._fileIndex.getIndexMeta('scanSchema') } catch {}
if (schema && schema.version === SCAN_SCHEMA_VERSION) {
try {
await this._scanFromManifest()
return
} catch (err) {
this.emit('manifest-scan-fallback', err)
// fall through to a full, always-correct rescan
}
}
await this._scan()
try { await this._fileIndex.putIndexMeta('scanSchema', { version: SCAN_SCHEMA_VERSION }) } catch {}
}
/**
* Fast boot: reconstruct the manifest from the persistent fileIndex (ONE
* sequential hyperbee scan) and reconcile only the hash-dir delta on disk.
* The served `<dir>/<hash>/` dirs are content-addressed + immutable, so an
* unchanged dir is never re-walked/re-stat'd. Throws on any inconsistency so
* _scanBoot can fall back to a full scan (correctness over speed).
*/
async _scanFromManifest () {
const persisted = await this._fileIndex.listFiles('')
if (persisted.length === 0) throw new Error('empty index — full scan required')
// Current top-level hash-dirs on disk (one readdir, NOT a recursive walk).
const currentDirs = new Set()
for (const name of fs.readdirSync(this.dir)) {
let st
try { st = fs.statSync(path.join(this.dir, name)) } catch { continue }
if (st.isDirectory()) currentDirs.add(name)
}
// Group persisted entries by first path segment (= hash-dir name).
const filesByDir = new Map()
for (const e of persisted) {
const slash = e.path.indexOf('/')
const seg = slash === -1 ? e.path : e.path.slice(0, slash)
if (!filesByDir.has(seg)) filesByDir.set(seg, [])
filesByDir.get(seg).push(e)
}
// Prune GC'd dirs: indexed dirs no longer on disk (bidirectional diff).
const goneDirs = []
for (const dir of filesByDir.keys()) if (!currentDirs.has(dir)) goneDirs.push(dir)
let pruned = 0
for (const dir of goneDirs) {
for (const e of filesByDir.get(dir)) { await this._fileIndex.delFile(e.path); pruned++ }
filesByDir.delete(dir)
}
// Reconstruct the manifest from surviving (unchanged) dirs — no walk/stat.
// Regular files resolve their diskPath by pure path join (zero fs calls);
// only symlinks touch the fs (existsSync + self-heal of the durable blob).
this.manifest = { files: [], totalSize: 0, rootTreeHash: null }
for (const entries of filesByDir.values()) {
for (const e of entries) {
this.manifest.files.push({
path: e.path,
size: e.size,
diskPath: this._diskPathFor(e),
hash: e.contentHash || null
})
this.manifest.totalSize += e.size
}
}
const reused = this.manifest.files.length
// New dirs (on disk, absent from the index) → incremental index (files +
// trees + symlink blobs), appending to this.manifest.
const newDirs = []
for (const dir of currentDirs) if (!filesByDir.has(dir)) newDirs.push(dir)
if (newDirs.length) await this._scanIncremental(newDirs)
this.emit('scan-cache-stats', {
hits: reused,
misses: this.manifest.files.length - reused,
hashedBytes: 0,
totalFiles: this.manifest.files.length,
prunedFiles: pruned,
newDirs: newDirs.length,
fastPath: true
})
// manifest.rootTreeHash is left null on the fast path — vestigial for a
// multi-tree store (clients fetch specific tree hashes, which stay
// individually persisted). The full-scan fallback still computes it.
this.emit('indexed', this.manifest.files.length, this.manifest.totalSize)
}
/**
* diskPath for a persisted file entry. Symlink entries (non-null linkname)
* resolve to the durable content-addressed blob, self-healed from the
* persisted linkname if missing (no tree re-read). Regular files resolve to
* their in-tree path (stable — dirs are immutable), with no fs call.
*/
_diskPathFor (entry) {
if (entry.linkname != null && entry.contentHash) {
const blob = path.join(this._symlinkBlobsDir, entry.contentHash)
if (!fs.existsSync(blob)) {
try { fs.writeFileSync(blob, Buffer.from(entry.linkname, 'utf8')) } catch {}
}
return blob
}
return path.join(this.dir, entry.path)
}
async _scan () {
const files = walkDir(this.dir)
let totalSize = 0
// Durable dir for symlink-target blobs (set in the constructor, ensured in
// _open). Each unique link-target string is materialized to
// <symlinkBlobsDir>/<contentHash> so the diskPath-based content-request flow
// can serve it via fs.readFileSync. Single source of bytes; the protocol
// layer needn't know about symlinks.
fs.mkdirSync(this._symlinkBlobsDir, { recursive: true })
// Mtime cache stats — measured at debug/observability level. When
// the fileIndex already has a matching (size, mtime) for a path,
// skip the readFileSync+hashChunk. For pgh-sidecar's content-
// addressed store the tree-hash dirs are immutable; this turns
// the legacy full-scan from O(disk bytes) into O(stat calls),
// typically ~100-1000× faster on warm restart.
let _cacheHits = 0
let _cacheMisses = 0
let _hashedBytes = 0
for (let i = 0; i < files.length; i++) {
const f = files[i]
let diskPath, contentHash, mtime, wasCached = false
if (f.isSymlink) {
const linkBytes = Buffer.from(f.linkTarget, 'utf8')
contentHash = hashChunk(linkBytes)
diskPath = path.join(this._symlinkBlobsDir, contentHash)
if (!fs.existsSync(diskPath)) fs.writeFileSync(diskPath, linkBytes)
mtime = Date.now()
} else {
diskPath = path.join(this.dir, f.path)
const statMtime = fs.statSync(diskPath).mtimeMs
// mtime-based skip: if fileIndex has this path with matching size
// + mtime, the bytes haven't changed → reuse stored contentHash.
// Turns the legacy full-scan from O(disk bytes) to O(stat calls)
// when the index is warm. For pgh-sidecar's content-addressed
// store (tree-hash dirs are immutable) this is the common path
// on every restart.
const cached = await this._fileIndex.getFile(f.path).catch(() => null)
if (cached && cached.size === f.size && cached.mtime === statMtime && cached.contentHash) {
contentHash = cached.contentHash
mtime = statMtime
wasCached = true
_cacheHits++
} else {
contentHash = hashFile(diskPath)
mtime = statMtime
_cacheMisses++
_hashedBytes += f.size
}
}
f.diskPath = diskPath
// Only write the fileIndex entry on cache miss / for symlinks —
// skips the autobase write tail entirely on a warm rescan.
if (!wasCached) {
await this._fileIndex.putFile(f.path, {
contentHash,
size: f.size,
mtime,
// Persist the symlink target so the fast boot path can rebuild the
// diskPath (durable blob) without an lstat. null for regular files.
linkname: f.isSymlink ? f.linkTarget : null
})
}
// For large files, prepare chunk map via TransferManager. The
// chunk map is in-memory only, so cache hits still need this
// call (no disk re-read since the bytes haven't changed).
if (f.size >= CHUNK_THRESHOLD) {
await this._transferManager.prepareFile(diskPath, f.path)
}
totalSize += f.size
if ((i + 1) % 1000 === 0) this.emit('index-progress', i + 1, files.length)
}
this.emit('scan-cache-stats', {
hits: _cacheHits,
misses: _cacheMisses,
hashedBytes: _hashedBytes,
totalFiles: files.length
})
this.manifest = {
files: files.map(f => ({
path: f.path,
size: f.size,
diskPath: f.diskPath, // points to symlink blob OR source file, used by _onConnection
hash: null // populated lazily from fileIndex
})),
totalSize,
rootTreeHash: null // filled in by tree publish pass (0.5b)
}
// Populate hashes from the index
for (const f of this.manifest.files) {
const meta = await this._fileIndex.getFile(f.path)
if (meta) f.hash = meta.contentHash
}
// 0.5b: tree publish pass — compute + persist tree hashes for every
// directory. Additive to the file-level index; existing client flows
// (file-by-path sync) remain unchanged. Clients that know about trees
// (pgh) can start from manifest.rootTreeHash and fetch subtrees.
try {
const trees = walkTree(this.dir)
for (const [relPath, tree] of trees) {
await this._fileIndex.putTree(tree.hash, {
entries: tree.entries,
size: tree.size
})
await this._fileIndex.putTreePath(relPath || '/', tree.hash)
}
const root = trees.get('')
if (root) this.manifest.rootTreeHash = root.hash
this.emit('trees-indexed', trees.size, this.manifest.rootTreeHash)
} catch (err) {
// Non-fatal: trees are an additive layer. If walkTree fails (e.g.
// a name that can't round-trip through normalizeName), fall back to
// file-level-only publishing. Caller sees rootTreeHash === null.
this.emit('tree-publish-error', err)
}
this.emit('indexed', this.manifest.files.length, totalSize)
}
/**
* Attach this server's hyper-overlay/v2 content channel to an EXISTING
* connection, so the peer can request any served file by its content
* hash. Idempotent-ish: rebuilds the content-hash → disk-path map from
* the CURRENT manifest on every call (so freshly-rescanned trees are
* served), then attaches the overlay protocol onto the connection's
* Protomux.
*
* Reuses the DeployServer's own `_overlayProtocol` (sharing its
* FileIndex / TransferManager) — NOT a separate FileIndex — so trees
* added by `rescan({ newHashes })` are immediately reachable here.
*
* Use this to mux content-serving onto a connection that already
* carries other protocols (e.g. a registry-replication connection): the
* caller does its own `corestore.replicate(conn)` first, then calls
* this. The keet "many protocols, one connection" pattern.
*
* No-op when trees are disabled (`enableTrees: false`).
*
* @param {Duplex} conn — an open connection (secret-stream / duplex)
* @returns {boolean} true if the content channel was attached
*/
/**
* content-hash → disk-path and path → disk-path lookups for the overlay,
* built once per manifest rather than once per connection.
*
* These were rebuilt from scratch inside serveOnConnection on EVERY inbound
* connection. On tabertron's tree store (39,484 files / 2,977 trees) that
* measured 3.8 MB of garbage and ~79,000 map insertions per connection, for
* a result that is identical every time until the manifest changes.
*
* The cache key is manifest IDENTITY, which is exactly right for the two
* ways manifest.files changes:
*
* _scan / _scanBoot assign a NEW manifest object -> identity differs,
* so the next caller rebuilds.
* _scanIncremental push()es into the EXISTING array -> identity is
* unchanged, and it maintains these same Map objects
* itself with its own .set() calls, so reuse is not
* merely an optimization here: rebuilding would be
* redundant work that throws away nothing.
*
* A version counter would be wrong for the same reason — the incremental
* path deliberately keeps the live maps current so peers connected BEFORE a
* rescan can fetch new bytes without reconnecting.
*/
_overlayLookupMaps () {
if (this._lookupMaps && this._lookupMapsFor === this.manifest) {
return this._lookupMaps
}
const contentHashPaths = new Map()
const filePaths = new Map()
for (const f of (this.manifest && this.manifest.files) || []) {
if (f.hash) {
// f.diskPath points to the symlink-target blob for symlink
// entries and to the actual source file for regular files.
// Falls back to dir-relative path for back-compat with any
// older manifest shapes that lack f.diskPath.
const diskPath = f.diskPath || path.join(this.dir, f.path)
contentHashPaths.set(f.hash, diskPath)
filePaths.set(f.path, diskPath)
}
}
this._lookupMaps = { contentHashPaths, filePaths }
this._lookupMapsFor = this.manifest
return this._lookupMaps
}
serveOnConnection (conn) {
if (!this._overlayProtocol) return false
try {
const mux = Protomux.from(conn)
const { contentHashPaths, filePaths } = this._overlayLookupMaps()
this._overlayProtocol._contentHashPaths = contentHashPaths
this._overlayProtocol._filePaths = filePaths
this._overlayProtocol.attach(mux)
// attach() pairs the mux; only detach() unpairs it. Arm the inverse
// HERE rather than leaving it to the caller: this is the public way to
// mux content-serving onto a connection the caller owns, and it had no
// public inverse, so an external caller could not release the pairing
// even knowing it should. pgh-sidecar serves its registry-swarm
// connections this way and pinned every one of them for the life of the
// process. 0.4.11 fixed only connections arriving via our own swarm.
this._armOverlayRelease(conn)
return true
} catch (err) {
this.emit('error', new Error('failed to attach overlay-v2 protocol: ' + err.message))
return false
}
}
/**
* Release the overlay protocol's per-mux state for a dead connection.
*
* OverlayProtocol keeps `_peers` (pruned by the channel's own onclose) and
* `_paired` (pruned ONLY by detach). attach() deliberately leaves the
* mux.pair() registration alive across a channel close so a restarting peer
* can re-pair for the connection's whole life — correct while the connection
* lives, a leak once it dies. One DeployServer attaches ONE shared
* _overlayProtocol to every inbound mux, so without this every connection
* ever made stays pinned, along with its mux and that mux's stream.
*
* Left unpruned this reached anon-rss ~22 GB in 3817s on tabertron
* (2026-09-02) under deploy-server connection churn, and was OOM-killed.
*
* Protomux.from is idempotent, so this resolves the same mux
* serveOnConnection attached. Best-effort: detach() closes the channel, and
* on an already-closed connection that is a no-op we do not want to throw
* out of a 'close' handler.
*/
/**
* Drop the overlay pairing when `conn` closes. One listener per connection
* (WeakSet-guarded), and _detachOverlay is idempotent — _paired.delete is
* guarded and _peers.get returns early — so overlapping with
* _onConnection's own close/error handlers is harmless.
*/
_armOverlayRelease (conn) {
if (!conn || typeof conn.once !== 'function') return
if (this._armedConns.has(conn)) return
this._armedConns.add(conn)
conn.once('close', () => this._detachOverlay(conn))
}
_detachOverlay (conn) {
if (!this._overlayProtocol) return
try {
this._overlayProtocol.detach(Protomux.from(conn))
} catch {}
}
_onConnection (conn) {
const peerKey = conn.remotePublicKey
? conn.remotePublicKey.toString('hex').slice(0, 12)
: 'unknown'
this.stats.totalConnections++
const proto = attachProtocol(conn)
if (!proto) {
this.emit('error', new Error('Failed to attach protocol for peer ' + peerKey))
return
}
// Also attach hyper-overlay/v2 on the same stream (protomux handles
// multiplexing). Populates contentHashPaths from the fileIndex so
// content-request can serve any file by its contentHash.
this.serveOnConnection(conn)
const session = { peerKey, destroyed: false, filesServed: 0, bytesServed: 0 }
this._sessions.add(session)
this.emit('peer', { key: peerKey, state: 'connected' })
conn.on('close', () => {
this._sessions.delete(session)
this._detachOverlay(conn)
this.emit('peer', { key: peerKey, state: 'disconnected' })
})
conn.on('error', (err) => {
this._sessions.delete(session)
this._detachOverlay(conn)
this.emit('error', err)
})
// v1 fileList broadcast removed (2026-05-04): with a shared
// dependencies cache holding 100k+ files (1207 trees on this
// machine), `JSON.stringify(this.manifest.files).length` exceeds
// secret-stream's 16 MB MAX_ATOMIC_WRITE limit. The unsolicited
// broadcast on every peer connection killed the session before
// any other traffic (including the v2 overlay-protocol tree-
// request flow that pgh app install actually uses) could flow.
// v1 onFileRequest below still serves individual file lookups if
// a v1-only peer ever asks; only the eager dump is gone.
// Handle file requests
proto.onFileRequest(async (req) => {
if (session.destroyed) return
this.emit('peer', { key: peerKey, state: 'transferring' })
const fullPath = path.join(this.dir, req.path)
let stat
try { stat = fs.statSync(fullPath) } catch {
await proto.sendControl({ type: 'error', message: 'File not found', path: req.path })
return
}
if (stat.size >= CHUNK_THRESHOLD) {
// Large file: use TransferManager chunking
const prepared = await this._transferManager.prepareFile(fullPath, req.path)
if (!prepared) {
await proto.sendControl({ type: 'error', message: 'Prepare failed', path: req.path })
return
}
await proto.sendChunkInfo({
path: req.path,
size: prepared.size,
chunks: prepared.chunks
})
for (let i = 0; i < prepared.chunks.length; i++) {
if (session.destroyed) break
const bytes = this._transferManager.readChunk(fullPath, prepared.chunks[i].offset, prepared.chunks[i].length)
if (!bytes) continue
await proto.sendChunkData(i, bytes)
session.bytesServed += bytes.length
this.stats.bytesServed += bytes.length
}
} else {
// Small file: send as single whole-file blob
await proto.sendChunkInfo({
path: req.path,
size: stat.size,
chunks: [{ hash: hashFile(fullPath), offset: 0, length: stat.size }]
})
const data = fs.readFileSync(fullPath)
await proto.sendChunkData(0, data)
session.bytesServed += data.length
this.stats.bytesServed += data.length
}
await proto.sendControl({ type: 'file-end', path: req.path })
session.filesServed++
this.stats.filesServed++
this.emit('progress', {
peer: peerKey,
file: req.path,
filesServed: session.filesServed,
bytesServed: session.bytesServed
})
})
// Handle control messages
proto.onControl((msg) => {
if (msg.type === 'up-to-date') {
this.emit('peer', { key: peerKey, state: 'done' })
} else if (msg.type === 'done') {
this.emit('peer', { key: peerKey, state: 'done' })
this._sessions.delete(session)
}
})
}
}
// ── DeployClient ─────────────────────────────────────────────
export class DeployClient extends ReadyResource {
constructor (topic, destDir, opts) {
super()
opts = opts || {}
this.topic = typeof topic === 'string' ? Buffer.from(topic, 'hex') : topic
this.destDir = path.resolve(destDir)
this.swarm = opts.swarm || null
this._ownSwarm = !this.swarm
this._corestore = opts.corestore || null
this._ownStore = !this._corestore
this._fileIndex = null
this._transferManager = null
this._timeout = opts.timeout !== undefined ? opts.timeout : 30000
this._verify = opts.verify !== false
this.connected = false
this.stats = {
filesTransferred: 0,
filesSkipped: 0,
bytesReceived: 0,
totalFiles: 0,
elapsed: 0,
errors: []
}
this._pullResolve = null
this._pullReject = null
this._startTime = null
this._connectTimer = null
this._completed = false
this._completedStats = null
this.ready().catch(safetyCatch)
}
async _open () {
fs.mkdirSync(this.destDir, { recursive: true })
// Setup corestore + overlay primitives
if (!this._corestore) {
const tmp = path.join(os.tmpdir(), 'hyper-deploy-cli-' + Date.now())
this._corestore = new Corestore(tmp)
}
this._fileIndex = new FileIndex(this._corestore)
await this._fileIndex.ready()
this._transferManager = new TransferManager(this._fileIndex)
// Start swarm
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()
}
async _close () {
if (this._connectTimer) {
clearTimeout(this._connectTimer)
this._connectTimer = null
}
if (this._pullReject) {
this._pullReject(new Error('Client closed during pull'))
this._pullReject = null
this._pullResolve = null
}
if (this._fileIndex) await this._fileIndex.close()
if (this._ownStore && this._corestore) await this._corestore.close()
if (this._ownSwarm && this.swarm) {
await this.swarm.destroy()
this.swarm = null
}
}
pull () {
// Already completed (transfer happened during ready/connection)
if (this._completed) return Promise.resolve(this._completedStats)
if (this._pullResolve) return Promise.reject(new Error('Pull already in progress'))
if (!this._startTime) this._startTime = Date.now()
return new Promise((resolve, reject) => {
this._pullResolve = resolve
this._pullReject = reject
if (this._timeout > 0 && !this.connected) {
this._connectTimer = setTimeout(() => {
if (!this.connected && this._pullReject) {
this._pullReject(new Error('Connection timeout after ' + this._timeout + 'ms'))
this._pullReject = null
this._pullResolve = null
}
}, this._timeout)
}
})
}
_finish () {
if (this._completed) return
this.stats.elapsed = Date.now() - (this._startTime || Date.now())
this._completed = true
this._completedStats = { ...this.stats }
if (this._pullResolve) {
const resolve = this._pullResolve
this._pullResolve = null
this._pullReject = null
resolve(this._completedStats)
}
this.emit('complete', this._completedStats)
}
_fail (err) {
this.stats.elapsed = Date.now() - (this._startTime || Date.now())
this.emit('error', err)
if (this._pullReject) {
const reject = this._pullReject
this._pullReject = null
this._pullResolve = null
reject(err)
}
}
_onConnection (conn) {
if (this._connectTimer) {
clearTimeout(this._connectTimer)
this._connectTimer = null
}
this.connected = true
if (!this._startTime) this._startTime = Date.now()
const peerKey = conn.remotePublicKey
? conn.remotePublicKey.toString('hex').slice(0, 12)
: 'unknown'
this.emit('connected', peerKey)
conn.on('close', () => {
this.connected = false
if (this._pullReject && this.stats.filesTransferred + this.stats.filesSkipped < this.stats.totalFiles) {
this._fail(new Error('Connection lost during transfer'))
}
})
conn.on('error', (err) => {
this.connected = false
this._fail(err)
})
const proto = attachProtocol(conn)
if (!proto) {
this._fail(new Error('Failed to attach protocol'))
return
}
// State for current file transfer
let pendingFiles = []
let currentFileIdx = 0
let currentPath = null
// Receive file list from server, diff against local
proto.onFileList((files) => {
this.stats.totalFiles = files.length
this.emit('manifest', files.length)
// Diff: which files do we need?
const needed = []
let skipped = 0
for (const f of files) {
const localPath = path.join(this.destDir, f.path)
try {
const stat = fs.statSync(localPath)
if (stat.size === f.size) {
const localHash = hashFile(localPath)
if (localHash === f.contentHash) {
skipped++
continue
}
}
} catch {}
needed.push(f)
}
this.stats.filesSkipped = skipped
this.emit('diff', needed.length, skipped)
if (needed.length === 0) {
proto.sendControl({ type: 'up-to-date' }).catch(safetyCatch)
this.emit('up-to-date')
this._finish()
return
}
pendingFiles = needed
currentFileIdx = 0
// Request first file
proto.sendFileRequest(needed[0].path).catch(safetyCatch)
})
// Receive chunk info for a file — setup TransferManager receive
proto.onChunkInfo((info) => {
currentPath = info.path
const targetPath = path.join(this.destDir, info.path)
this._transferManager.startReceive(targetPath, {
size: info.size,
chunks: info.chunks
})
})
// Receive chunk data
proto.onChunkData((index, data) => {
if (!currentPath) return
const targetPath = path.join(this.destDir, currentPath)
const result = this._transferManager.writeChunk(targetPath, index, data)
if (!result.ok) {
this.stats.errors.push({ path: currentPath, message: result.error })
this.emit('error', new Error(result.error))
}
this.stats.bytesReceived += data.length
})
// Receive control messages (file-end, error, etc.)
proto.onControl((msg) => {
if (msg.type === 'file-end') {
const targetPath = path.join(this.destDir, msg.path)
if (this._transferManager.isComplete(targetPath)) {
const fin = this._transferManager.finalize(targetPath)
if (!fin.ok) {
this.stats.errors.push({ path: msg.path, message: fin.error })
}
}
// Post-transfer verification (whole-file hash)
if (this._verify && pendingFiles[currentFileIdx]) {
try {
const actual = hashFile(targetPath)
const expected = pendingFiles[currentFileIdx].contentHash
if (actual !== expected) {
const err = 'Hash mismatch for ' + msg.path
this.stats.errors.push({ path: msg.path, message: err })
this.emit('error', new Error(err))
try { fs.unlinkSync(targetPath) } catch {}
}
} catch (err) {
this.stats.errors.push({ path: msg.path, message: err.message })
}
}
this.stats.filesTransferred++
currentPath = null
this.emit('progress', {
file: msg.path,
filesTransferred: this.stats.filesTransferred,
filesSkipped: this.stats.filesSkipped,
bytesReceived: this.stats.bytesReceived,
totalFiles: this.stats.totalFiles
})
// Request next file or finish
currentFileIdx++
if (currentFileIdx < pendingFiles.length) {
proto.sendFileRequest(pendingFiles[currentFileIdx].path).catch(safetyCatch)
} else {
proto.sendControl({ type: 'done' }).catch(safetyCatch)
this._finish()
}
} else if (msg.type === 'error') {
this.stats.errors.push({ path: msg.path, message: msg.message })
this.emit('error', new Error(msg.message))
}
})
}
}