/**
* ReleasePuller — replicate another daemon's release artifacts into ours.
*
* WHY IT IS NEEDED
*
* The deploy topic is derived from each daemon's OWN identity
* (`services/deploy.js`: `topic: identity.publicKey`), so two forge backends
* announce on different topics and never attempt replication. Measured
* 2026-08-07: tabertron held 16 repos / 558 versions, stortron 2 repos / 4
* versions — neither a superset, and the ACTIVE public backend was the one with
* almost nothing. A file-backed release is visible only on whichever daemon
* built it.
*
* WHAT IT IS BUILT ON
*
* NOT `DeployClient.pull()`. That waits for an `onFileList` message, and
* hyper-deploy 0.4.1 (5b80e84) removed the only caller of `sendFileList` — the
* unsolicited broadcast that was blowing secret-stream's 16 MiB atomic-write
* limit on a 100k-file dependency cache. `sendFileList` is dead code, so
* `pull()` hangs until timeout against any current server.
*
* Instead this reuses three things already proven in production:
*
* trigger — the origin's ReleaseEventLog (services/release-event-log.js), a
* single-writer Hypercore the mirror replicates like RepoEventLog
* and OrgEventLog. One entry per index mutation. This replaced a
* 15-minute polling interval; see startReleasePull below.
* discovery — the origin's `_index.json` (ReleaseManager._writeIndex),
* fetched over the existing authenticated hyper-svc RPC channel
* rather than guessed at. Bounded by RELEASE count, not by the
* shared dependency cache that broke v1.
* bytes — `overlayV2.fetchFile(contentHash, { destPath })`, the same
* content-addressed, hash-verified, chunked path that already
* moves attachments cross-daemon for /files/raw. hyper-svc
* auto-attaches the V2 overlay to every swarm connection, so the
* transport between two forges already exists.
*
* There is no payload ceiling on that path: chunk lists are paged specifically
* to stay under the 16 MiB limit. (pear-git-v3 #8 read that limit as a size
* ceiling and concluded replication broke above 16 MB; wrong on three counts,
* and the issue is closed with the correction.)
*
* CONFLICT RULE
*
* A puller NEVER authors a version. Where the local copy of a
* (repo, version, artifact) differs from the publisher's, the publisher's copy
* wins and the replacement is logged. Hash inequality is NOT treated as
* corruption: two honest peers routinely disagree, because builds are not
* byte-reproducible unless written to be — one observed divergence
* (hardops.zip) had IDENTICAL SIZE with a different hash. Where a release
* declares itself reproducible, a mismatch IS signal and is surfaced loudly.
*
* SEQUENCING
*
* This must land BEFORE any rule forbidding mirrors to build. Until a puller
* exists, `pgh deploy trigger` on the active backend is the only way to get an
* artifact served publicly; removing it first strands everything built on the
* standby.
*/
import fs from '#node:fs'
import { join, dirname } from '#node:path'
import { hashChunk } from 'hyper-overlay/chunker'
export const INDEX_FILE = '_index.json'
// Overlay paths are namespaced so release artifacts cannot collide with the
// attachment spool, which shares the same overlay instance.
export const OVERLAY_PREFIX = 'release:'
// Retry backoff bounds for a failed reconcile (see startReleasePull).
const MAX_BACKOFF_SHIFT = 5
const MAX_RETRY_MS = 5 * 60 * 1000
/**
* Decide what to fetch. Pure — no I/O, no network — so the rule that governs
* whether a peer overwrites local bytes is testable in isolation.
*
* @param {object} remoteIndex the publisher's parsed _index.json
* @param {(relPath: string) => ({size:number, hash:string}|null)} localStat
*/
/**
* Turn a failed fetchFile() into a message that says what actually happened.
*
* fetchFile returns null for three unrelated reasons and the caller used to
* report all of them as "no holder", which asserted a cause nobody measured.
* The scheduler already computes the distinction and hands it to onEnd; this
* just refuses to throw it away.
*
* end === null -> no scheduler ever ran. fetchFile gave up during its 3s
* peer wait, so the holder question was never asked.
* receivedBytes 0 -> peers were connected and none served a chunk. This is
* the only case the words "no holder" describe.
* receivedBytes>0 -> a holder existed and sent bytes, then the transfer
* stalled. A partial, not an absent holder.
*
* @param {object} f the index entry being fetched ({ path, hash })
* @param {object|null} end the scheduler's terminal diagnostic, if one ran
*/
export function describeFetchFailure (f, end) {
const where = (f && f.path) || '(unknown path)'
const hash = f && f.hash ? ' (' + String(f.hash).slice(0, 12) + ')' : ''
if (!end || typeof end !== 'object') {
return 'no peer connected for ' + where + hash +
' - gave up during the peer wait; the holder lookup never ran'
}
const peers = Number.isFinite(end.peers) ? end.peers : 0
const got = Number.isFinite(end.receivedBytes) ? end.receivedBytes : 0
const total = Number.isFinite(end.totalBytes) ? end.totalBytes : 0
const totalChunks = Number.isFinite(end.totalChunks) ? end.totalChunks : 0
const remaining = Number.isFinite(end.chunksRemaining) ? end.chunksRemaining : totalChunks
const done = Math.max(0, totalChunks - remaining)
const ms = Number.isFinite(end.elapsedMs) ? end.elapsedMs : 0
const why = end.reason ? ' - ' + end.reason : ''
if (got > 0) {
return 'stalled fetching ' + where + hash + ' - received ' + got + ' of ' + total +
' bytes, ' + done + ' of ' + totalChunks + ' chunks, from ' + peers +
' peer(s) in ' + ms + 'ms' + why
}
return 'no holder for ' + where + hash + ' - ' + peers +
' peer(s) connected, none served any chunk (' + done + ' of ' + totalChunks +
' chunks) after ' + ms + 'ms' + why
}
export function planPull (remoteIndex, localStat, isDeleted) {
const plan = { fetch: [], skip: [], replace: [], suspect: [], tombstoned: [] }
for (const f of (remoteIndex && remoteIndex.files) || []) {
if (!f || !f.path || !f.hash) continue
// Tombstone gate. A version the origin retired is never fetched again.
// Without this the puller chases bytes whose only holder is gone and
// retries forever -- observed in prod 2026-08-29 after `pgh deploy gc`
// removed versions on one peer:
// error: no holder for favela-android/versions/master-8d988ca/...
// fetched 0, replaced 0, skipped 213, errors 1 -- PARTIAL, will retry
// repo-mirror.js already honours the equivalent repo tombstone
// (OP_RETIRED); the release path emitted OP_DELETED and nothing consumed it.
//
// A tombstone changes what we PULL, never what is on disk: a local copy is
// NOT scheduled for replacement or removal. Byte removal stays a manual
// operator step on each peer -- see services/retired-repos.js.
//
// Checked before localStat so a dead version costs no I/O.
if (isDeleted && f.repo && f.version && isDeleted(f.repo, f.version)) {
plan.tombstoned.push(f)
continue
}
const local = localStat(f.path)
if (!local) { plan.fetch.push(f); continue }
if (local.size === f.size && local.hash === f.hash) { plan.skip.push(f); continue }
if (f.reproducible) {
plan.suspect.push({ ...f, localHash: local.hash, localSize: local.size })
}
plan.replace.push(f)
}
return plan
}
export class ReleasePuller {
/**
* @param {object} opts
* @param {string} opts.releaseDir local release dir (merge destination)
* @param {object} opts.overlay overlayV2 — needs fetchFile(hash, {destPath})
* @param {object} [opts.deploy] DeployServer, for rescan() after a merge
* @param {object} [opts.releases] ReleaseManager, to re-register pulled bytes
* @param {function} [opts.log]
* @param {function} [opts.hashFile] injectable for tests
*/
constructor (opts = {}) {
if (!opts.releaseDir) throw new Error('ReleasePuller requires { releaseDir }')
if (!opts.overlay) throw new Error('ReleasePuller requires { overlay }')
this._releaseDir = opts.releaseDir
this._overlay = opts.overlay
this._deploy = opts.deploy || null
this._releases = opts.releases || null
this._log = opts.log || (() => {})
this._hashFile = opts.hashFile || null
// Replicated tombstone set (services/deleted-releases.js). Optional: when
// absent the puller behaves exactly as before and will chase versions the
// origin has retired -- which is the 2026-08-29 incident. Wire it.
this._deleted = opts.deleted || null
this.stats = { pulls: 0, fetched: 0, skipped: 0, replaced: 0, suspect: 0, tombstoned: 0, bytes: 0, errors: [] }
}
async _hash (abs) {
if (this._hashFile) return this._hashFile(abs)
return hashChunk(fs.readFileSync(abs))
}
/**
* @param {string} relPath
* @param {number|null} [expectSize] the publisher's size for this path
*/
async _localStat (relPath, expectSize = null) {
const abs = join(this._releaseDir, relPath)
let st
try { st = fs.statSync(abs) } catch { return null }
if (!st.isFile()) return null
// Size-first. A size mismatch ALREADY proves the file differs, so hashing it
// is pure waste — and this ran on every local counterpart of every remote
// index entry. With the poll gone the diff runs per event rather than every
// 15 minutes, but a fully-backfilled mirror still has ~1120 counterparts, so
// an unconditional hash meant re-reading the whole release dir (~286 MiB)
// just to conclude "nothing changed". planPull treats a null hash with a
// mismatched size as a replace, which is the same verdict a full hash gives.
if (expectSize !== null && st.size !== expectSize) return { size: st.size, hash: null }
try { return { size: st.size, hash: await this._hash(abs) } } catch { return null }
}
async _buildLocalStat (remoteIndex) {
// Hash local counterparts ONCE, up front, so planPull can stay pure.
const map = new Map()
for (const f of (remoteIndex && remoteIndex.files) || []) {
if (!f || !f.path) continue
map.set(f.path, await this._localStat(f.path, typeof f.size === 'number' ? f.size : null))
}
return (p) => map.get(p) || null
}
/**
* Pull from one publisher.
*
* @param {object} client connected hyper-svc Client for the origin daemon
* @param {object} [opts] { dryRun }
*/
async pullFrom (client, opts = {}) {
const dryRun = !!opts.dryRun
this.stats.pulls++
let remoteIndex
try {
remoteIndex = await client.call('release-index', {})
} catch (err) {
const msg = 'release-index RPC failed: ' + (err.message || String(err)) +
' (an origin older than 0.1.206 does not serve it)'
this.stats.errors.push(msg)
this._log('[release-pull] ' + msg)
return { ok: false, reason: msg }
}
if (!remoteIndex || !Array.isArray(remoteIndex.files)) {
const msg = 'origin returned no usable release index'
this.stats.errors.push(msg)
return { ok: false, reason: msg }
}
// Scope to the repos named by the event(s) that woke us, so reconciliation
// costs O(what changed) rather than O(every release ever published). Absent
// a scope (manual full pull, or a cold mirror catching up) the whole index
// is considered, which is the old behaviour.
const scope = Array.isArray(opts.repos) && opts.repos.length
? new Set(opts.repos)
: null
if (scope) {
remoteIndex = {
...remoteIndex,
files: remoteIndex.files.filter(f => f && scope.has(f.repo)),
channels: (remoteIndex.channels || []).filter(c => c && scope.has(c.repo))
}
}
// `stats` is CUMULATIVE for the puller's lifetime and never reset, so
// reporting it per pull produced counts that only ever climbed — the live
// log read `[ifsrenderer-linux] fetched 61, skipped 1994` across DIFFERENT
// repos, none of which that pull touched. Snapshot here so the caller can
// report what THIS pull did while the lifetime totals stay available.
const before = {
fetched: this.stats.fetched,
replaced: this.stats.replaced,
skipped: this.stats.skipped,
suspect: this.stats.suspect,
bytes: this.stats.bytes,
errors: this.stats.errors.length
}
const delta = () => ({
fetched: this.stats.fetched - before.fetched,
replaced: this.stats.replaced - before.replaced,
skipped: this.stats.skipped - before.skipped,
suspect: this.stats.suspect - before.suspect,
bytes: this.stats.bytes - before.bytes,
errors: this.stats.errors.slice(before.errors)
})
const localStat = await this._buildLocalStat(remoteIndex)
const plan = planPull(remoteIndex, localStat, this._deleted ? this._deleted.isDeleted : null)
this.stats.skipped += plan.skip.length
this.stats.tombstoned += plan.tombstoned.length
// Surfaced, not silent: an operator needs to see that a version is being
// skipped BY POLICY rather than failing to transfer. The old failure mode
// was an unexplained `errors 1 -- PARTIAL, will retry` forever.
if (plan.tombstoned.length) {
const vs = [...new Set(plan.tombstoned.map(f => f.repo + '/' + f.version))]
this._log('[release-pull] skipping ' + plan.tombstoned.length + ' file(s) in ' +
vs.length + ' tombstoned version(s): ' + vs.slice(0, 5).join(', ') +
(vs.length > 5 ? ', +' + (vs.length - 5) + ' more' : ''))
}
for (const s of plan.suspect) {
this.stats.suspect++
this._log('[release-pull] ⚠ REPRODUCIBLE MISMATCH ' + s.path +
' local=' + String(s.localHash).slice(0, 12) +
' publisher=' + String(s.hash).slice(0, 12) +
' — a reproducible recipe should have produced identical bytes')
}
const todo = [...plan.fetch, ...plan.replace]
if (dryRun) {
return {
ok: true,
dryRun: true,
plan,
wouldFetch: todo.length,
wouldSkip: plan.skip.length,
repos: scope ? [...scope] : null
}
}
const replaceSet = new Set(plan.replace.map(f => f.path))
for (const f of todo) {
const abs = join(this._releaseDir, f.path)
try {
fs.mkdirSync(dirname(abs), { recursive: true })
// Land via a temp path then rename: a crash mid-transfer must not leave
// a truncated artifact that the next diff would read as complete.
const tmp = abs + '.pull-tmp'
// Capture the scheduler's terminal diagnostic so a failure can say which
// of the three fetchFile null-paths it actually was.
let endInfo = null
const got = await this._overlay.fetchFile(f.hash, {
destPath: tmp,
onEnd: (info) => { endInfo = info }
})
if (!got) {
this.stats.errors.push(describeFetchFailure(f, endInfo))
try { fs.unlinkSync(tmp) } catch {}
try { fs.unlinkSync(tmp + '.overlay-partial') } catch {}
continue
}
// fetchFile verifies incrementally against contentHash and rejects on
// mismatch, so arriving here means the bytes are already verified.
//
// But NOT necessarily at `tmp`. hyper-overlay short-circuits a content
// hash it already holds on disk and returns WHERE THOSE BYTES ARE,
// ignoring destPath entirely:
//
// const localDisk = this._contentHashPaths.get(contentHash)
// if (localDisk) { ... return { destPath: localDisk, local: true, ... } }
//
// Renaming `tmp` unconditionally therefore failed with ENOENT on the
// SOURCE — a temp file that was never written — and since every retry
// takes the same branch, the version stayed incomplete forever. It bit
// favela-android master-5d6d7e4 on the ACTIVE public backend for a day.
//
// The bytes still have to APPEAR at this version's path: downloads are
// served straight off disk by path (GET /releases/:repo/:version/:path),
// so an artifact held only under another version is unreachable here.
// They do not have to be a second copy — a hardlink satisfies the path
// and costs nothing, which matters because the artifacts that take this
// branch are exactly the ones UNCHANGED between builds.
const landed = got.destPath || tmp
if (landed === tmp) {
fs.renameSync(tmp, abs)
} else if (landed !== abs) {
// Link to our own temp and rename over, rather than unlinking `abs`
// first: that keeps the atomic-replace the rename above provides, and
// never removes a real artifact to make room for one that might fail.
const linkTmp = abs + '.pull-link'
try { fs.unlinkSync(linkTmp) } catch {}
try {
fs.linkSync(landed, linkTmp)
} catch (err) {
// Cross-device (or a filesystem without links): fall back to bytes.
if (err.code !== 'EXDEV' && err.code !== 'EPERM') throw err
fs.copyFileSync(landed, linkTmp)
}
fs.renameSync(linkTmp, abs)
try { fs.unlinkSync(tmp) } catch {}
}
if (replaceSet.has(f.path)) this.stats.replaced++
this.stats.fetched++
this.stats.bytes += (got.size || 0)
this._log('[release-pull] ' + (replaceSet.has(f.path) ? 'replaced ' : 'fetched ') +
f.path + ' (' + (got.size || 0) + ' B' + (got.local ? ', local' : '') + ')')
} catch (err) {
this.stats.errors.push(f.path + ': ' + (err.message || String(err)))
// hyper-overlay streams into `<destPath>.overlay-partial` before moving
// it into place, so removing only `<destPath>` left the partial behind —
// one failed 136 MiB APK fetch orphaned 136 MiB on disk, and every retry
// would add another.
try { fs.unlinkSync(abs + '.pull-tmp') } catch {}
try { fs.unlinkSync(abs + '.pull-tmp.overlay-partial') } catch {}
}
}
// Channel pointers LAST, and only for versions we can actually serve — a
// pointer to absent bytes advertises a broken download.
//
// Serveability is judged from the INDEX, not from the directory: the fetch
// loop above calls mkdirSync on the version dir before requesting bytes, so
// an existsSync on that directory is true even when every fetch failed.
// A version counts as serveable only if every file the publisher listed for
// it is present locally with the advertised size.
const serveable = new Set()
const byVersion = new Map()
for (const f of remoteIndex.files) {
if (!f || !f.repo || !f.version) continue
const k = f.repo + ' ' + f.version
if (!byVersion.has(k)) byVersion.set(k, [])
byVersion.get(k).push(f)
}
for (const [k, group] of byVersion) {
const complete = group.every(f => {
try { return fs.statSync(join(this._releaseDir, f.path)).size === f.size } catch { return false }
})
if (complete) serveable.add(k)
}
for (const c of remoteIndex.channels || []) {
try {
if (!c || !c.repo || !c.channel || !c.version) continue
if (!serveable.has(c.repo + ' ' + c.version)) {
this._log('[release-pull] channel ' + c.repo + '/' + c.channel +
' NOT pinned to ' + c.version + ' — not all of its files landed')
continue
}
const p = join(this._releaseDir, c.repo, 'channels', c.channel + '.json')
fs.mkdirSync(dirname(p), { recursive: true })
fs.writeFileSync(p, JSON.stringify({ version: c.version, pinnedAt: c.pinnedAt || Date.now() }, null, 2))
} catch (err) {
this.stats.errors.push('channel ' + c.repo + '/' + c.channel + ': ' + err.message)
}
}
// Re-index so this peer now SERVES what it pulled — otherwise replication
// is a star with the origin at the centre instead of gossip.
if (this._releases && typeof this._releases._writeIndex === 'function') {
try { await this._releases._writeIndex() } catch {}
}
if (this._deploy && typeof this._deploy.rescan === 'function') {
try { await this._deploy.rescan() } catch {}
}
// Every error path above pushes to stats.errors but none of them logged,
// so an operator saw a COUNT and never the text — and because the count is
// cumulative it then repeated on every later line forever ("errors 1" on
// stortron, whose artifact could no longer be identified from the logs).
const d = delta()
for (const e of d.errors) this._log('[release-pull] error: ' + e)
// PARTIAL is not success. This returned a bare { ok: true } however many
// individual files failed, and reconcile() re-queues only on `!ok` — so a
// pull that fetched one file of four reset the failure counter, drained the
// pending set, and stranded that version forever. The release-event cursor
// is already past it, so nothing else would ever mention that repo again.
//
// Observed on stortron: ifsrenderer-linux linux-port-14855c7 held 1 of 4
// files for ~12 hours with its channel still pinned to the previous release,
// while the bytes were fetchable the whole time. `ok` stays true because the
// pull itself ran — this is not a transport failure and the summary below is
// still worth logging — but `partial` tells the caller to come back.
return { ok: true, partial: d.errors.length > 0, plan, delta: d, stats: { ...this.stats } }
}
}
/**
* Event-driven pull from this daemon's origin.
*
* THERE IS NO TIMER. Reconciliation is triggered by `notify({repo})`, which the
* daemon calls for each entry it reads off the origin's replicated
* ReleaseEventLog — i.e. at the moment the origin's release index actually
* changed. The previous 15-minute polling interval (gated behind
* PGV3_RELEASE_PULL=1) is gone: it existed only because the push half of
* hyper-deploy was removed upstream, not because releases needed sampling.
*
* Consequences of the swap, all of them the point:
* - a publish reaches the mirror in seconds, not up to 15 minutes;
* - a quiet fleet does no work at all, instead of re-hashing the entire
* release dir every interval to discover nothing changed;
* - reconciliation is SCOPED to the repos named by the events, so cost is
* O(what changed);
* - catch-up after downtime is the log's job (a mirror streams from its
* persisted cursor), not a poll's.
*
* The debounce below is a COALESCER, not a poll: it batches a burst of entries
* (a publish appends one per mutation, and CI can publish several versions back
* to back) into one scoped reconcile. It fires once per burst and then stops —
* the same pattern as scheduleMirrorReannounce in pear-git.js.
*
* Manual `pullNow()` stays available and unscoped, for operators — writer-gated
* at the RPC, so a dry run against a live daemon needs no config change.
*
* It only ever pulls from the ORIGIN named in this daemon's own deployment
* bundle, never an arbitrary peer: a forge must not silently mirror a
* stranger's binaries. The origin's serverKey comes from the bundle
* (`originServerKey`), NOT from local identity — deriving a topic from local
* identity is precisely why the two backends never replicated in the first
* place.
*
* @returns {{ stop, pullNow, notify, pending } | null} null when not a mirror
*/
export function startReleasePull (opts = {}) {
const {
puller,
originServerKey,
connect, // async ({serverKey}) => hyper-svc Client
debounceMs = 2000,
log = () => {}
} = opts
if (!puller) throw new Error('startReleasePull requires { puller }')
if (!originServerKey) {
log('[release-pull] no origin configured — this daemon is not a mirror, nothing to pull')
return null
}
if (typeof connect !== 'function') throw new Error('startReleasePull requires { connect }')
let running = false
let stopped = false
let debounce = null
// Consecutive failed reconciles, for the retry backoff. Reset by any success.
let failures = 0
// Repos named by events that have not yet been reconciled.
const pending = new Set()
// Set when an event lands DURING a pull: that pull may have read the origin's
// index before the newer mutation was written, so it must run again. Same
// hazard RepoPuller handles with _needsRePull.
let again = false
async function pullNow (pullOpts = {}) {
// Never overlap: two concurrent pulls would race on the same temp paths.
if (running) return { ok: false, reason: 'a pull is already in progress' }
running = true
let client = null
try {
client = await connect({ serverKey: originServerKey })
const res = await puller.pullFrom(client, pullOpts)
if (res.ok && !res.dryRun) {
// THIS pull's numbers. `stats` is a lifetime accumulator, so reporting
// it here made every line climb regardless of what the pull did.
const s = res.delta || res.stats || {}
log('[release-pull] origin ' + originServerKey.slice(0, 12) +
(pullOpts.repos && pullOpts.repos.length ? ' [' + pullOpts.repos.join(',') + ']' : '') +
' → fetched ' + s.fetched + ', replaced ' + s.replaced +
', skipped ' + s.skipped + ', errors ' + (s.errors ? s.errors.length : 0) +
(res.partial ? ' — PARTIAL, will retry' : ''))
} else if (!res.ok) {
log('[release-pull] failed: ' + res.reason)
}
return res
} catch (err) {
log('[release-pull] error: ' + (err.message || String(err)))
return { ok: false, reason: err.message || String(err) }
} finally {
if (client) { try { await client.close() } catch {} }
running = false
}
}
async function reconcile () {
if (stopped || pending.size === 0) return
const repos = [...pending]
// Clear BEFORE the pull. An event arriving mid-pull re-populates `pending`
// and sets `again`, so it gets its own pass rather than being swallowed by
// this one — which may already have read a stale index.
pending.clear()
again = false
const res = await pullNow({ repos })
// A PARTIAL pull re-queues too. It is not a transport failure — the pull ran
// and fetched what it could — but some file the index named was not
// fetchable, most often because the origin had not finished registering it
// with the overlay when we asked. Treating that as success is what stranded
// versions permanently: `pending` was already cleared, `failures` reset, and
// the release-event cursor had moved past. Re-queuing routes it through the
// same capped backoff as any other failure, so a briefly-unservable artifact
// costs a few retries and an unrecoverable one settles at one attempt per
// MAX_RETRY_MS rather than hammering the origin.
if (!res.ok || res.partial) {
// ANY failure re-queues. This used to restore the work for exactly one
// reason ('a pull is already in progress') and drop it for every other —
// including the 15s `swarm-rpc: connection timeout` that fires when the
// mirror answers its own request for the origin, which on stortron was a
// third of all reconciles. Nothing replays a dropped event: the daemon
// saves the release-log cursor right after notify(), and notify only
// QUEUES, so the cursor is already past it, across restarts too.
for (const r of repos) pending.add(r)
again = true
failures++
} else {
failures = 0
}
if (again || pending.size > 0) schedule()
}
// Retry delay grows with consecutive failures so an origin that is genuinely
// unreachable is not hammered every debounce — but it is capped, and reset by
// any success, so a transient timeout costs one short wait rather than
// stranding the repo.
function retryDelay () {
if (failures === 0) return debounceMs
return Math.min(debounceMs * Math.pow(2, Math.min(failures, MAX_BACKOFF_SHIFT)), MAX_RETRY_MS)
}
function schedule () {
if (stopped || debounce) return
debounce = setTimeout(() => {
debounce = null
reconcile().catch((err) => log('[release-pull] reconcile error: ' + (err && err.message)))
}, retryDelay())
if (debounce && typeof debounce.unref === 'function') debounce.unref()
}
/**
* One release-index mutation observed on the origin's log.
* @param {object} e decoded ReleaseEventLog entry — only `repo` is used
*/
function notify (e = {}) {
if (stopped) return
const repo = e && e.repo
if (!repo) return
pending.add(repo)
if (running) again = true
schedule()
}
log('[release-pull] event-driven — origin ' + originServerKey.slice(0, 12) +
', reconciling on release-log events (no polling interval)')
return {
pullNow,
notify,
pending: () => [...pending],
stop () {
stopped = true
if (debounce) { clearTimeout(debounce); debounce = null }
}
}
}