| … | … | @@ -1588,11 +1588,20 @@ export async function startPearGit (opts = {}) { |
| 1588 | 1588 | if (svc._releases && svc.modules.overlayV2 && !svc._releasePull) { |
| 1589 | 1589 | try { |
| 1590 | 1590 | const { ReleasePuller, startReleasePull } = await import('./services/release-puller.js') |
| 1591 | const { DeletedReleases } = await import('./services/deleted-releases.js') | |
| 1592 | // Replicated tombstone set. Created EMPTY here because the origin's | |
| 1593 | // release log is not opened until followReleaseLog() below; the same | |
| 1594 | // object is folded there and mutated live, so the puller sees tombstones | |
| 1595 | // as they arrive. Without it the puller re-requests versions the origin | |
| 1596 | // has deleted and retries forever -- see services/deleted-releases.js. | |
| 1597 | const deletedReleases = new DeletedReleases() | |
| 1598 | svc._deletedReleases = deletedReleases | |
| 1591 | 1599 | const puller = new ReleasePuller({ |
| 1592 | 1600 | releaseDir: svc._releases._releaseDir, |
| 1593 | 1601 | overlay: svc.modules.overlayV2, |
| 1594 | 1602 | deploy: svc._deploy || null, |
| 1595 | 1603 | releases: svc._releases, |
| 1604 | deleted: deletedReleases, | |
| 1596 | 1605 | log: (l) => console.log(l) |
| 1597 | 1606 | }) |
| 1598 | 1607 | svc._releasePull = startReleasePull({ |
| … | … | @@ -1658,6 +1667,23 @@ export async function startPearGit (opts = {}) { |
| 1658 | 1667 | // arrived pre-subscription (the same priming RepoPuller._runLoop does). |
| 1659 | 1668 | try { await originReleaseLog.core.update({ wait: true }) } catch {} |
| 1660 | 1669 | |
| 1670 | // Fold the whole log from seq 0 -- NOT from the saved cursor. The cursor | |
| 1671 | // only tracks what has been NOTIFIED; a delete that landed before this | |
| 1672 | // peer's cursor would otherwise be invisible and the version would be | |
| 1673 | // chased forever. The fold is a pure replay, so re-reading is free. | |
| 1674 | if (svc._deletedReleases) { | |
| 1675 | try { | |
| 1676 | const { foldDeleted } = await import('./services/deleted-releases.js') | |
| 1677 | await foldDeleted(originReleaseLog, svc._deletedReleases) | |
| 1678 | if (svc._deletedReleases.size) { | |
| 1679 | console.log('[release-pull] ' + svc._deletedReleases.size + | |
| 1680 | ' tombstoned version(s) will not be pulled') | |
| 1681 | } | |
| 1682 | } catch (err) { | |
| 1683 | console.error('[release-pull] tombstone fold failed:', err && err.message) | |
| 1684 | } | |
| 1685 | } | |
| 1686 | ||
| 1661 | 1687 | const cursorPath = join(opts.storage, 'release-event-cursor.json') |
| 1662 | 1688 | let cursor = {} |
| 1663 | 1689 | try { cursor = JSON.parse(fs.readFileSync(cursorPath, 'utf8')) || {} } catch {} |
| … | … | @@ -1681,6 +1707,9 @@ export async function startPearGit (opts = {}) { |
| 1681 | 1707 | for await (const entry of originReleaseLog.entries({ live: true, start: startSeq })) { |
| 1682 | 1708 | if (releaseFollowStopped) break |
| 1683 | 1709 | try { |
| 1710 | // Apply BEFORE notify: a delete arriving in the same batch must | |
| 1711 | // already be tombstoned when the reconcile it triggers plans its pull. | |
| 1712 | if (svc._deletedReleases) svc._deletedReleases.applyEntry(entry, seq) | |
| 1684 | 1713 | svc._releasePull.notify(entry) |
| 1685 | 1714 | saveCursor(seq) |
| 1686 | 1715 | } catch (err) { |
| … | … | @@ -0,0 +1,138 @@ |
| 1 | /** | |
| 2 | * DeletedReleases — the replicated "stop pulling this version" set. | |
| 3 | * | |
| 4 | * ───────────────────────────────────────────────────────────────────────── | |
| 5 | * THIS MODULE NEVER DELETES ANYTHING, AND MUST NEVER LEARN HOW. | |
| 6 | * | |
| 7 | * A tombstone is metadata: it changes what a peer PULLS, never what is on | |
| 8 | * disk. Removing release bytes is a manual operator action on each peer. | |
| 9 | * Same rule, same reasoning as services/retired-repos.js — read that first. | |
| 10 | * ───────────────────────────────────────────────────────────────────────── | |
| 11 | * | |
| 12 | * WHY THIS EXISTS (prod incident, 2026-08-29): `pgh deploy gc` removed 55 | |
| 13 | * favela-android versions on tabertron. The bytes were the only copy, so | |
| 14 | * stortron's ReleasePuller could never satisfy its fetch and logged, forever: | |
| 15 | * | |
| 16 | * [release-pull] error: no holder for | |
| 17 | * favela-android/versions/master-8d988ca/favela-android.apk | |
| 18 | * → fetched 0, replaced 0, skipped 213, errors 1 — PARTIAL, will retry | |
| 19 | * | |
| 20 | * The active public backend crashed (exit 1) and both daemons flapped. | |
| 21 | * | |
| 22 | * The event already existed — ReleaseManager.deleteVersion appends | |
| 23 | * `op: OP_DELETED` to the ReleaseEventLog — but NOTHING consumed it. | |
| 24 | * repo-mirror.js honours the equivalent repo tombstone (OP_RETIRED) via | |
| 25 | * RetiredRepos; the release path had the emitter and no reader. This is the | |
| 26 | * missing reader. | |
| 27 | * | |
| 28 | * DERIVED, NOT STORED. The ReleaseEventLog is append-only and replayed from | |
| 29 | * seq 0 with no persisted cursor, so this set is a pure fold over it — | |
| 30 | * nothing to migrate, rebuilt identically on every boot. Later ops win by | |
| 31 | * sequence number, so a republish (OP_PUBLISHED) after a delete revives the | |
| 32 | * version and a mistake is reversible. | |
| 33 | * | |
| 34 | * ⚠ ROLLOUT GATE: a peer running older code IGNORES the tombstone and keeps | |
| 35 | * requesting the version. Every peer must run this code BEFORE any operator | |
| 36 | * removes release bytes — exactly the ordering that was violated to cause the | |
| 37 | * incident above. | |
| 38 | */ | |
| 39 | ||
| 40 | /** ReleaseEventLog op codes — mirrors services/release-event-log.js. */ | |
| 41 | export const OP_PUBLISHED = 0 | |
| 42 | export const OP_PROMOTED = 1 | |
| 43 | export const OP_UNPINNED = 2 | |
| 44 | export const OP_DELETED = 3 | |
| 45 | ||
| 46 | /** Composite key. A version is only meaningful within its repo. */ | |
| 47 | function key (repo, version) { | |
| 48 | return repo + '/' + version | |
| 49 | } | |
| 50 | ||
| 51 | export class DeletedReleases { | |
| 52 | constructor () { | |
| 53 | // "repo\0version" -> { deleted: boolean, seq: number } | |
| 54 | this._state = new Map() | |
| 55 | } | |
| 56 | ||
| 57 | /** | |
| 58 | * Fold one ReleaseEventLog entry. Later sequence numbers win, so replaying | |
| 59 | * the log from 0 converges regardless of arrival order. | |
| 60 | * | |
| 61 | * Channel-only ops (OP_PROMOTED / OP_UNPINNED) carry an empty version and | |
| 62 | * are ignored — they say nothing about whether bytes exist. | |
| 63 | * | |
| 64 | * @param {object} entry - decoded ReleaseEventLog entry | |
| 65 | * @param {number} [seq] - the entry's log sequence number | |
| 66 | * @returns {boolean} whether the version is deleted AFTER applying this entry | |
| 67 | */ | |
| 68 | applyEntry (entry, seq) { | |
| 69 | if (!entry || !entry.repo || !entry.version) return false | |
| 70 | if (entry.op !== OP_DELETED && entry.op !== OP_PUBLISHED) { | |
| 71 | return this.has(entry.repo, entry.version) | |
| 72 | } | |
| 73 | ||
| 74 | const k = key(entry.repo, entry.version) | |
| 75 | const prev = this._state.get(k) | |
| 76 | const at = typeof seq === 'number' ? seq : (prev ? prev.seq : 0) | |
| 77 | ||
| 78 | // An out-of-order replay of an OLDER entry must not clobber newer state. | |
| 79 | if (prev && typeof seq === 'number' && seq < prev.seq) return prev.deleted | |
| 80 | ||
| 81 | const deleted = entry.op === OP_DELETED | |
| 82 | this._state.set(k, { deleted, seq: at }) | |
| 83 | return deleted | |
| 84 | } | |
| 85 | ||
| 86 | /** @returns {boolean} true if this version is currently tombstoned */ | |
| 87 | has (repo, version) { | |
| 88 | const e = this._state.get(key(repo, version)) | |
| 89 | return !!(e && e.deleted) | |
| 90 | } | |
| 91 | ||
| 92 | /** @returns {Array<{repo,version}>} currently-tombstoned versions, sorted */ | |
| 93 | list () { | |
| 94 | const out = [] | |
| 95 | for (const [k, e] of this._state) { | |
| 96 | if (!e.deleted) continue | |
| 97 | const i = k.indexOf('/') | |
| 98 | out.push({ repo: k.slice(0, i), version: k.slice(i + 1) }) | |
| 99 | } | |
| 100 | return out.sort((a, b) => (a.repo + a.version).localeCompare(b.repo + b.version)) | |
| 101 | } | |
| 102 | ||
| 103 | get size () { | |
| 104 | return this.list().length | |
| 105 | } | |
| 106 | ||
| 107 | /** | |
| 108 | * A predicate suitable for handing to planPull(). Bound so it can be passed | |
| 109 | * as a bare function. | |
| 110 | */ | |
| 111 | get isDeleted () { | |
| 112 | return (repo, version) => this.has(repo, version) | |
| 113 | } | |
| 114 | ||
| 115 | /** Drop all state. Used by tests and by a full re-fold. */ | |
| 116 | clear () { | |
| 117 | this._state.clear() | |
| 118 | } | |
| 119 | } | |
| 120 | ||
| 121 | /** | |
| 122 | * Fold an entire ReleaseEventLog into a DeletedReleases set. | |
| 123 | * | |
| 124 | * Reads to the current end of the log and returns. | |
| 125 | * | |
| 126 | * @param {object} log - a ReleaseEventLog (needs .entries()) | |
| 127 | * @param {DeletedReleases} [set] | |
| 128 | * @returns {Promise<DeletedReleases>} | |
| 129 | */ | |
| 130 | export async function foldDeleted (log, set = new DeletedReleases()) { | |
| 131 | if (!log || typeof log.entries !== 'function') return set | |
| 132 | let seq = 0 | |
| 133 | for await (const entry of log.entries({ live: false, start: 0 })) { | |
| 134 | set.applyEntry(entry, seq) | |
| 135 | seq++ | |
| 136 | } | |
| 137 | return set | |
| 138 | } |
| … | … | @@ -78,10 +78,28 @@ const MAX_RETRY_MS = 5 * 60 * 1000 |
| 78 | 78 | * @param {object} remoteIndex the publisher's parsed _index.json |
| 79 | 79 | * @param {(relPath: string) => ({size:number, hash:string}|null)} localStat |
| 80 | 80 | */ |
| 81 | export function planPull (remoteIndex, localStat) { | |
| 82 | const plan = { fetch: [], skip: [], replace: [], suspect: [] } | |
| 81 | export function planPull (remoteIndex, localStat, isDeleted) { | |
| 82 | const plan = { fetch: [], skip: [], replace: [], suspect: [], tombstoned: [] } | |
| 83 | 83 | for (const f of (remoteIndex && remoteIndex.files) || []) { |
| 84 | 84 | if (!f || !f.path || !f.hash) continue |
| 85 | // Tombstone gate. A version the origin retired is never fetched again. | |
| 86 | // Without this the puller chases bytes whose only holder is gone and | |
| 87 | // retries forever -- observed in prod 2026-08-29 after `pgh deploy gc` | |
| 88 | // removed versions on one peer: | |
| 89 | // error: no holder for favela-android/versions/master-8d988ca/... | |
| 90 | // fetched 0, replaced 0, skipped 213, errors 1 -- PARTIAL, will retry | |
| 91 | // repo-mirror.js already honours the equivalent repo tombstone | |
| 92 | // (OP_RETIRED); the release path emitted OP_DELETED and nothing consumed it. | |
| 93 | // | |
| 94 | // A tombstone changes what we PULL, never what is on disk: a local copy is | |
| 95 | // NOT scheduled for replacement or removal. Byte removal stays a manual | |
| 96 | // operator step on each peer -- see services/retired-repos.js. | |
| 97 | // | |
| 98 | // Checked before localStat so a dead version costs no I/O. | |
| 99 | if (isDeleted && f.repo && f.version && isDeleted(f.repo, f.version)) { | |
| 100 | plan.tombstoned.push(f) | |
| 101 | continue | |
| 102 | } | |
| 85 | 103 | const local = localStat(f.path) |
| 86 | 104 | if (!local) { plan.fetch.push(f); continue } |
| 87 | 105 | if (local.size === f.size && local.hash === f.hash) { plan.skip.push(f); continue } |
| … | … | @@ -112,7 +130,11 @@ export class ReleasePuller { |
| 112 | 130 | this._releases = opts.releases || null |
| 113 | 131 | this._log = opts.log || (() => {}) |
| 114 | 132 | this._hashFile = opts.hashFile || null |
| 115 | this.stats = { pulls: 0, fetched: 0, skipped: 0, replaced: 0, suspect: 0, bytes: 0, errors: [] } | |
| 133 | // Replicated tombstone set (services/deleted-releases.js). Optional: when | |
| 134 | // absent the puller behaves exactly as before and will chase versions the | |
| 135 | // origin has retired -- which is the 2026-08-29 incident. Wire it. | |
| 136 | this._deleted = opts.deleted || null | |
| 137 | this.stats = { pulls: 0, fetched: 0, skipped: 0, replaced: 0, suspect: 0, tombstoned: 0, bytes: 0, errors: [] } | |
| 116 | 138 | } |
| 117 | 139 | |
| 118 | 140 | async _hash (abs) { |
| … | … | @@ -214,8 +236,19 @@ export class ReleasePuller { |
| 214 | 236 | }) |
| 215 | 237 | |
| 216 | 238 | const localStat = await this._buildLocalStat(remoteIndex) |
| 217 | const plan = planPull(remoteIndex, localStat) | |
| 239 | const plan = planPull(remoteIndex, localStat, this._deleted ? this._deleted.isDeleted : null) | |
| 218 | 240 | this.stats.skipped += plan.skip.length |
| 241 | this.stats.tombstoned += plan.tombstoned.length | |
| 242 | ||
| 243 | // Surfaced, not silent: an operator needs to see that a version is being | |
| 244 | // skipped BY POLICY rather than failing to transfer. The old failure mode | |
| 245 | // was an unexplained `errors 1 -- PARTIAL, will retry` forever. | |
| 246 | if (plan.tombstoned.length) { | |
| 247 | const vs = [...new Set(plan.tombstoned.map(f => f.repo + '/' + f.version))] | |
| 248 | this._log('[release-pull] skipping ' + plan.tombstoned.length + ' file(s) in ' + | |
| 249 | vs.length + ' tombstoned version(s): ' + vs.slice(0, 5).join(', ') + | |
| 250 | (vs.length > 5 ? ', +' + (vs.length - 5) + ' more' : '')) | |
| 251 | } | |
| 219 | 252 | |
| 220 | 253 | for (const s of plan.suspect) { |
| 221 | 254 | this.stats.suspect++ |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "pear-git-v3", |
| 3 | "version": "0.1.239", | |
| 3 | "version": "0.1.240", | |
| 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,72 @@ |
| 1 | /** | |
| 2 | * DeletedReleases — the fold that stops a puller chasing retired versions. | |
| 3 | * See lib/services/deleted-releases.js for the 2026-08-29 incident this | |
| 4 | * exists to prevent. | |
| 5 | */ | |
| 6 | import test from 'brittle' | |
| 7 | import { | |
| 8 | DeletedReleases, foldDeleted, | |
| 9 | OP_PUBLISHED, OP_PROMOTED, OP_DELETED | |
| 10 | } from '../lib/services/deleted-releases.js' | |
| 11 | ||
| 12 | test('a deleted version is tombstoned; an untouched one is not', (t) => { | |
| 13 | const s = new DeletedReleases() | |
| 14 | s.applyEntry({ op: OP_PUBLISHED, repo: 'r', version: 'v1' }, 0) | |
| 15 | s.applyEntry({ op: OP_PUBLISHED, repo: 'r', version: 'v2' }, 1) | |
| 16 | s.applyEntry({ op: OP_DELETED, repo: 'r', version: 'v1' }, 2) | |
| 17 | ||
| 18 | t.ok(s.has('r', 'v1'), 'deleted version is tombstoned') | |
| 19 | t.absent(s.has('r', 'v2'), 'live version is not') | |
| 20 | t.is(s.size, 1) | |
| 21 | t.alike(s.list(), [{ repo: 'r', version: 'v1' }]) | |
| 22 | }) | |
| 23 | ||
| 24 | test('versions are scoped per repo — same version string, different repo', (t) => { | |
| 25 | const s = new DeletedReleases() | |
| 26 | s.applyEntry({ op: OP_DELETED, repo: 'a', version: 'v1' }, 0) | |
| 27 | t.ok(s.has('a', 'v1')) | |
| 28 | t.absent(s.has('b', 'v1'), 'a delete in repo a says nothing about repo b') | |
| 29 | }) | |
| 30 | ||
| 31 | test('later seq wins, and an out-of-order replay cannot clobber newer state', (t) => { | |
| 32 | const s = new DeletedReleases() | |
| 33 | s.applyEntry({ op: OP_DELETED, repo: 'r', version: 'v1' }, 5) | |
| 34 | t.ok(s.has('r', 'v1')) | |
| 35 | ||
| 36 | // An older entry arriving late must not un-delete it. | |
| 37 | s.applyEntry({ op: OP_PUBLISHED, repo: 'r', version: 'v1' }, 2) | |
| 38 | t.ok(s.has('r', 'v1'), 'stale replay ignored') | |
| 39 | ||
| 40 | // A genuine republish at a NEWER seq revives it — mistakes are reversible. | |
| 41 | s.applyEntry({ op: OP_PUBLISHED, repo: 'r', version: 'v1' }, 9) | |
| 42 | t.absent(s.has('r', 'v1'), 'republish revives the version') | |
| 43 | }) | |
| 44 | ||
| 45 | test('channel-only ops carry no version and are ignored', (t) => { | |
| 46 | const s = new DeletedReleases() | |
| 47 | s.applyEntry({ op: OP_DELETED, repo: 'r', version: 'v1' }, 0) | |
| 48 | s.applyEntry({ op: OP_PROMOTED, repo: 'r', version: '', channel: 'stable' }, 1) | |
| 49 | t.ok(s.has('r', 'v1'), 'unaffected by a channel op') | |
| 50 | t.is(s.size, 1) | |
| 51 | }) | |
| 52 | ||
| 53 | test('foldDeleted replays a whole log; isDeleted is a bare predicate', async (t) => { | |
| 54 | const log = { | |
| 55 | async * entries () { | |
| 56 | yield { op: OP_PUBLISHED, repo: 'r', version: 'v1' } | |
| 57 | yield { op: OP_PUBLISHED, repo: 'r', version: 'v2' } | |
| 58 | yield { op: OP_DELETED, repo: 'r', version: 'v1' } | |
| 59 | } | |
| 60 | } | |
| 61 | const s = await foldDeleted(log) | |
| 62 | t.is(s.size, 1) | |
| 63 | ||
| 64 | const pred = s.isDeleted | |
| 65 | t.ok(pred('r', 'v1'), 'predicate works detached from the instance') | |
| 66 | t.absent(pred('r', 'v2')) | |
| 67 | }) | |
| 68 | ||
| 69 | test('foldDeleted tolerates a missing or malformed log', async (t) => { | |
| 70 | t.is((await foldDeleted(null)).size, 0) | |
| 71 | t.is((await foldDeleted({})).size, 0, 'no entries() — returns an empty set, does not throw') | |
| 72 | }) |
| … | … | @@ -470,3 +470,55 @@ test('pullNow closes the client even when the pull throws', async (t) => { |
| 470 | 470 | t.absent(res.ok, 'failure reported, not thrown') |
| 471 | 471 | t.is(closed, 1, 'connection released — a leaked swarm client per failed pull would accumulate') |
| 472 | 472 | }) |
| 473 | ||
| 474 | // ── Tombstones ─────────────────────────────────────────────── | |
| 475 | // | |
| 476 | // A version deleted on the origin must never be fetched again. Without this | |
| 477 | // the puller chases bytes whose only holder is gone and retries forever: | |
| 478 | // [release-pull] error: no holder for favela-android/versions/master-8d988ca/... | |
| 479 | // → fetched 0, replaced 0, skipped 213, errors 1 — PARTIAL, will retry | |
| 480 | // observed in prod 2026-08-29 after `pgh deploy gc` removed versions on one | |
| 481 | // peer. repo-mirror.js already honours the equivalent repo tombstone | |
| 482 | // (OP_RETIRED); the release path emitted OP_DELETED but nothing consumed it. | |
| 483 | ||
| 484 | test('planPull: a tombstoned version is never fetched', (t) => { | |
| 485 | const remote = { | |
| 486 | files: [ | |
| 487 | { path: 'r/versions/v1/a.zip', repo: 'r', version: 'v1', size: 10, hash: 'aaa' }, | |
| 488 | { path: 'r/versions/v2/b.zip', repo: 'r', version: 'v2', size: 10, hash: 'bbb' } | |
| 489 | ] | |
| 490 | } | |
| 491 | const isDeleted = (repo, version) => repo === 'r' && version === 'v1' | |
| 492 | ||
| 493 | const plan = planPull(remote, () => null, isDeleted) | |
| 494 | ||
| 495 | t.is(plan.fetch.length, 1, 'only the live version is fetched') | |
| 496 | t.is(plan.fetch[0].version, 'v2', 'and it is the one not tombstoned') | |
| 497 | t.is(plan.tombstoned.length, 1, 'the dead version is accounted for, not silently dropped') | |
| 498 | t.is(plan.tombstoned[0].version, 'v1') | |
| 499 | }) | |
| 500 | ||
| 501 | test('planPull: a tombstone does not delete local bytes, it only stops fetching', (t) => { | |
| 502 | // The never-delete rule: a tombstone changes what we PULL, never what is on | |
| 503 | // disk. A local copy of a tombstoned version must not be planned for | |
| 504 | // replacement or removal — byte removal stays a manual operator step. | |
| 505 | const remote = { | |
| 506 | files: [{ path: 'r/versions/v1/a.zip', repo: 'r', version: 'v1', size: 10, hash: 'aaa' }] | |
| 507 | } | |
| 508 | const localDiffers = () => ({ size: 99, hash: 'LOCAL-DIFFERENT' }) | |
| 509 | ||
| 510 | const plan = planPull(remote, localDiffers, () => true) | |
| 511 | ||
| 512 | t.is(plan.fetch.length, 0, 'nothing fetched') | |
| 513 | t.is(plan.replace.length, 0, 'local bytes NOT scheduled for replacement') | |
| 514 | t.is(plan.tombstoned.length, 1, 'reported as tombstoned instead') | |
| 515 | }) | |
| 516 | ||
| 517 | test('planPull: omitting the predicate keeps the old behaviour', (t) => { | |
| 518 | const remote = { | |
| 519 | files: [{ path: 'r/versions/v1/a.zip', repo: 'r', version: 'v1', size: 10, hash: 'aaa' }] | |
| 520 | } | |
| 521 | const plan = planPull(remote, () => null) | |
| 522 | t.is(plan.fetch.length, 1, 'no predicate → fetch as before') | |
| 523 | t.alike(plan.tombstoned, [], 'bucket present but empty') | |
| 524 | }) |