/**
* RepoPuller — tracks a remote origin peer's RepoEventLog and pulls git
* repos from it on demand. One instance per configured `opts.origin`.
*
* Boots only when the orchestrator is configured with
* `opts.origin = { serverKey, repoEventLogKey }` (legacy `opts.mirror`
* still accepted as an alias).
*
* Naming history: was `RepoMirror` (rejected because "mirror" implied
* read-only) → `RepoSync` (rejected because it collides with the
* `hyper-sync` npm dep that does the actual git work) → `RepoPuller`,
* which describes what it actually does without overlap.
*
* What it does
* ------------
* - Registers the origin peer as a sync remote so `hyper-sync.pull(...)`
* knows where to fetch from.
* - Opens the origin's repo-event log by key — corestore replicates it
* automatically over the shared Hyperswarm topic.
* - Streams decoded entries (live) starting from the persisted cursor
* when `opts.cursorPath` is set, or from seq 0 on first boot.
* - For each entry, enqueues a pull. The queue is keyed by repo name —
* a flurry of pushes to the same repo collapses to one pull (latest
* wins). When a duplicate arrives while the in-flight pull is mid-
* fetch, a "needs re-pull" flag is set so we'll re-queue once the
* current pull finishes (the in-flight may have missed pushes that
* landed at origin AFTER its fetch started).
* - After a successful pull, persists the cursor (last successfully-
* processed seq per primaryServerKey) atomically.
*
* RepoPuller is idempotent — clone-if-missing + fetch-if-present —
* so even on a corrupt/missing cursor we can fall back to seq 0
* without harm.
*/
import b4a from 'b4a'
import fs from '#node:fs'
import path from '#node:path'
import { foldRetired, OP_RETIRED } from './retired-repos.js'
// Per-block deadline for the boot tombstone fold (see _runLoop). Generous:
// the log is metadata (~100 B/entry) and blocks arrive in batches once a peer
// is connected, so this only fires when a block is genuinely unreachable.
const FOLD_BLOCK_TIMEOUT_MS = 15000
export class RepoPuller {
constructor ({ sync, issueRegistry, log, primaryServerKey,
maxConcurrentClones = 3, cursorPath = null, onMount = null,
remoteName = 'primary', repoFilter = null, retired = null }) {
if (!sync) throw new Error('RepoPuller requires { sync }')
if (!log) throw new Error('RepoPuller requires { log }')
if (!primaryServerKey) throw new Error('RepoPuller requires { primaryServerKey }')
this._sync = sync
this._issueRegistry = issueRegistry
this._log = log
// Shared RetiredRepos set (see services/retired-repos.js). Optional: when
// absent the puller behaves exactly as before. Shared rather than owned so
// the boot-rescan and watcher gates observe the same tombstones this
// puller folds off the replicated log.
this._retired = retired
this._primaryServerKey = typeof primaryServerKey === 'string'
? primaryServerKey
: b4a.toString(primaryServerKey, 'hex')
// Name this puller registers its remote under in the sync server.
// Defaults to 'primary' for the classic single-origin mirror. A daemon
// that pulls from MULTIPLE peers (e.g. the source running reverse
// pullers toward each mirror — bidirectional repo discovery, #16) must
// give each puller a UNIQUE name or they'd clobber each other's remote
// entry (hyper-sync keys remotes by name).
this._remoteName = remoteName
// Optional scope filter (entry) => boolean. Default null = pull every repo
// the origin's RepoEventLog announces (classic full mirror). An org-scoped
// reverse puller passes a filter admitting ONLY repos in orgs this daemon
// has joined, preserving the per-org replication-scope guarantee (#35) so a
// co-member's non-org repos never reach this disk.
this._repoFilter = typeof repoFilter === 'function' ? repoFilter : null
this._max = maxConcurrentClones
this._cursorPath = cursorPath
// Fired with the repo name after a NEW repo is successfully pulled+bound,
// so the daemon can re-announce its mirror-writer keys (post-boot
// enrollment — see scheduleMirrorReannounce in pear-git.js).
this._onMount = typeof onMount === 'function' ? onMount : null
// Coalescing queue keyed by repo name → { entry, seq }. Latest entry
// for a given name wins; the older one is dropped (hyper-sync's
// `pull` is fetch-if-present, so the new fetch will pick up
// everything since the last fetched commit anyway).
this._queue = new Map()
this._inflight = 0
this._inflightNames = new Set()
// When a new entry arrives for a repo whose pull is already in-
// flight, we record the newer entry here. After the in-flight
// completes, _process re-enqueues. Origin may have pushed AFTER
// the in-flight fetch started, so the in-flight result is not
// necessarily the latest — re-pull catches that.
this._needsRePull = new Map()
this._started = false
this._stopped = false
this._cursor = this._loadCursor()
// Hook for tests — every clone attempt fires { name, ok, err? }.
this._onProgress = null
}
_loadCursor () {
if (!this._cursorPath) return {}
try {
const raw = fs.readFileSync(this._cursorPath, 'utf8')
const parsed = JSON.parse(raw)
if (parsed && typeof parsed === 'object') return parsed
} catch {
// Missing or corrupt → cold-boot from seq 0. Idempotent fetch
// means historical entries replay safely (just slow). Logged
// once so operators see "cold-boot: replaying from 0" in the
// daemon log when the cursor file is absent / unreadable.
}
return {}
}
_saveCursor () {
if (!this._cursorPath) return
try {
fs.mkdirSync(path.dirname(this._cursorPath), { recursive: true })
const tmp = this._cursorPath + '.tmp'
fs.writeFileSync(tmp, JSON.stringify(this._cursor, null, 2))
fs.renameSync(tmp, this._cursorPath)
} catch {
// Cursor write failure is non-fatal — worst case, next boot
// re-replays. Don't abort processing on a disk hiccup.
}
}
async start () {
if (this._started) return
this._started = true
// Register the origin so sync.pull(remoteName, name) can reach it.
if (!this._sync.remotes.has(this._remoteName)) {
this._sync.addRemote(this._remoteName, this._primaryServerKey)
}
// Fire-and-forget the live read loop. The caller (startPearGit)
// shouldn't block on the swarm having time to replicate any entries.
this._runLoop().catch((err) => {
// eslint-disable-next-line no-console
console.error('[repo-mirror] read loop exited:', err && err.message)
})
}
async _runLoop () {
// Prime the core: wait for it to know its real length from peers.
// Without this, a live read stream might start with length 0 and
// hang because the core never emits 'append' for blocks that
// arrived before subscription.
try {
await this._log.core.update({ wait: true })
} catch {}
// Resume from the persisted cursor. `cursor[serverKey] = N` means
// "we've successfully processed entries up to and including index
// N". Stream from N+1 onwards. Cold boot (no cursor entry) starts
// at 0.
const lastSeq = this._cursor[this._primaryServerKey]
const startSeq = typeof lastSeq === 'number' && lastSeq >= 0 ? lastSeq + 1 : 0
if (startSeq === 0) {
// eslint-disable-next-line no-console
console.log('[repo-mirror] cold-boot: replaying log from seq 0')
} else {
// eslint-disable-next-line no-console
console.log('[repo-mirror] warm-boot: resuming log from seq', startSeq)
}
// Re-fold the WHOLE log for tombstones before streaming anything.
//
// retired-repos.js calls the set "derived, not stored ... rebuilt
// identically on every boot", and on the ORIGIN that holds: pear-git.js
// folds its own writable log from seq 0. A MIRROR has no such log — the
// origin wrote the tombstones to ITS log — so the mirror's only source
// was the fold in _enqueue, which rides on a stream opened at
// `cursor + 1`. A tombstone below the cursor was therefore folded once,
// in memory, and forgotten on the next restart: the mirror resumed
// listing and announcing a repo the origin had retired, permanently.
// (Production, 2026-09-02: the origin held 15 tombstones, the mirror —
// the load balancer's ACTIVE backend — reported none.)
//
// The cursor is not the problem and does not move: it exists to avoid
// ~80 minutes of redundant `git fetch` per restart. This pass reads
// METADATA only (~100 B per entry, no git), so the expensive replay the
// cursor guards against is unaffected.
//
// Unconditional, not just on warm boot. On a cold boot the stream does
// fold from seq 0, but only as it goes — so a repo created at seq 3 and
// retired at seq 40 gets cloned first and tombstoned after, and the
// never-delete rule then leaves those bytes on disk for good.
//
// Ordering is safe: applyEntry keeps the highest seq per name, and this
// runs BEFORE the live stream, so a later un-retire still wins.
//
// Bounded, because this is a REPLICATED core and we await it. A block
// this mirror has not downloaded waits on the network with no deadline,
// so an unbounded fold would stall the loop completely whenever the
// origin is unreachable — strictly worse than the bug above. The timeout
// makes the stream throw instead; we log, keep whatever folded, and fall
// through to the live stream, which is exactly the old behaviour.
if (this._retired) {
try {
await foldRetired(this._log, this._retired, { timeout: FOLD_BLOCK_TIMEOUT_MS })
} catch (err) {
// eslint-disable-next-line no-console
console.error('[repo-mirror] tombstone re-fold incomplete:', err && err.message)
}
// Re-prune the sync map now that the set is populated.
//
// The fold above is the FIRST time a mirror learns its tombstones — its
// own RepoEventLog carries none, so the set was empty when the sync
// service ran its boot scan and pruned against it. Everything retired is
// therefore still sitting in sync.repos, and status() (so `pgh repo` and
// the SSR /repos page) lists it. Without this the mirror knows the
// tombstones and advertises the repos anyway.
//
// In-memory prune, no disk walk. Idempotent, and a no-op on an origin,
// where the set was already complete before the boot scan.
if (this._retired.size > 0 && this._sync && typeof this._sync.pruneRetired === 'function') {
try {
const pruned = this._sync.pruneRetired()
if (pruned > 0) {
// eslint-disable-next-line no-console
console.log('[repo-mirror] pruned ' + pruned + ' retired repo(s) from the listing after folding the origin log')
}
} catch (err) {
// eslint-disable-next-line no-console
console.error('[repo-mirror] post-fold prune failed:', err && err.message)
}
}
}
// Snapshot the log's live frontier — the index at the moment we
// subscribe. Entries with seq < liveFrontier are BACKLOG (pre-
// existing on the log when we subscribed); seq >= liveFrontier are
// LIVE (appended AFTER subscription). Backlog dups can be dropped
// because the in-flight pull's `git fetch` already gets the
// latest from origin's bare repo. Live dups must trigger a re-
// pull because origin may have pushed AFTER the in-flight pull's
// fetch started.
const liveFrontier = this._log.core ? this._log.core.length : startSeq
let seq = startSeq
for await (const entry of this._log.entries({ live: true, start: startSeq })) {
if (this._stopped) break
this._enqueue(entry, seq, seq >= liveFrontier)
seq++
}
}
_enqueue (entry, seq, isLive) {
if (!entry || !entry.name) return
// Tombstone fold. Previously `if (entry.op === 2) return` — the retire
// event was decoded and then discarded, so a repo deleted by hand on one
// peer was re-cloned from another. Recording it here (and un-retiring on
// a later op 0/1) is what makes a manual delete STICK.
//
// This records metadata only. Nothing here deletes bytes, and this
// daemon must never gain the ability to — see retired-repos.js.
if (this._retired) this._retired.applyEntry(entry, seq)
if (entry.op === OP_RETIRED) return
if (this._retired && this._retired.has(entry.name)) return
// Scope gate: an org-scoped puller admits ONLY repos in orgs this daemon
// joined; the default (null) admits everything (full mirror). Checked
// live so repos added to an org after boot start matching without a
// restart. Filtered-out entries are dropped BEFORE any git init/fetch.
if (this._repoFilter && !this._repoFilter(entry)) return
if (this._inflightNames.has(entry.name)) {
if (isLive) {
// Origin pushed AFTER our in-flight pull's fetch started.
// Schedule a re-pull when the in-flight completes.
const prev = this._needsRePull.get(entry.name)
if (!prev || seq > prev.seq) this._needsRePull.set(entry.name, { entry, seq })
}
// Backlog dup: drop. The in-flight pull is fetching from origin
// RIGHT NOW — it'll get whatever's currently at origin's bare
// repo, which is at least as new as the dropped entry's head.
return
}
// Coalesce: replace any older queued entry for this name. Latest
// seq wins. This collapses a flurry of pushes to the same repo
// into one pull (the latest fetch picks up everything since the
// previous fetch).
const prev = this._queue.get(entry.name)
if (!prev || seq > prev.seq) this._queue.set(entry.name, { entry, seq })
this._drain()
}
_drain () {
while (!this._stopped && this._inflight < this._max && this._queue.size > 0) {
// Pop the first entry in insertion order. With Map this gives
// FIFO behavior across distinct names.
const it = this._queue.entries().next()
if (it.done) break
const [name, { entry, seq }] = it.value
this._queue.delete(name)
this._process(entry, seq)
}
}
async _process (entry, seq) {
this._inflight++
this._inflightNames.add(entry.name)
let ok = false
let err = null
try {
// Bind the tracker key FIRST, independent of the git pull. Tracker
// (issues/refs/meta) replication rides the autobase + overlay and does
// NOT depend on the git pull succeeding — decoupling them lets a repo's
// tracker bind even when its pull is slow/failing, and lets a tracker-
// only entry (e.g. the repo-less `bootstrap` tracker, #66) bind without
// a git repo to clone. Idempotent (bindRepoBootstrapKey no-ops on an
// already-bound key).
if (entry.issueBootstrapKey && this._issueRegistry
&& typeof this._issueRegistry.bindRepoBootstrapKey === 'function') {
this._issueRegistry.bindRepoBootstrapKey(entry.name, entry.issueBootstrapKey)
}
const res = await this._sync.pull(this._remoteName, entry.name)
// hyper-sync.pull() does NOT throw on a failed fetch — it RETURNS
// { synced:[], failed:[{name,error}] }. So a failed pull must be turned
// into a throw here, or the cursor advances over it and the repo is
// SKIPPED FOREVER on warm-boot (`resuming from seq N`) — refs replicate via
// the autobase, but git OBJECTS never clone. This was the real cause of
// org joiners getting refs-but-no-objects. pear-git #3.
const f = res && res.failed && res.failed.find((x) => x.name === entry.name)
if (f) throw new Error('pull failed for ' + entry.name + ': ' + (f.error || 'unknown error'))
// Cursor advances ONLY on genuine success. A failed pull leaves
// the cursor where it was so the next restart re-tries.
const prev = this._cursor[this._primaryServerKey]
if (typeof prev !== 'number' || seq > prev) {
this._cursor[this._primaryServerKey] = seq
this._saveCursor()
}
// The pull wrote refs and objects INSIDE an existing <repo>.git, which
// the repos-dir watcher cannot see (it is a non-recursive watch on the
// directory). Without this the in-memory record status() serves keeps
// the boot-time branch count, pack size and lastUpdated for a repo that
// has just advanced — and lastUpdated is what /repos orders on, so a
// long-lived mirror ranked active repos as if they were months quiet.
// Per-repo, not sync.scan(): a full walk per pulled repo would stall the
// event loop through the whole drain.
if (typeof this._sync.refreshRepo === 'function') {
try { this._sync.refreshRepo(entry.name) } catch {}
}
ok = true
} catch (e) {
err = e
// eslint-disable-next-line no-console
console.error('[repo-mirror] pull failed for', entry.name, '-', e && e.message)
} finally {
this._inflight--
this._inflightNames.delete(entry.name)
// If origin pushed for this repo while we were fetching, the
// in-flight may not have caught the latest commits. Re-enqueue
// the queued newer entry now that we're free.
if (this._needsRePull.has(entry.name)) {
const next = this._needsRePull.get(entry.name)
this._needsRePull.delete(entry.name)
const queuedPrev = this._queue.get(entry.name)
if (!queuedPrev || next.seq > queuedPrev.seq) {
this._queue.set(entry.name, next)
}
}
try { if (this._onProgress) this._onProgress({ name: entry.name, ok, err }) } catch {}
if (ok) { try { if (this._onMount) this._onMount(entry.name) } catch {} }
this._drain()
}
}
/**
* Snapshot for tests and /admin views.
*/
status () {
return {
inflight: this._inflight,
queued: this._queue.size,
pendingRePull: this._needsRePull.size,
primary: this._primaryServerKey,
maxConcurrent: this._max,
cursor: this._cursor[this._primaryServerKey] !== undefined
? this._cursor[this._primaryServerKey]
: null
}
}
async close () {
this._stopped = true
try { await this._log.close() } catch {}
// Drain any in-flight clones so tests and graceful shutdown don't
// race with pending git fetches.
const deadline = Date.now() + 5000
while (this._inflight > 0 && Date.now() < deadline) {
await new Promise((r) => setTimeout(r, 50))
}
}
}