// ReleasePuller — the rule that decides whether a peer overwrites local bytes,
// and the index that makes discovery possible at all.
//
// Context that shaped this (measured 2026-08-07, not assumed):
// - The deploy topic is each daemon's own identity, so two backends never
// replicate. tabertron held 16 repos / 558 versions, stortron 2 / 4 —
// neither a superset, and the ACTIVE public backend was the empty one.
// - DeployClient.pull() cannot be used: it waits on onFileList, and
// hyper-deploy 0.4.1 removed the only caller of sendFileList.
// - Two honest peers routinely hold the same (repo, version) with DIFFERENT
// bytes. hardops.zip diverged at IDENTICAL SIZE (6,326,509) with different
// hashes, so a size-only diff is not merely imprecise, it is wrong on real
// data. That case is pinned below.
import test from 'brittle'
import { mkdtempSync, readFileSync, writeFileSync, mkdirSync, existsSync } from '#node:fs'
import { join } from '#node:path'
import { tmpdir } from '#node:os'
import b4a from 'b4a'
import { ReleaseManager } from '../lib/release-manager.js'
import { ReleasePuller, planPull } from '../lib/services/release-puller.js'
function tmpDir (p) { return mkdtempSync(join(tmpdir(), p)) }
// ── the planner, in isolation ──────────────────────────────────
test('planPull: missing → fetch, identical → skip, differing → replace', (t) => {
const remote = {
files: [
{ path: 'r/versions/v1/new.zip', size: 10, hash: 'aaa' },
{ path: 'r/versions/v1/same.zip', size: 20, hash: 'bbb' },
{ path: 'r/versions/v1/diff.zip', size: 30, hash: 'ccc' }
]
}
const local = (p) => ({
'r/versions/v1/same.zip': { size: 20, hash: 'bbb' },
'r/versions/v1/diff.zip': { size: 30, hash: 'DIFFERENT' }
})[p] || null
const plan = planPull(remote, local)
t.is(plan.fetch.length, 1, 'one missing file to fetch')
t.is(plan.fetch[0].path, 'r/versions/v1/new.zip')
t.is(plan.skip.length, 1, 'identical file skipped')
t.is(plan.replace.length, 1, 'differing file replaced — publisher is authoritative')
t.is(plan.replace[0].path, 'r/versions/v1/diff.zip')
t.is(plan.suspect.length, 0, 'not flagged: nothing claimed to be reproducible')
})
test('planPull: identical SIZE with a different hash still replaces', (t) => {
// The real hardops.zip case. A size-only diff would call these equal.
const remote = { files: [{ path: 'hardops/versions/v1/hardops.zip', size: 6326509, hash: '17f9e403' }] }
const local = () => ({ size: 6326509, hash: 'b9dac90e' })
const plan = planPull(remote, local)
t.is(plan.skip.length, 0, 'NOT skipped despite equal size')
t.is(plan.replace.length, 1, 'replaced on hash difference alone')
})
test('planPull: a reproducible release flags mismatch as signal, and still replaces', (t) => {
const remote = { files: [{ path: 'a/versions/v1/x.zip', size: 5, hash: 'aaa', reproducible: true }] }
const plan = planPull(remote, () => ({ size: 5, hash: 'zzz' }))
t.is(plan.suspect.length, 1, 'surfaced — a reproducible recipe should have agreed')
t.is(plan.suspect[0].localHash, 'zzz', 'carries both sides for the operator')
t.is(plan.replace.length, 1, 'publisher still wins; the flag is a warning, not a veto')
})
test('planPull: entries without a hash are ignored, not blindly fetched', (t) => {
const plan = planPull({ files: [{ path: 'a/versions/v1/x.zip', size: 5 }] }, () => null)
t.is(plan.fetch.length, 0, 'no hash means no way to verify — skip it')
})
// ── the index ──────────────────────────────────────────────────
test('_index.json advertises hashes and reuses them across rebuilds', async (t) => {
const dir = tmpDir('rm-idx-')
const rm = new ReleaseManager({ releaseDir: dir })
await rm.publish({
repo: 'demo',
version: 'v1',
channel: 'stable',
artifacts: [{ path: 'a.zip', data: b4a.from('hello') }, { path: 'b.txt', data: b4a.from('x') }]
})
const idx = JSON.parse(readFileSync(join(dir, '_index.json'), 'utf8'))
t.is(idx.schema, 1)
t.is(idx.files.length, 3, 'both artifacts indexed, plus manifest.json')
t.ok(idx.files.every(f => f.hash && f.size >= 0), 'every entry carries a hash')
t.is(idx.channels.length, 1, 'channel pointer travels with the index')
t.is(idx.channels[0].version, 'v1')
const firstHash = idx.files.find(f => f.path === 'demo/versions/v1/a.zip').hash
await rm.publish({ repo: 'demo', version: 'v2', artifacts: [{ path: 'a.zip', data: b4a.from('hello2') }] })
const idx2 = JSON.parse(readFileSync(join(dir, '_index.json'), 'utf8'))
t.is(idx2.files.length, 5, 'new version added (v1: 2+manifest, v2: 1+manifest)')
t.is(idx2.files.find(f => f.path === 'demo/versions/v1/a.zip').hash, firstHash,
'unchanged artifact keeps its cached hash — a 244 MB tarball is not rehashed per publish')
t.not(idx2.files.find(f => f.path === 'demo/versions/v2/a.zip').hash, firstHash,
'different content, different hash')
})
test('_writeIndex only re-registers CHANGED entries in the overlay', async (t) => {
// _writeIndex runs on every mutation, and a mutation happens on every
// push-triggered build. A forge with 558 versions re-registering everything
// each time would issue ~560 overlay index writes for content that has not
// moved. Unchanged entries must be skipped.
const dir = tmpDir('rm-reg-')
const calls = []
const overlay = { async registerFile (p, disk, meta) { calls.push(p); return { ok: true } } }
const rm = new ReleaseManager({ releaseDir: dir, overlay })
await rm.publish({ repo: 'demo', version: 'v1', artifacts: [{ path: 'a.zip', data: b4a.from('one') }] })
const first = calls.length
t.is(first, 2, 'first publish registers the artifact + its manifest')
calls.length = 0
await rm.publish({ repo: 'demo', version: 'v2', artifacts: [{ path: 'a.zip', data: b4a.from('two') }] })
t.is(calls.length, 2, 'second publish registers ONLY the new version, not v1 again')
t.ok(calls.every(p => p.includes('/v2/')), 'and nothing from v1 was touched')
t.ok(calls.every(p => p.startsWith('release:')), 'namespaced away from the attachment spool')
})
// ── end to end, with a fake transport ──────────────────────────
// Stands in for overlayV2.fetchFile — content-addressed, hash-verified. The
// real one pulls chunked over the swarm; the contract we depend on is
// "give me this hash, write it to destPath".
function fakeOverlay (publisherDir, index) {
const byHash = new Map()
for (const f of index.files) byHash.set(f.hash, join(publisherDir, f.path))
return {
fetched: [],
async fetchFile (hash, opts = {}) {
const src = byHash.get(hash)
if (!src) return null
const buf = readFileSync(src)
writeFileSync(opts.destPath, buf)
this.fetched.push(hash)
return { destPath: opts.destPath, local: false, size: buf.length }
}
}
}
test('pullFrom merges a publisher into an empty peer, then the peer can serve it', async (t) => {
const pubDir = tmpDir('rp-pub-')
const pub = new ReleaseManager({ releaseDir: pubDir })
await pub.publish({
repo: 'demo',
version: 'v1',
channel: 'stable',
artifacts: [{ path: 'app.zip', data: b4a.from('PAYLOAD') }, { path: 'notes.txt', data: b4a.from('n') }]
})
const pubIndex = JSON.parse(readFileSync(join(pubDir, '_index.json'), 'utf8'))
const peerDir = tmpDir('rp-peer-')
const peerRm = new ReleaseManager({ releaseDir: peerDir })
const overlay = fakeOverlay(pubDir, pubIndex)
const logs = []
const puller = new ReleasePuller({
releaseDir: peerDir, overlay, releases: peerRm, log: (l) => logs.push(l)
})
const client = { call: async (m) => { t.is(m, 'release-index', 'discovery via RPC'); return pubIndex } }
const res = await puller.pullFrom(client)
t.ok(res.ok, 'pull succeeded')
// 2 artifacts + manifest.json. The manifest travels too, or the version
// reads as `incomplete` on this side and never gets re-served.
t.is(puller.stats.fetched, 3, 'both artifacts AND the manifest pulled')
t.is(puller.stats.replaced, 0, 'nothing replaced on an empty peer')
t.ok(existsSync(join(peerDir, 'demo/versions/v1/manifest.json')), 'manifest landed')
t.alike(readFileSync(join(peerDir, 'demo/versions/v1/app.zip')), b4a.from('PAYLOAD'), 'bytes landed')
t.ok(existsSync(join(peerDir, 'demo/channels/stable.json')), 'channel pointer written')
t.absent(existsSync(join(peerDir, 'demo/versions/v1/app.zip.pull-tmp')), 'no temp file left behind')
// The peer must now advertise what it pulled, or replication is a star.
const peerIndex = JSON.parse(readFileSync(join(peerDir, '_index.json'), 'utf8'))
t.is(peerIndex.files.length, 3, 'peer re-indexed and can now serve what it pulled')
t.alike(
peerIndex.files.map(f => f.hash).sort(),
pubIndex.files.map(f => f.hash).sort(),
'peer now advertises byte-identical content to the publisher')
})
test('pullFrom replaces a locally-built differing copy, and skips identical ones', async (t) => {
const pubDir = tmpDir('rp-pub2-')
const pub = new ReleaseManager({ releaseDir: pubDir })
await pub.publish({
repo: 'demo',
version: 'v1',
artifacts: [{ path: 'app.zip', data: b4a.from('PUBLISHER') }, { path: 'same.txt', data: b4a.from('same') }]
})
const pubIndex = JSON.parse(readFileSync(join(pubDir, '_index.json'), 'utf8'))
// Peer built its own v1: one artifact differs, one is byte-identical.
const peerDir = tmpDir('rp-peer2-')
const peerRm = new ReleaseManager({ releaseDir: peerDir })
await peerRm.publish({
repo: 'demo',
version: 'v1',
artifacts: [{ path: 'app.zip', data: b4a.from('LOCALBUILD') }, { path: 'same.txt', data: b4a.from('same') }]
})
const overlay = fakeOverlay(pubDir, pubIndex)
const puller = new ReleasePuller({ releaseDir: peerDir, overlay, releases: peerRm })
const client = { call: async () => pubIndex }
const res = await puller.pullFrom(client)
t.ok(res.ok)
// app.zip differs; so does manifest.json (different builtAt on each build).
// Both are replaced — the publisher is authoritative for the whole version,
// its manifest included.
t.is(puller.stats.replaced, 2, 'differing artifact AND manifest replaced')
t.is(puller.stats.skipped, 1, 'the byte-identical artifact was not refetched')
t.alike(readFileSync(join(peerDir, 'demo/versions/v1/app.zip')), b4a.from('PUBLISHER'),
'publisher is authoritative — a puller never authors a version')
})
test('dryRun reports the plan and writes nothing', async (t) => {
const pubDir = tmpDir('rp-pub3-')
const pub = new ReleaseManager({ releaseDir: pubDir })
await pub.publish({ repo: 'demo', version: 'v1', artifacts: [{ path: 'app.zip', data: b4a.from('X') }] })
const pubIndex = JSON.parse(readFileSync(join(pubDir, '_index.json'), 'utf8'))
const peerDir = tmpDir('rp-peer3-')
const overlay = fakeOverlay(pubDir, pubIndex)
const puller = new ReleasePuller({ releaseDir: peerDir, overlay })
const res = await puller.pullFrom({ call: async () => pubIndex }, { dryRun: true })
t.ok(res.dryRun, 'reported as a dry run')
t.is(res.wouldFetch, 2, 'the artifact and its manifest')
t.absent(existsSync(join(peerDir, 'demo/versions/v1/app.zip')), 'nothing written')
t.is(overlay.fetched.length, 0, 'no bytes requested')
})
test('a missing holder is an error, not a truncated artifact', async (t) => {
const peerDir = tmpDir('rp-peer4-')
const index = { schema: 1, files: [{ path: 'demo/versions/v1/app.zip', size: 3, hash: 'nope' }], channels: [] }
const overlay = { async fetchFile () { return null } } // nobody has it
const puller = new ReleasePuller({ releaseDir: peerDir, overlay })
const res = await puller.pullFrom({ call: async () => index })
t.ok(res.ok, 'the pull as a whole still completes')
t.is(puller.stats.fetched, 0, 'nothing counted as fetched')
t.is(puller.stats.errors.length, 1, 'the gap is reported')
t.absent(existsSync(join(peerDir, 'demo/versions/v1/app.zip')), 'no partial file left on disk')
t.absent(existsSync(join(peerDir, 'demo/versions/v1/app.zip.pull-tmp')), 'temp cleaned up')
})
test('an origin too old to serve release-index fails clearly', async (t) => {
const peerDir = tmpDir('rp-peer5-')
const puller = new ReleasePuller({ releaseDir: peerDir, overlay: { async fetchFile () { return null } } })
const client = { call: async () => { throw new Error('unknown method: release-index') } }
const res = await puller.pullFrom(client)
t.absent(res.ok, 'reported as a failure')
t.ok(/release-index/.test(res.reason), 'names the missing RPC rather than failing opaquely')
})
test('a channel is not pinned to a version whose bytes never landed', async (t) => {
// Guards the ordering: pointers are written last, and only for versions we
// can actually serve. A pointer to absent bytes advertises a broken download.
const peerDir = tmpDir('rp-peer6-')
const index = {
schema: 1,
files: [{ path: 'demo/versions/v9/app.zip', size: 3, hash: 'missing' }],
channels: [{ repo: 'demo', channel: 'stable', version: 'v9' }]
}
const puller = new ReleasePuller({
releaseDir: peerDir, overlay: { async fetchFile () { return null } }
})
await puller.pullFrom({ call: async () => index })
t.absent(existsSync(join(peerDir, 'demo/channels/stable.json')),
'no pointer to a version we could not fetch')
})
// The scheduler is policy, the puller is mechanism. These pin the policy:
// never pulls on its own (there is no interval any more — only events), only
// ever pulls from a configured ORIGIN, and never overlaps itself.
test('startReleasePull does nothing until an event arrives — there is no interval', async (t) => {
const { startReleasePull } = await import('../lib/services/release-puller.js')
const calls = []
const puller = { pullFrom: async () => { calls.push(1); return { ok: true, stats: {} } } }
const handle = startReleasePull({
puller, originServerKey: 'a'.repeat(64), connect: async () => ({ close: async () => {} })
})
t.ok(handle, 'a handle exists')
t.absent(handle.enabled, 'no enabled flag: the PGV3_RELEASE_PULL timer gate is gone entirely')
// The old design sampled every 15 minutes whether or not anything changed.
// Waiting well past any plausible debounce must still yield nothing, because
// nothing has happened.
await new Promise(r => setTimeout(r, 60))
t.is(calls.length, 0, 'a quiet origin costs a quiet mirror ZERO work')
// Manual pull stays available for operators — a dry run against a live daemon
// must need no config change.
await handle.pullNow()
t.is(calls.length, 1, 'manual pullNow still works')
handle.stop()
})
test('a burst of events coalesces into ONE scoped reconcile', async (t) => {
const { startReleasePull } = await import('../lib/services/release-puller.js')
const seen = []
const puller = {
pullFrom: async (client, opts) => {
seen.push(opts && opts.repos)
return { ok: true, stats: { fetched: 1, replaced: 0, skipped: 0, errors: [] } }
}
}
const handle = startReleasePull({
puller,
originServerKey: 'd'.repeat(64),
connect: async () => ({ close: async () => {} }),
debounceMs: 20
})
t.teardown(() => handle.stop())
// CI publishing three versions of two repos back to back: six entries.
handle.notify({ repo: 'alpha' })
handle.notify({ repo: 'alpha' })
handle.notify({ repo: 'beta' })
handle.notify({ repo: 'alpha' })
handle.notify({ repo: 'beta' })
handle.notify({ repo: 'beta' })
await new Promise(r => setTimeout(r, 120))
t.is(seen.length, 1, 'six entries → ONE pull; the debounce is a coalescer, not a poll')
t.alike([...seen[0]].sort(), ['alpha', 'beta'], 'scoped to exactly the repos that changed')
})
test('an event landing DURING a pull gets its own pass', async (t) => {
const { startReleasePull } = await import('../lib/services/release-puller.js')
const seen = []
let release = null
const puller = {
pullFrom: async (client, opts) => {
seen.push(opts && opts.repos)
// Hold the first pull open so the second event provably arrives mid-flight.
if (seen.length === 1) await new Promise(r => { release = r })
return { ok: true, stats: { fetched: 0, replaced: 0, skipped: 0, errors: [] } }
}
}
const handle = startReleasePull({
puller,
originServerKey: 'e'.repeat(64),
connect: async () => ({ close: async () => {} }),
debounceMs: 10
})
t.teardown(() => handle.stop())
handle.notify({ repo: 'alpha' })
await new Promise(r => setTimeout(r, 40))
t.is(seen.length, 1, 'first pull in flight')
// This mutation may post-date the index the in-flight pull already read, so
// swallowing it would leave the mirror silently behind with no later trigger.
handle.notify({ repo: 'beta' })
release()
await new Promise(r => setTimeout(r, 120))
t.is(seen.length, 2, 'the mid-flight event is not swallowed')
t.alike(seen[1], ['beta'], 'and the second pass is scoped to it')
})
test('a scoped pull ignores repos the event did not name', async (t) => {
const pubDir = tmpDir('relpub-scope-')
const peerDir = tmpDir('relpeer-scope-')
const pubRm = new ReleaseManager({ releaseDir: pubDir })
await pubRm.publish({ repo: 'alpha', version: 'v1', artifacts: [{ path: 'a.zip', data: b4a.from('AAA') }] })
await pubRm.publish({ repo: 'beta', version: 'v1', artifacts: [{ path: 'b.zip', data: b4a.from('BBB') }] })
const pubIndex = JSON.parse(readFileSync(join(pubDir, '_index.json'), 'utf8'))
const puller = new ReleasePuller({ releaseDir: peerDir, overlay: fakeOverlay(pubDir, pubIndex) })
const res = await puller.pullFrom({ call: async () => pubIndex }, { repos: ['alpha'] })
t.ok(res.ok)
t.ok(existsSync(join(peerDir, 'alpha/versions/v1/a.zip')), 'the named repo is reconciled')
t.absent(existsSync(join(peerDir, 'beta/versions/v1/b.zip')),
'the unnamed repo is untouched — cost is O(what changed), not O(every release ever)')
})
test('a size mismatch is decided WITHOUT hashing the local file', async (t) => {
// The diff used to hash every local counterpart unconditionally. On a
// backfilled mirror that meant re-reading the whole release dir (~286 MiB
// measured) to reach a verdict a stat already settles.
const peerDir = tmpDir('relpeer-sizefirst-')
mkdirSync(join(peerDir, 'demo/versions/v1'), { recursive: true })
writeFileSync(join(peerDir, 'demo/versions/v1/app.zip'), 'SHORT')
writeFileSync(join(peerDir, 'demo/versions/v1/same.zip'), 'IDENTICAL')
const hashed = []
const puller = new ReleasePuller({
releaseDir: peerDir,
overlay: { async fetchFile () { return null } },
hashFile: async (abs) => { hashed.push(abs); return 'h:' + readFileSync(abs, 'utf8') }
})
const index = {
files: [
// differs in size → verdict needs no hash
{ path: 'demo/versions/v1/app.zip', repo: 'demo', version: 'v1', size: 999, hash: 'remote' },
// same size → the hash is genuinely required to tell them apart
{ path: 'demo/versions/v1/same.zip', repo: 'demo', version: 'v1', size: 9, hash: 'h:IDENTICAL' }
]
}
const res = await puller.pullFrom({ call: async () => index }, { dryRun: true })
t.is(res.wouldSkip, 1, 'the byte-identical file is skipped')
t.is(res.plan.replace.length, 1, 'the size-mismatched file is still replaced')
t.is(hashed.length, 1, 'exactly ONE hash — only the same-size candidate')
t.ok(hashed[0].endsWith('same.zip'), 'and it is the one whose size matched')
})
test('startReleasePull refuses to run without an origin', async (t) => {
const { startReleasePull } = await import('../lib/services/release-puller.js')
const logs = []
const handle = startReleasePull({
puller: { pullFrom: async () => ({ ok: true }) },
originServerKey: null, // a PRIMARY has none
connect: async () => ({}),
log: (l) => logs.push(l)
})
t.is(handle, null, 'no origin, no puller — never invents a peer to pull from')
t.ok(logs.some(l => /not a mirror/.test(l)), 'and says why')
})
test('pullNow does not overlap itself', async (t) => {
const { startReleasePull } = await import('../lib/services/release-puller.js')
let active = 0
let maxActive = 0
const puller = {
async pullFrom () {
active++; maxActive = Math.max(maxActive, active)
await new Promise(r => setTimeout(r, 40))
active--
return { ok: true, stats: { fetched: 0, replaced: 0, skipped: 0, errors: [] } }
}
}
const handle = startReleasePull({
puller,
originServerKey: 'b'.repeat(64),
connect: async () => ({ close: async () => {} }),
})
t.teardown(() => handle.stop())
const [a, b] = await Promise.all([handle.pullNow(), handle.pullNow()])
t.is(maxActive, 1, 'a large artifact can outlast the interval — the second call must not race it')
const refused = [a, b].filter(r => !r.ok && /already in progress/.test(r.reason || ''))
t.is(refused.length, 1, 'the overlapping call is refused with a reason')
})
test('pullNow closes the client even when the pull throws', async (t) => {
const { startReleasePull } = await import('../lib/services/release-puller.js')
let closed = 0
const handle = startReleasePull({
puller: { pullFrom: async () => { throw new Error('boom') } },
originServerKey: 'c'.repeat(64),
connect: async () => ({ close: async () => { closed++ } }),
})
t.teardown(() => handle.stop())
const res = await handle.pullNow()
t.absent(res.ok, 'failure reported, not thrown')
t.is(closed, 1, 'connection released — a leaked swarm client per failed pull would accumulate')
})
// ── Tombstones ───────────────────────────────────────────────
//
// A version deleted on the origin must never be fetched again. Without this
// the puller chases bytes whose only holder is gone and retries forever:
// [release-pull] error: no holder for favela-android/versions/master-8d988ca/...
// → fetched 0, replaced 0, skipped 213, errors 1 — PARTIAL, will retry
// observed in prod 2026-08-29 after `pgh deploy gc` removed versions on one
// peer. repo-mirror.js already honours the equivalent repo tombstone
// (OP_RETIRED); the release path emitted OP_DELETED but nothing consumed it.
test('planPull: a tombstoned version is never fetched', (t) => {
const remote = {
files: [
{ path: 'r/versions/v1/a.zip', repo: 'r', version: 'v1', size: 10, hash: 'aaa' },
{ path: 'r/versions/v2/b.zip', repo: 'r', version: 'v2', size: 10, hash: 'bbb' }
]
}
const isDeleted = (repo, version) => repo === 'r' && version === 'v1'
const plan = planPull(remote, () => null, isDeleted)
t.is(plan.fetch.length, 1, 'only the live version is fetched')
t.is(plan.fetch[0].version, 'v2', 'and it is the one not tombstoned')
t.is(plan.tombstoned.length, 1, 'the dead version is accounted for, not silently dropped')
t.is(plan.tombstoned[0].version, 'v1')
})
test('planPull: a tombstone does not delete local bytes, it only stops fetching', (t) => {
// The never-delete rule: a tombstone changes what we PULL, never what is on
// disk. A local copy of a tombstoned version must not be planned for
// replacement or removal — byte removal stays a manual operator step.
const remote = {
files: [{ path: 'r/versions/v1/a.zip', repo: 'r', version: 'v1', size: 10, hash: 'aaa' }]
}
const localDiffers = () => ({ size: 99, hash: 'LOCAL-DIFFERENT' })
const plan = planPull(remote, localDiffers, () => true)
t.is(plan.fetch.length, 0, 'nothing fetched')
t.is(plan.replace.length, 0, 'local bytes NOT scheduled for replacement')
t.is(plan.tombstoned.length, 1, 'reported as tombstoned instead')
})
test('planPull: omitting the predicate keeps the old behaviour', (t) => {
const remote = {
files: [{ path: 'r/versions/v1/a.zip', repo: 'r', version: 'v1', size: 10, hash: 'aaa' }]
}
const plan = planPull(remote, () => null)
t.is(plan.fetch.length, 1, 'no predicate → fetch as before')
t.alike(plan.tombstoned, [], 'bucket present but empty')
})