| … | … | @@ -140,12 +140,28 @@ export class RefReconcilerManager { |
| 140 | 140 | let repos = [] |
| 141 | 141 | try { repos = this._issueRegistry.listRepos() || [] } catch { repos = [] } |
| 142 | 142 | let scanned = 0 |
| 143 | let skipped = 0 | |
| 143 | 144 | for (const repoName of repos) { |
| 144 | 145 | try { |
| 145 | 146 | const existing = this._recs.get(repoName) |
| 146 | if (existing) await existing.reconcile() | |
| 147 | else await this.ensureFor(repoName) // starts + reconciles once | |
| 148 | scanned++ | |
| 147 | if (existing) { | |
| 148 | // Ask before working. A materialized, quiet repo gains nothing from | |
| 149 | // another listRefs() — the tracker's own `update` event already drives | |
| 150 | // its reconcile — and doing it anyway across every repo is what made | |
| 151 | // the sweep degrade the daemon for ~57s of every 300s (146 repos: | |
| 152 | // 11ms -> ~450ms TTFB, git subprocess count 0 throughout, so the cost | |
| 153 | // is per-tracker autobase work). A reconciler that predates needsPass() | |
| 154 | // is never skipped. | |
| 155 | if (typeof existing.needsPass === 'function' && !existing.needsPass()) { | |
| 156 | skipped++ | |
| 157 | } else { | |
| 158 | await existing.reconcile() | |
| 159 | scanned++ | |
| 160 | } | |
| 161 | } else { | |
| 162 | await this.ensureFor(repoName) // starts + reconciles once | |
| 163 | scanned++ | |
| 164 | } | |
| 149 | 165 | } catch (err) { |
| 150 | 166 | this._log.warn('reconcileAll ' + repoName + ': ' + (err && err.message)) |
| 151 | 167 | } |
| … | … | @@ -158,7 +174,7 @@ export class RefReconcilerManager { |
| 158 | 174 | // and handing the loop back between repos keeps the daemon responsive. |
| 159 | 175 | await new Promise((resolve) => setImmediate(resolve)) |
| 160 | 176 | } |
| 161 | return { scanned } | |
| 177 | return { scanned, skipped } | |
| 162 | 178 | } |
| 163 | 179 | |
| 164 | 180 | async closeAll () { |
| … | … | @@ -55,6 +55,20 @@ export class RefReconciler { |
| 55 | 55 | this._running = null // in-flight reconcile (serialize concurrent triggers) |
| 56 | 56 | this._queued = false // a trigger arrived while a reconcile was running |
| 57 | 57 | |
| 58 | // ── Sweep-skip bookkeeping (see needsPass) ─────────────────────────── | |
| 59 | // A ref we could not materialize because its objects were missing. Until | |
| 60 | // that clears, the repo is genuinely out of sync and the sweep MUST retry | |
| 61 | // it however quiet the tracker looks. | |
| 62 | this._deferred = false | |
| 63 | this._lastPassAt = 0 | |
| 64 | this._lastVersion = null // view.version at the end of the last pass | |
| 65 | // Backstop: re-pass even when nothing looks changed, so silent drift (a | |
| 66 | // missed `update`, an out-of-band edit to the bare repo) still heals. | |
| 67 | // Jittered per instance and re-rolled after every pass, so the repos do | |
| 68 | // not all fall due on the same sweep and re-create the burst this avoids. | |
| 69 | this._backstopBaseMs = typeof opts.backstopMs === 'number' ? opts.backstopMs : 30 * 60 * 1000 | |
| 70 | this._backstopMs = this._rollBackstop() | |
| 71 | ||
| 58 | 72 | // Debounced update trigger. Two distinct reasons, both measured in prod: |
| 59 | 73 | // |
| 60 | 74 | // 1. COALESCING. `update` fires on every view advance. On a MIRROR every |
| … | … | @@ -124,8 +138,56 @@ export class RefReconciler { |
| 124 | 138 | return this._running |
| 125 | 139 | } |
| 126 | 140 | |
| 141 | _rollBackstop () { | |
| 142 | // 0.75x–1.25x of the base, so N reconcilers spread instead of synchronising. | |
| 143 | return Math.round(this._backstopBaseMs * (0.75 + Math.random() * 0.5)) | |
| 144 | } | |
| 145 | ||
| 146 | _viewVersion () { | |
| 147 | try { | |
| 148 | const view = this._tracker && this._tracker.base && this._tracker.base.view | |
| 149 | return view && typeof view.version === 'number' ? view.version : null | |
| 150 | } catch { | |
| 151 | return null | |
| 152 | } | |
| 153 | } | |
| 154 | ||
| 155 | /** | |
| 156 | * Does the periodic sweep need to run a pass on this repo? | |
| 157 | * | |
| 158 | * The sweep is a BACKSTOP — per-repo reconcile is already event-driven off the | |
| 159 | * tracker's `update`, so a repo that is materialized and quiet gains nothing | |
| 160 | * from another `listRefs()` (an `await base.update()` plus a full ref | |
| 161 | * range-scan). Across 146 repos that redundant work degraded the daemon from | |
| 162 | * 11ms to ~450ms TTFB for ~57s of every 300s. | |
| 163 | * | |
| 164 | * Answer true whenever we cannot PROVE there is nothing to do: | |
| 165 | * - never completed a pass, or | |
| 166 | * - a ref is still waiting on objects, or | |
| 167 | * - the linearized view advanced since the last pass, or | |
| 168 | * - the view version is unreadable (never skip on a broken signal), or | |
| 169 | * - the backstop deadline expired. | |
| 170 | * | |
| 171 | * @returns {boolean} | |
| 172 | */ | |
| 173 | needsPass () { | |
| 174 | if (this._deferred) return true | |
| 175 | if (!this._lastPassAt) return true | |
| 176 | if (Date.now() - this._lastPassAt >= this._backstopMs) return true | |
| 177 | const v = this._viewVersion() | |
| 178 | if (v === null || v !== this._lastVersion) return true | |
| 179 | return false | |
| 180 | } | |
| 181 | ||
| 127 | 182 | async _reconcileOnce () { |
| 183 | // Any ref left unmaterialized this pass re-arms the sweep for this repo. | |
| 184 | let deferred = false | |
| 128 | 185 | const refs = await this._tracker.listRefs() // [{ ref, hash, updatedBy, timestamp }] |
| 186 | // The version whose refs we are about to materialize. Sampled HERE — not at | |
| 187 | // the end of the pass — because listRefs() does the `base.update()` that | |
| 188 | // advances the view, and any advance AFTER this point is work this pass did | |
| 189 | // not do. Recording a later version would skip it forever. | |
| 190 | const versionReconciled = this._viewVersion() | |
| 129 | 191 | const desired = new Map() |
| 130 | 192 | for (const r of refs) if (r.hash) desired.set(r.ref, r.hash) |
| 131 | 193 | |
| … | … | @@ -144,11 +206,13 @@ export class RefReconciler { |
| 144 | 206 | await this._ensureObjects([oid]) |
| 145 | 207 | } catch (err) { |
| 146 | 208 | this._log.warn(`ensureObjects failed for ${ref}@${oid}: ${err.message}`) |
| 209 | deferred = true | |
| 147 | 210 | continue |
| 148 | 211 | } |
| 149 | 212 | } |
| 150 | 213 | if (!await this._hasObject(oid)) { |
| 151 | 214 | this._log.warn(`skip ${ref}: object ${oid} not present (no source)`) |
| 215 | deferred = true | |
| 152 | 216 | continue |
| 153 | 217 | } |
| 154 | 218 | } |
| … | … | @@ -192,6 +256,12 @@ export class RefReconciler { |
| 192 | 256 | this._log.warn('onChange hook failed: ' + (err && err.message)) |
| 193 | 257 | } |
| 194 | 258 | } |
| 259 | ||
| 260 | // Record what the sweep needs to decide whether to come back (needsPass). | |
| 261 | this._deferred = deferred | |
| 262 | this._lastPassAt = Date.now() | |
| 263 | this._lastVersion = versionReconciled | |
| 264 | this._backstopMs = this._rollBackstop() | |
| 195 | 265 | } |
| 196 | 266 | |
| 197 | 267 | /** |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "pear-git-v3", |
| 3 | "version": "0.1.219", | |
| 3 | "version": "0.1.220", | |
| 4 | 4 | "description": "P2P git forge — multi-writer issue/org/release/peer-mirror daemon. Built bottom-up after the failed v2.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "index.js", |
| … | … | @@ -0,0 +1,130 @@ |
| 1 | // The periodic sweep must be able to SKIP a repo that provably has nothing to | |
| 2 | // do — and must never skip one that does. | |
| 3 | // | |
| 4 | // Measured on the public backend (146 repos, 0.1.219): reconcileAll degraded the | |
| 5 | // daemon from 11ms to ~450ms TTFB for ~57s of every 300s — a ~19% duty cycle, so | |
| 6 | // about one request in five. `git` subprocess count was 0 for the whole window, | |
| 7 | // so the cost is the per-tracker `listRefs()` (an `await base.update()` plus a | |
| 8 | // full ref range-scan), not git plumbing. Yielding between repos (0.1.180) | |
| 9 | // bounded a single stall but not the total, because the work itself was unchanged. | |
| 10 | // | |
| 11 | // The sweep is a BACKSTOP — per-repo reconcile is already event-driven off the | |
| 12 | // tracker's `update` — so a materialized, quiet repo gains nothing from another | |
| 13 | // pass. needsPass() decides, and it must fail SAFE: answer true whenever we | |
| 14 | // cannot prove there is nothing to do. | |
| 15 | ||
| 16 | import test from 'brittle' | |
| 17 | import { mkdtempSync, writeFileSync } from '#node:fs' | |
| 18 | import { join } from '#node:path' | |
| 19 | import { tmpdir } from '#node:os' | |
| 20 | ||
| 21 | import { RefReconciler } from '../lib/services/ref-reconciler.js' | |
| 22 | import { execFileSync } from '../lib/_compat-exec.js' | |
| 23 | ||
| 24 | function git (args, opts = {}) { | |
| 25 | return execFileSync('git', args, { encoding: 'utf8', ...opts }).toString().trim() | |
| 26 | } | |
| 27 | ||
| 28 | // Tracker whose linearized view version we can advance, the way replication does. | |
| 29 | function fakeTracker (refs, { version = 1, withVersion = true } = {}) { | |
| 30 | const t = { | |
| 31 | base: { | |
| 32 | on () {}, | |
| 33 | off () {}, | |
| 34 | view: withVersion ? { version } : {} | |
| 35 | }, | |
| 36 | async listRefs () { return t._refs.slice() }, | |
| 37 | _refs: refs.slice(), | |
| 38 | setVersion (v) { if (t.base.view) t.base.view.version = v } | |
| 39 | } | |
| 40 | return t | |
| 41 | } | |
| 42 | ||
| 43 | async function makeRepo () { | |
| 44 | const dir = mkdtempSync(join(tmpdir(), 'pgv3-skip-')) | |
| 45 | const bare = join(dir, 'repo.git') | |
| 46 | const work = join(dir, 'work') | |
| 47 | git(['init', '--bare', '-q', bare]) | |
| 48 | git(['init', '-q', work]) | |
| 49 | git(['-C', work, 'config', 'user.email', 't@t']) | |
| 50 | git(['-C', work, 'config', 'user.name', 'T']) | |
| 51 | git(['-C', work, 'config', 'commit.gpgsign', 'false']) | |
| 52 | writeFileSync(join(work, 'a.txt'), 'one\n') | |
| 53 | git(['-C', work, 'add', '.']) | |
| 54 | git(['-C', work, 'commit', '-q', '-m', 'c1']) | |
| 55 | const oid = git(['-C', work, 'rev-parse', 'HEAD']) | |
| 56 | git(['-C', work, 'push', '-q', bare, 'HEAD:refs/heads/main']) | |
| 57 | return { bare, oid } | |
| 58 | } | |
| 59 | ||
| 60 | test('a reconciler that has never passed always needs one', async (t) => { | |
| 61 | const { bare, oid } = await makeRepo() | |
| 62 | const rec = new RefReconciler({ tracker: fakeTracker([{ ref: 'refs/heads/main', hash: oid }]), repoPath: bare }) | |
| 63 | t.ok(rec.needsPass(), 'no pass has completed yet, so nothing is proven') | |
| 64 | }) | |
| 65 | ||
| 66 | test('a materialized, quiet repo does NOT need another pass', async (t) => { | |
| 67 | const { bare, oid } = await makeRepo() | |
| 68 | const rec = new RefReconciler({ tracker: fakeTracker([{ ref: 'refs/heads/main', hash: oid }]), repoPath: bare }) | |
| 69 | ||
| 70 | await rec.reconcile() | |
| 71 | ||
| 72 | t.absent(rec.needsPass(), 'view unchanged + nothing deferred => the sweep can skip it') | |
| 73 | }) | |
| 74 | ||
| 75 | test('an advancing view re-arms the sweep', async (t) => { | |
| 76 | const { bare, oid } = await makeRepo() | |
| 77 | const tracker = fakeTracker([{ ref: 'refs/heads/main', hash: oid }]) | |
| 78 | const rec = new RefReconciler({ tracker, repoPath: bare }) | |
| 79 | ||
| 80 | await rec.reconcile() | |
| 81 | t.absent(rec.needsPass(), 'quiet after the pass') | |
| 82 | ||
| 83 | tracker.setVersion(2) // a peer's write replicated in | |
| 84 | t.ok(rec.needsPass(), 'the view advanced, so there is work to do') | |
| 85 | }) | |
| 86 | ||
| 87 | // The safety case. A ref whose objects never arrived is exactly the state the | |
| 88 | // sweep exists to heal, so a quiet view must NOT be enough to skip it. | |
| 89 | test('a ref left unmaterialized keeps the repo in the sweep', async (t) => { | |
| 90 | const { bare } = await makeRepo() | |
| 91 | const missing = 'd'.repeat(40) // a well-formed oid that is not in the repo | |
| 92 | const rec = new RefReconciler({ | |
| 93 | tracker: fakeTracker([{ ref: 'refs/heads/orphan', hash: missing }]), | |
| 94 | repoPath: bare | |
| 95 | }) | |
| 96 | ||
| 97 | await rec.reconcile() | |
| 98 | ||
| 99 | // rev-parse exits non-zero for an absent ref, which the helper turns into a throw. | |
| 100 | let materialized = true | |
| 101 | try { git(['--git-dir', bare, 'rev-parse', '--verify', 'refs/heads/orphan']) } catch { materialized = false } | |
| 102 | ||
| 103 | t.absent(materialized, 'the ref was not materialized (its object is absent)') | |
| 104 | t.ok(rec.needsPass(), 'deferred work outranks a quiet view — the sweep must retry') | |
| 105 | }) | |
| 106 | ||
| 107 | test('an unreadable view version never allows a skip', async (t) => { | |
| 108 | const { bare, oid } = await makeRepo() | |
| 109 | const rec = new RefReconciler({ | |
| 110 | tracker: fakeTracker([{ ref: 'refs/heads/main', hash: oid }], { withVersion: false }), | |
| 111 | repoPath: bare | |
| 112 | }) | |
| 113 | ||
| 114 | await rec.reconcile() | |
| 115 | ||
| 116 | t.ok(rec.needsPass(), 'no usable change signal => fail safe and do the work') | |
| 117 | }) | |
| 118 | ||
| 119 | test('the backstop deadline re-arms an otherwise quiet repo', async (t) => { | |
| 120 | const { bare, oid } = await makeRepo() | |
| 121 | const rec = new RefReconciler({ | |
| 122 | tracker: fakeTracker([{ ref: 'refs/heads/main', hash: oid }]), | |
| 123 | repoPath: bare, | |
| 124 | backstopMs: 0 // every pass is immediately overdue | |
| 125 | }) | |
| 126 | ||
| 127 | await rec.reconcile() | |
| 128 | ||
| 129 | t.ok(rec.needsPass(), 'silent drift still heals because the backstop expires') | |
| 130 | }) |
| … | … | @@ -52,6 +52,48 @@ test('reconcileAll yields to the event loop between repos', async (t) => { |
| 52 | 52 | t.ok(ticks > 0, `event loop got control during the sweep (${ticks} ticks; 0 == starved)`) |
| 53 | 53 | }) |
| 54 | 54 | |
| 55 | /** | |
| 56 | * The sweep is a BACKSTOP: per-repo reconcile is already event-driven off each | |
| 57 | * tracker's `update`. Re-running `listRefs()` — an `await base.update()` plus a | |
| 58 | * full ref range-scan — across every repo every 5 minutes is therefore redundant | |
| 59 | * for any repo that is quiet and materialized. | |
| 60 | * | |
| 61 | * Measured on the public backend (146 repos, 0.1.219): the sweep degraded TTFB | |
| 62 | * from 11ms to ~450ms for ~57s of every 300s — a ~19% duty cycle, so roughly one | |
| 63 | * request in five. `git` subprocess count was 0 throughout, so the cost is the | |
| 64 | * per-tracker autobase work, not git plumbing. Yielding between repos (0.1.180) | |
| 65 | * bounded a single stall but not the total, because the work itself is unchanged. | |
| 66 | * | |
| 67 | * So the sweep must ask each reconciler whether it actually has anything to do. | |
| 68 | */ | |
| 69 | test('reconcileAll skips reconcilers that report no pending work', async (t) => { | |
| 70 | const calls = [] | |
| 71 | const mgr = new RefReconcilerManager({ | |
| 72 | issueRegistry: { listRepos: () => ['quiet', 'deferred', 'due', 'legacy'] }, | |
| 73 | overlay: {}, | |
| 74 | reposDir: '/nonexistent' | |
| 75 | }) | |
| 76 | ||
| 77 | const rec = (name, needsPass) => ({ | |
| 78 | reconcile: async () => { calls.push(name) }, | |
| 79 | ...(needsPass === undefined ? {} : { needsPass: () => needsPass }) | |
| 80 | }) | |
| 81 | ||
| 82 | mgr._recs.set('quiet', rec('quiet', false)) // materialized + nothing new | |
| 83 | mgr._recs.set('deferred', rec('deferred', true)) // a ref waiting on objects | |
| 84 | mgr._recs.set('due', rec('due', true)) // past its backstop deadline | |
| 85 | mgr._recs.set('legacy', rec('legacy')) // no needsPass -> must not be skipped | |
| 86 | ||
| 87 | const res = await mgr.reconcileAll() | |
| 88 | ||
| 89 | t.absent(calls.includes('quiet'), 'the quiet repo was NOT reconciled') | |
| 90 | t.ok(calls.includes('deferred'), 'a repo with deferred objects still reconciles') | |
| 91 | t.ok(calls.includes('due'), 'a repo past its backstop still reconciles') | |
| 92 | t.ok(calls.includes('legacy'), 'a reconciler without needsPass() is never skipped') | |
| 93 | t.is(res.skipped, 1, 'reports how many were skipped') | |
| 94 | t.is(res.scanned, 3, 'skipped repos are not counted as scanned') | |
| 95 | }) | |
| 96 | ||
| 55 | 97 | test('reconcileAll still sweeps every repo when one throws', async (t) => { |
| 56 | 98 | const names = ['a', 'b', 'c'] |
| 57 | 99 | const seen = [] |