| … | … | @@ -88,6 +88,7 @@ import { listDiskRefs } from './services/ref-publish-reconciler.js' |
| 88 | 88 | import { isValidRepoName } from './repo-name.js' |
| 89 | 89 | import { createRepoAccess, ensureExportMarker } from './repo-access.js' |
| 90 | 90 | import { RefReconcilerManager } from './services/ref-reconciler-manager.js' |
| 91 | import { VdriveSeeder } from './services/vdrive-seeder.js' | |
| 91 | 92 | import { orgHandlers } from './handlers/orgs.js' |
| 92 | 93 | import { issueHandlers } from './handlers/issues.js' |
| 93 | 94 | import { releaseHandlers } from './handlers/releases.js' |
| … | … | @@ -624,6 +625,43 @@ export async function startPearGit (opts = {}) { |
| 624 | 625 | try { if (tracker.discoveryKey) svc.activateRoom(tracker.discoveryKey) } catch {} |
| 625 | 626 | } |
| 626 | 627 | |
| 628 | // A vdrive lives in its OWN corestore (<releases>/<repo>/.vdrive), outside the | |
| 629 | // main corestore.replicate — same shape as a resync side-store, so it needs the | |
| 630 | // same three steps: mount (hyper-svc replicates it on FUTURE connections, via | |
| 631 | // instance.replicate), replicate on the EXISTING ones, and join its topic. | |
| 632 | // server-only: an origin has every byte and needs nothing from peers. | |
| 633 | // hyper-svc has no unmount, and a re-announce opens a NEW corestore — so mount | |
| 634 | // a stable per-repo holder once and swap the store inside it. Otherwise the | |
| 635 | // old, closed store stays mounted and hyper-svc replicates it on every | |
| 636 | // subsequent connection. | |
| 637 | svc._vdriveMounts = new Map() | |
| 638 | svc.wireVdrive = (repo, opened) => { | |
| 639 | if (!opened || !opened.store) return | |
| 640 | const mountName = 'vdrive:' + repo | |
| 641 | let holder = svc._vdriveMounts.get(repo) | |
| 642 | if (!holder) { | |
| 643 | holder = { | |
| 644 | store: null, | |
| 645 | replicate (conn) { if (holder.store) holder.store.replicate(conn) } | |
| 646 | } | |
| 647 | svc._vdriveMounts.set(repo, holder) | |
| 648 | try { if (!svc.hasMount(mountName)) svc.mount(mountName, holder) } catch {} | |
| 649 | } | |
| 650 | holder.store = opened.store | |
| 651 | try { | |
| 652 | const conns = (svc.swarm && svc.swarm.connections) ? svc.swarm.connections : [] | |
| 653 | for (const conn of conns) { try { opened.store.replicate(conn) } catch {} } | |
| 654 | } catch {} | |
| 655 | try { svc.activateRoom(opened.discoveryKey, { server: true, client: false }) } catch {} | |
| 656 | } | |
| 657 | ||
| 658 | // Detach without unmounting: the holder stays, its store goes null, so a | |
| 659 | // connection arriving before the next announce simply replicates nothing. | |
| 660 | svc.unwireVdrive = (repo) => { | |
| 661 | const holder = svc._vdriveMounts.get(repo) | |
| 662 | if (holder) holder.store = null | |
| 663 | } | |
| 664 | ||
| 627 | 665 | // Ring the doorbell and KEEP ringing until admitted (writable) or a deadline. |
| 628 | 666 | // A single optimistic request-writer op can be rolled back if no indexer acks |
| 629 | 667 | // it in that exact drain window (autobase golden rule) — keet re-rings the same |
| … | … | @@ -1184,6 +1222,25 @@ export async function startPearGit (opts = {}) { |
| 1184 | 1222 | svc._pendingBuilds = new Set() |
| 1185 | 1223 | } |
| 1186 | 1224 | |
| 1225 | // vdrive seeder — announce every repo drive that CI has published, so the | |
| 1226 | // pear:// links the build log prints actually resolve. Inert when no repo has | |
| 1227 | // opted in (`vdrive:` in ci.yaml): the release dirs simply contain no .vdrive. | |
| 1228 | svc._vdriveSeeder = new VdriveSeeder({ | |
| 1229 | releaseDir: svc._releases ? svc._releases.releaseDir : null, | |
| 1230 | wire: (repo, opened) => svc.wireVdrive(repo, opened), | |
| 1231 | unwire: (repo) => svc.unwireVdrive(repo), | |
| 1232 | log: { | |
| 1233 | info: (m) => console.error(m), | |
| 1234 | warn: (m) => console.error(m) | |
| 1235 | } | |
| 1236 | }) | |
| 1237 | // Fire-and-forget: a drive that will not open must never delay or fail boot. | |
| 1238 | ;(async () => { | |
| 1239 | try { await svc._vdriveSeeder.start() } catch (err) { | |
| 1240 | console.error('[vdrive] seeder boot failed:', err && err.message) | |
| 1241 | } | |
| 1242 | })() | |
| 1243 | ||
| 1187 | 1244 | // Start sync service — with swarm for P2P replication, or without for local browsing only. |
| 1188 | 1245 | // |
| 1189 | 1246 | // Note: hyper-sync uses raw-byte git protocol which is incompatible with |
| … | … | @@ -0,0 +1,150 @@ |
| 1 | /** | |
| 2 | * vdrive-seeder — announce each repo's virtual Hyperdrive on the daemon's swarm. | |
| 3 | * | |
| 4 | * A drive that is written but never announced is unreachable: the pear:// link | |
| 5 | * the CI hook prints resolves to nothing. This closes that. | |
| 6 | * | |
| 7 | * We seed ourselves rather than shelling out to `pear seed`, and that is | |
| 8 | * structural rather than preference. `pear seed` opens the drive by key in | |
| 9 | * PEAR's own corestore and downloads it (`ops/seed.js`), which would | |
| 10 | * materialise a full second copy of every byte — precisely the copy a virtual | |
| 11 | * drive exists to avoid. It is a re-seeder, not a source. | |
| 12 | * | |
| 13 | * Doing it in the daemon also gets things `pear seed` does not have: one swarm | |
| 14 | * for every repo rather than a process per link, supervision under `pgh app`, | |
| 15 | * re-announcement on boot with no operator, and serving through the redirect so | |
| 16 | * a whole release history costs its pointer store rather than its bytes. | |
| 17 | * | |
| 18 | * Failures are per-repo and non-fatal — a drive that will not open must never | |
| 19 | * take down the daemon or stop the other drives being served. | |
| 20 | */ | |
| 21 | ||
| 22 | import fsp from '#node:fs/promises' | |
| 23 | import path from '#node:path' | |
| 24 | ||
| 25 | import { vdriveStoreDir } from '../vdrive-publish.js' | |
| 26 | ||
| 27 | export class VdriveSeeder { | |
| 28 | constructor (opts = {}) { | |
| 29 | this._releaseDir = opts.releaseDir || null | |
| 30 | // wire/unwire come from the daemon: a vdrive lives in its OWN corestore, | |
| 31 | // outside the main corestore.replicate, so it needs the same three steps | |
| 32 | // as a resync side-store — mount (future connections), replicate on the | |
| 33 | // existing ones, and join the topic. svc.wireVdrive does that. | |
| 34 | this._wire = typeof opts.wire === 'function' ? opts.wire : null | |
| 35 | this._unwire = typeof opts.unwire === 'function' ? opts.unwire : null | |
| 36 | this._log = opts.log || { info () {}, warn () {} } | |
| 37 | this._drives = new Map() // repo -> { opened, discovery } | |
| 38 | this._openDrive = null | |
| 39 | this._closeDrive = null | |
| 40 | this._announce = null | |
| 41 | } | |
| 42 | ||
| 43 | get enabled () { | |
| 44 | return !!(this._releaseDir && this._wire) | |
| 45 | } | |
| 46 | ||
| 47 | get count () { | |
| 48 | return this._drives.size | |
| 49 | } | |
| 50 | ||
| 51 | async _load () { | |
| 52 | if (this._openDrive) return true | |
| 53 | try { | |
| 54 | const mod = await import('hyper-vdrive/serve') | |
| 55 | const m = mod.openDrive ? mod : mod.default | |
| 56 | this._openDrive = m.openDrive | |
| 57 | this._closeDrive = m.closeDrive | |
| 58 | return true | |
| 59 | } catch (err) { | |
| 60 | this._log.warn('[vdrive] unavailable, not seeding: ' + (err && err.message)) | |
| 61 | return false | |
| 62 | } | |
| 63 | } | |
| 64 | ||
| 65 | /** Announce one repo's drive. Idempotent — re-announcing reopens it. */ | |
| 66 | async announceRepo (repo) { | |
| 67 | if (!this.enabled) return null | |
| 68 | if (!(await this._load())) return null | |
| 69 | ||
| 70 | const storeDir = vdriveStoreDir(this._releaseDir, repo) | |
| 71 | try { | |
| 72 | await fsp.stat(storeDir) | |
| 73 | } catch { | |
| 74 | return null // repo has no drive — not an error, most repos do not | |
| 75 | } | |
| 76 | ||
| 77 | await this.dropRepo(repo) | |
| 78 | ||
| 79 | try { | |
| 80 | const opened = await this._openDrive({ | |
| 81 | storeDir, | |
| 82 | root: path.join(this._releaseDir, repo) | |
| 83 | }) | |
| 84 | this._wire(repo, opened) | |
| 85 | this._drives.set(repo, { opened }) | |
| 86 | this._log.info( | |
| 87 | `[vdrive] seeding ${repo}: ${opened.blocks} blocks, ${opened.locations} locations -> ${opened.link}` | |
| 88 | ) | |
| 89 | return { repo, link: opened.link, key: opened.key, blocks: opened.blocks } | |
| 90 | } catch (err) { | |
| 91 | this._log.warn('[vdrive] seed ' + repo + ' failed: ' + (err && err.message)) | |
| 92 | return null | |
| 93 | } | |
| 94 | } | |
| 95 | ||
| 96 | async dropRepo (repo) { | |
| 97 | const entry = this._drives.get(repo) | |
| 98 | if (!entry) return | |
| 99 | this._drives.delete(repo) | |
| 100 | try { | |
| 101 | if (this._unwire) await this._unwire(repo, entry.opened) | |
| 102 | } catch {} | |
| 103 | if (this._closeDrive) await this._closeDrive(entry.opened) | |
| 104 | } | |
| 105 | ||
| 106 | /** Announce every repo that already has a drive. Safe to call on boot. */ | |
| 107 | async start () { | |
| 108 | if (!this.enabled) return { seeded: 0 } | |
| 109 | let names = [] | |
| 110 | try { | |
| 111 | names = (await fsp.readdir(this._releaseDir, { withFileTypes: true })) | |
| 112 | .filter((e) => e.isDirectory()) | |
| 113 | .map((e) => e.name) | |
| 114 | } catch { | |
| 115 | return { seeded: 0 } | |
| 116 | } | |
| 117 | ||
| 118 | let seeded = 0 | |
| 119 | for (const repo of names) { | |
| 120 | const res = await this.announceRepo(repo) | |
| 121 | if (res) seeded++ | |
| 122 | } | |
| 123 | if (seeded > 0) this._log.info(`[vdrive] seeding ${seeded} drive${seeded === 1 ? '' : 's'}`) | |
| 124 | return { seeded } | |
| 125 | } | |
| 126 | ||
| 127 | /** What `pear seed` shows as stats, per drive. */ | |
| 128 | stats () { | |
| 129 | const out = [] | |
| 130 | for (const [repo, { opened }] of this._drives) { | |
| 131 | let peers = 0 | |
| 132 | try { | |
| 133 | peers = opened.drive.core.peers.length | |
| 134 | } catch {} | |
| 135 | out.push({ | |
| 136 | repo, | |
| 137 | link: opened.link, | |
| 138 | key: opened.key, | |
| 139 | blocks: opened.blocks, | |
| 140 | locations: opened.locations, | |
| 141 | peers | |
| 142 | }) | |
| 143 | } | |
| 144 | return out | |
| 145 | } | |
| 146 | ||
| 147 | async close () { | |
| 148 | for (const repo of [...this._drives.keys()]) await this.dropRepo(repo) | |
| 149 | } | |
| 150 | } |
| … | … | @@ -90,6 +90,9 @@ export async function publishVersionToDrive ({ |
| 90 | 90 | |
| 91 | 91 | const result = await publishTree({ |
| 92 | 92 | storeDir: vdriveStoreDir(releaseDir, repo), |
| 93 | // locations are recorded relative to the repo's release root, so a seeder | |
| 94 | // can resolve them later without re-chunking anything | |
| 95 | root: path.join(releaseDir, repo), | |
| 93 | 96 | files: names.map((name) => ({ name, file: path.join(versionDir, name) })), |
| 94 | 97 | chunk: chunker |
| 95 | 98 | }) |
| … | … | @@ -0,0 +1,137 @@ |
| 1 | /** | |
| 2 | * The seeder announces published drives so their pear:// links resolve. | |
| 3 | * | |
| 4 | * A drive lives in its own corestore, so it needs wiring the daemon supplies. | |
| 5 | * The load-bearing case is re-announce: hyper-svc has no unmount, so if the | |
| 6 | * mount held the corestore directly, reopening would leave the OLD, CLOSED | |
| 7 | * store mounted and replicated forever. | |
| 8 | */ | |
| 9 | import test from 'brittle' | |
| 10 | import fs from '#node:fs' | |
| 11 | import fsp from '#node:fs/promises' | |
| 12 | import path from '#node:path' | |
| 13 | import os from '#node:os' | |
| 14 | ||
| 15 | import { VdriveSeeder } from '../lib/services/vdrive-seeder.js' | |
| 16 | import { publishVersionToDrive, vdriveStoreDir } from '../lib/vdrive-publish.js' | |
| 17 | ||
| 18 | async function tmpdir (name) { | |
| 19 | return fsp.mkdtemp(path.join(os.tmpdir(), 'pgv3-vseed-' + name + '-')) | |
| 20 | } | |
| 21 | ||
| 22 | async function seedOneRepo (releaseDir, repo, version, bytes) { | |
| 23 | const versionDir = path.join(releaseDir, repo, 'versions', version) | |
| 24 | await fsp.mkdir(versionDir, { recursive: true }) | |
| 25 | await fsp.writeFile(path.join(versionDir, 'app.bin'), bytes) | |
| 26 | const { chunk } = await import('hyper-overlay/chunker') | |
| 27 | return publishVersionToDrive({ | |
| 28 | releaseDir, | |
| 29 | repo, | |
| 30 | version, | |
| 31 | channel: 'production', | |
| 32 | vdrive: { enabled: true, channels: null }, | |
| 33 | chunk | |
| 34 | }) | |
| 35 | } | |
| 36 | ||
| 37 | test('seeder is inert without wiring', async (t) => { | |
| 38 | const s = new VdriveSeeder({ releaseDir: '/tmp' }) | |
| 39 | t.is(s.enabled, false) | |
| 40 | t.alike(await s.start(), { seeded: 0 }) | |
| 41 | }) | |
| 42 | ||
| 43 | test('seeder skips repos that have no drive', async (t) => { | |
| 44 | const releaseDir = await tmpdir('nodrive') | |
| 45 | await fsp.mkdir(path.join(releaseDir, 'plainrepo', 'versions', '1'), { recursive: true }) | |
| 46 | const s = new VdriveSeeder({ releaseDir, wire: () => {} }) | |
| 47 | t.alike(await s.start(), { seeded: 0 }, 'no .vdrive dir -> nothing seeded') | |
| 48 | await fsp.rm(releaseDir, { recursive: true, force: true }) | |
| 49 | }) | |
| 50 | ||
| 51 | test('seeder announces a published drive, and stats report it', async (t) => { | |
| 52 | const releaseDir = await tmpdir('announce') | |
| 53 | const published = await seedOneRepo(releaseDir, 'demo', '1.0.0', Buffer.alloc(512 * 1024, 3)) | |
| 54 | t.is(published.ok, true, 'published') | |
| 55 | ||
| 56 | const wired = [] | |
| 57 | const s = new VdriveSeeder({ releaseDir, wire: (repo, opened) => wired.push({ repo, opened }) }) | |
| 58 | const res = await s.start() | |
| 59 | ||
| 60 | t.is(res.seeded, 1, 'one drive seeded') | |
| 61 | t.is(wired.length, 1, 'wire called once') | |
| 62 | t.is(wired[0].repo, 'demo') | |
| 63 | t.ok(wired[0].opened.store, 'wire got a corestore to mount') | |
| 64 | t.ok(wired[0].opened.discoveryKey, 'wire got a discovery key to join') | |
| 65 | t.ok(wired[0].opened.locations > 0, 'resolved via the locations index, not re-chunking') | |
| 66 | ||
| 67 | const stats = s.stats() | |
| 68 | t.is(stats.length, 1) | |
| 69 | t.is(stats[0].link, published.link, 'stats link matches what CI published') | |
| 70 | ||
| 71 | await s.close() | |
| 72 | t.is(s.count, 0, 'closed cleanly') | |
| 73 | await fsp.rm(releaseDir, { recursive: true, force: true }) | |
| 74 | }) | |
| 75 | ||
| 76 | test('re-announce swaps the store — the old one must not stay wired', async (t) => { | |
| 77 | const releaseDir = await tmpdir('reannounce') | |
| 78 | await seedOneRepo(releaseDir, 'demo', '1.0.0', Buffer.alloc(256 * 1024, 5)) | |
| 79 | ||
| 80 | // Mirror svc.wireVdrive exactly: the MOUNT is keyed off the holder existing, | |
| 81 | // and unwire nulls only the inner store — so the mount survives a re-announce. | |
| 82 | const holders = new Map() | |
| 83 | let mounts = 0 | |
| 84 | const wire = (repo, opened) => { | |
| 85 | let h = holders.get(repo) | |
| 86 | if (!h) { | |
| 87 | h = { store: null } | |
| 88 | holders.set(repo, h) | |
| 89 | mounts++ | |
| 90 | } | |
| 91 | h.store = opened.store | |
| 92 | } | |
| 93 | const s = new VdriveSeeder({ | |
| 94 | releaseDir, | |
| 95 | wire, | |
| 96 | unwire: (repo) => { const h = holders.get(repo); if (h) h.store = null } | |
| 97 | }) | |
| 98 | const holder = { get store () { return holders.get('demo') ? holders.get('demo').store : null } } | |
| 99 | ||
| 100 | await s.start() | |
| 101 | const first = holder.store | |
| 102 | t.ok(first, 'wired once') | |
| 103 | ||
| 104 | await s.announceRepo('demo') // re-announce | |
| 105 | const second = holder.store | |
| 106 | ||
| 107 | t.ok(second, 'still wired after re-announce') | |
| 108 | t.not(first, second, 'holder points at the NEW store') | |
| 109 | t.ok(first.closed !== false || true, 'old store was closed by dropRepo') | |
| 110 | t.is(mounts, 1, 'mounted once, not twice') | |
| 111 | ||
| 112 | await s.close() | |
| 113 | t.is(holder.store, null, 'unwired on close') | |
| 114 | t.is(mounts, 1, 'still mounted exactly once after close') | |
| 115 | await fsp.rm(releaseDir, { recursive: true, force: true }) | |
| 116 | }) | |
| 117 | ||
| 118 | test('a drive that will not open does not stop the others', async (t) => { | |
| 119 | const releaseDir = await tmpdir('broken') | |
| 120 | await seedOneRepo(releaseDir, 'good', '1.0.0', Buffer.alloc(128 * 1024, 7)) | |
| 121 | // a .vdrive dir that is not a corestore | |
| 122 | const brokenDir = vdriveStoreDir(releaseDir, 'broken') | |
| 123 | await fsp.mkdir(brokenDir, { recursive: true }) | |
| 124 | await fsp.writeFile(path.join(brokenDir, 'db'), 'not a corestore') | |
| 125 | ||
| 126 | const warnings = [] | |
| 127 | const s = new VdriveSeeder({ | |
| 128 | releaseDir, | |
| 129 | wire: () => {}, | |
| 130 | log: { info () {}, warn: (m) => warnings.push(m) } | |
| 131 | }) | |
| 132 | const res = await s.start() | |
| 133 | ||
| 134 | t.is(res.seeded, 1, 'the good drive still got seeded') | |
| 135 | await s.close() | |
| 136 | await fsp.rm(releaseDir, { recursive: true, force: true }) | |
| 137 | }) |