| … | … | @@ -153,7 +153,14 @@ async function handleCommand (command, serverKey, repoName, httpsUrl, stdout, st |
| 153 | 153 | async function handleConnect (service, serverKey, repoName, httpsUrl, stdout, stderr) { |
| 154 | 154 | stderr.write('[punch] Connecting to ' + serverKey.slice(0, 8) + '.../' + repoName + '\n') |
| 155 | 155 | |
| 156 | const connector = new TopicConnector({ timeout: 15000 }) | |
| 156 | // Durable, event-driven connect (see lib/durable-connect.js): keep the swarm | |
| 157 | // joined, take connections as they arrive, NO aggressive deadline racing the | |
| 158 | // DHT+hole-punch. `fallbackAfter` is a GENEROUS no-progress escape only — if | |
| 159 | // NO peer is reached within it, we fall through to the HTTPS path below | |
| 160 | // rather than hang forever. It must never race a live connect (the old hard | |
| 161 | // 15s did, and starved concurrent org-repo pulls into ZERO-object failures — | |
| 162 | // pear-git org issue #3). The data transfer itself stays unbounded. | |
| 163 | const connector = new TopicConnector({ fallbackAfter: 120000 }) | |
| 157 | 164 | await connector.init() |
| 158 | 165 | |
| 159 | 166 | try { |
| … | … | @@ -0,0 +1,112 @@ |
| 1 | /** | |
| 2 | * durableConnect — event-driven, no-aggressive-deadline connect over an | |
| 3 | * injected Hyperswarm. | |
| 4 | * | |
| 5 | * Replaces the fragile "swarm.once('connection') + setTimeout(15000, reject)" | |
| 6 | * model that this transport used to use. That model treats a P2P connection as | |
| 7 | * a synchronous timed dial and gives up when the stopwatch wins — which is what | |
| 8 | * starved concurrent org-repo pulls: N `git fetch punch://` each spin their own | |
| 9 | * swarm, compete for DHT/hole-punch budget, and the losers of that race hit the | |
| 10 | * 15s deadline and fail with ZERO objects (observed on ashatron 2026-07-18: | |
| 11 | * repos that started in the same burst all failed empty; less-contended ones | |
| 12 | * completed). See pear-git org issue #3. | |
| 13 | * | |
| 14 | * Instead we treat a connection as an event-driven resource — matching how keet | |
| 15 | * and hyper-overlay stay durable: | |
| 16 | * | |
| 17 | * - join(topic, {client:true}) once, then swarm.on('connection') — react to | |
| 18 | * EVERY connection as it arrives, and STAY joined so hyperswarm keeps | |
| 19 | * re-forming connections across peers while the topic is joined. | |
| 20 | * - NO wall-clock connect deadline racing the DHT+hole-punch. A connection | |
| 21 | * that takes 2s or 40s under contention is fine — it just waits. | |
| 22 | * - Reconcile ACROSS connections: resolve on the FIRST connection that passes | |
| 23 | * `validate(conn, info)`; a connection that fails validation is discarded | |
| 24 | * and the next is taken (never first-wins-fail). | |
| 25 | * - Cancellable via an AbortSignal AND, for callers that have a fallback path | |
| 26 | * (git-remote-punch has an HTTPS fallback), an OPTIONAL generous | |
| 27 | * `fallbackAfter` no-progress bound. This is a last-resort escape so an | |
| 28 | * offline server doesn't hang forever — NOT a connect deadline. It must be | |
| 29 | * long enough to never race a live hole-punch under load; the aggressive | |
| 30 | * 15s was the bug, a 120s escape is not. | |
| 31 | * | |
| 32 | * The swarm is INJECTED and never created/destroyed here — the caller owns its | |
| 33 | * lifetime (for the git-remote-punch helper that is one process; for a daemon | |
| 34 | * it is a long-lived shared swarm). | |
| 35 | * | |
| 36 | * @param {object} swarm - a joined-capable Hyperswarm (has join + on/off) | |
| 37 | * @param {Buffer} topic - 32-byte swarm topic | |
| 38 | * @param {object} [opts] | |
| 39 | * @param {(conn, info) => (boolean|Promise<boolean>)} [opts.validate] - liveness | |
| 40 | * predicate; default accepts the first connection. | |
| 41 | * @param {AbortSignal} [opts.signal] - caller cancellation. | |
| 42 | * @param {number} [opts.fallbackAfter] - ms; if > 0, reject after this long | |
| 43 | * with NO validated connection so the caller can fall back. 0/undefined = | |
| 44 | * wait indefinitely (the bound belongs to the caller). | |
| 45 | * @returns {Promise<object>} the first connection that passes validate(). | |
| 46 | */ | |
| 47 | export function durableConnect (swarm, topic, opts = {}) { | |
| 48 | const validate = opts.validate || (() => true) | |
| 49 | const signal = opts.signal | |
| 50 | const fallbackAfter = Number(opts.fallbackAfter) || 0 | |
| 51 | ||
| 52 | return new Promise((resolve, reject) => { | |
| 53 | let settled = false | |
| 54 | let fallbackTimer = null | |
| 55 | ||
| 56 | const cleanup = () => { | |
| 57 | swarm.off('connection', onConnection) | |
| 58 | if (fallbackTimer) clearTimeout(fallbackTimer) | |
| 59 | if (signal) { try { signal.removeEventListener('abort', onAbort) } catch {} } | |
| 60 | } | |
| 61 | ||
| 62 | const onConnection = (conn, info) => { | |
| 63 | if (settled) return | |
| 64 | Promise.resolve(validate(conn, info)) | |
| 65 | .then((ok) => { | |
| 66 | if (settled) return | |
| 67 | if (ok) { | |
| 68 | settled = true | |
| 69 | cleanup() | |
| 70 | resolve(conn) | |
| 71 | } else { | |
| 72 | // Stale/dead/wrong peer — discard and keep waiting. hyperswarm | |
| 73 | // keeps forming connections to other (live) peers while joined. | |
| 74 | try { conn.destroy() } catch {} | |
| 75 | } | |
| 76 | }) | |
| 77 | .catch(() => { try { conn.destroy() } catch {} }) | |
| 78 | } | |
| 79 | ||
| 80 | const onAbort = () => { | |
| 81 | if (settled) return | |
| 82 | settled = true | |
| 83 | cleanup() | |
| 84 | reject(new Error('durable connect aborted')) | |
| 85 | } | |
| 86 | ||
| 87 | if (signal) { | |
| 88 | if (signal.aborted) return onAbort() | |
| 89 | signal.addEventListener('abort', onAbort) | |
| 90 | } | |
| 91 | ||
| 92 | // Generous no-progress escape (NOT a connect race) — only for callers that | |
| 93 | // opt in because they have a fallback (e.g. HTTPS). Waits by default. | |
| 94 | if (fallbackAfter > 0) { | |
| 95 | fallbackTimer = setTimeout(() => { | |
| 96 | if (settled) return | |
| 97 | settled = true | |
| 98 | cleanup() | |
| 99 | reject(new Error('durable connect: no peer reached after ' + fallbackAfter + 'ms')) | |
| 100 | }, fallbackAfter) | |
| 101 | // NOT unref'd: this bound MUST fire even when nothing else keeps the loop | |
| 102 | // alive, so an offline server falls back instead of hanging. cleanup() | |
| 103 | // clears it the moment a connection validates or the caller aborts. | |
| 104 | } | |
| 105 | ||
| 106 | // on(), not once() — react to EVERY connection until one validates. | |
| 107 | swarm.on('connection', onConnection) | |
| 108 | swarm.join(topic, { server: false, client: true }) | |
| 109 | }) | |
| 110 | } | |
| 111 | ||
| 112 | export default durableConnect |
| … | … | @@ -5,6 +5,7 @@ import crypto from 'hypercore-crypto' |
| 5 | 5 | import Hyperswarm from 'hyperswarm' |
| 6 | 6 | |
| 7 | 7 | import { deriveTopicFromKey } from './derive-topic.js' |
| 8 | import { durableConnect } from './durable-connect.js' | |
| 8 | 9 | |
| 9 | 10 | /** |
| 10 | 11 | * Client-side P2P connector for punch:// URLs. |
| … | … | @@ -46,26 +47,27 @@ export class TopicConnector { |
| 46 | 47 | * @param {string} repoName - Repository name |
| 47 | 48 | * @returns {Promise<Stream>} - Encrypted Noise connection |
| 48 | 49 | */ |
| 49 | async connect (serverKey, repoName) { | |
| 50 | async connect (serverKey, repoName, opts = {}) { | |
| 50 | 51 | const topic = deriveTopicFromKey(serverKey, repoName) |
| 51 | 52 | |
| 52 | return new Promise((resolve, reject) => { | |
| 53 | const timeout = setTimeout(() => { | |
| 54 | reject(new Error('Connection timeout for ' + repoName)) | |
| 55 | }, this.options.timeout || 30000) | |
| 56 | ||
| 57 | this.swarm.join(topic, { server: false, client: true }) | |
| 58 | ||
| 59 | this.swarm.once('connection', (conn) => { | |
| 60 | clearTimeout(timeout) | |
| 61 | resolve(conn) | |
| 62 | }) | |
| 63 | ||
| 64 | this.swarm.once('error', (err) => { | |
| 65 | clearTimeout(timeout) | |
| 66 | reject(err) | |
| 67 | }) | |
| 68 | }) | |
| 53 | // Durable, event-driven connect (lib/durable-connect.js). NO aggressive | |
| 54 | // wall-clock deadline racing the DHT+hole-punch: keep the swarm joined, | |
| 55 | // react to EVERY connection (on, not once), reconcile across connections. | |
| 56 | // | |
| 57 | // WHY: a punch:// connection forms via DHT lookup + hole-punch, inherently | |
| 58 | // variable and much slower under CONCURRENCY — N `git fetch punch://` (the | |
| 59 | // org-repo reverse puller) each spin their own swarm and compete for | |
| 60 | // hole-punch budget. The old hard 15s deadline killed the losers of that | |
| 61 | // race, which failed with ZERO objects (observed on ashatron 2026-07-18: | |
| 62 | // repos that started in the same burst all failed empty; less-contended | |
| 63 | // ones completed). Removing the aggressive deadline lets the losers WAIT | |
| 64 | // out the contention instead of dying. See pear-git org issue #3. | |
| 65 | // | |
| 66 | // `fallbackAfter`/`timeout` (ms) is a GENEROUS no-progress escape ONLY — | |
| 67 | // it lets the helper's HTTPS fallback fire on a truly offline server. It | |
| 68 | // must never race a live connect. 0/undefined = wait indefinitely. | |
| 69 | const fallbackAfter = Number(this.options.fallbackAfter || this.options.timeout) || 0 | |
| 70 | return durableConnect(this.swarm, topic, { signal: opts.signal, fallbackAfter }) | |
| 69 | 71 | } |
| 70 | 72 | |
| 71 | 73 | /** |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "git-remote-punch", |
| 3 | "version": "0.1.2", | |
| 3 | "version": "0.1.3", | |
| 4 | 4 | "description": "Git remote helper for punch:// URLs — P2P git transport over Hyperswarm", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "bin": { |
| … | … | @@ -0,0 +1,119 @@ |
| 1 | import test from 'brittle' | |
| 2 | import { durableConnect } from '../lib/durable-connect.js' | |
| 3 | ||
| 4 | // ── minimal fakes (no runtime-specific imports, runs under node or bare) ── | |
| 5 | ||
| 6 | class Emitter { | |
| 7 | constructor () { this._l = new Map() } | |
| 8 | on (e, fn) { if (!this._l.has(e)) this._l.set(e, new Set()); this._l.get(e).add(fn); return this } | |
| 9 | off (e, fn) { if (this._l.has(e)) this._l.get(e).delete(fn); return this } | |
| 10 | emit (e, ...a) { if (this._l.has(e)) for (const fn of [...this._l.get(e)]) fn(...a) } | |
| 11 | } | |
| 12 | ||
| 13 | class FakeSwarm extends Emitter { | |
| 14 | constructor () { super(); this.joined = [] } | |
| 15 | join (topic, opts) { this.joined.push({ topic, opts }) } | |
| 16 | } | |
| 17 | ||
| 18 | class FakeConn extends Emitter { | |
| 19 | constructor (id) { super(); this.id = id; this.destroyed = false } | |
| 20 | destroy () { this.destroyed = true } | |
| 21 | } | |
| 22 | ||
| 23 | const TOPIC = Buffer.alloc(32, 7) | |
| 24 | const later = (ms, fn) => setTimeout(fn, ms) | |
| 25 | ||
| 26 | // The OLD connect model this transport used — the control. Reproduces the bug: | |
| 27 | // swarm.once('connection') + a wall-clock reject timer that races the DHT. | |
| 28 | function legacyConnect (swarm, topic, { timeout }) { | |
| 29 | return new Promise((resolve, reject) => { | |
| 30 | const t = setTimeout(() => reject(new Error('Connection timeout')), timeout) | |
| 31 | let done = false | |
| 32 | const onc = (conn) => { if (done) return; done = true; clearTimeout(t); swarm.off('connection', onc); resolve(conn) } | |
| 33 | swarm.on('connection', onc) | |
| 34 | swarm.join(topic, { server: false, client: true }) | |
| 35 | }) | |
| 36 | } | |
| 37 | ||
| 38 | // ── CONTROL: the old model FAILS a connection that arrives after the deadline ── | |
| 39 | test('control — legacy once+timer rejects a late connection (reproduces the bug)', async (t) => { | |
| 40 | const swarm = new FakeSwarm() | |
| 41 | const p = legacyConnect(swarm, TOPIC, { timeout: 50 }) | |
| 42 | later(150, () => swarm.emit('connection', new FakeConn('late'))) // arrives after 50ms deadline | |
| 43 | await t.exception(p, /Connection timeout/, 'a live-but-late peer is lost to the wall-clock timer') | |
| 44 | }) | |
| 45 | ||
| 46 | // ── MECHANISM: durableConnect resolves the SAME late connection ── | |
| 47 | test('mechanism — durableConnect resolves a late connection (no aggressive deadline)', async (t) => { | |
| 48 | const swarm = new FakeSwarm() | |
| 49 | const conn = new FakeConn('late') | |
| 50 | const p = durableConnect(swarm, TOPIC, {}) // no fallbackAfter → waits | |
| 51 | later(150, () => swarm.emit('connection', conn)) | |
| 52 | const got = await p | |
| 53 | t.is(got, conn, 'the late connection wins — it just waited instead of dying') | |
| 54 | t.is(swarm.joined.length, 1, 'joined the topic as client once') | |
| 55 | t.is(swarm.joined[0].opts.client, true, 'client:true') | |
| 56 | }) | |
| 57 | ||
| 58 | // ── THE ASSERTION: concurrency starves losers under the timer; durable fixes it ── | |
| 59 | // Models the observed ashatron failure: N concurrent `git fetch punch://`, each | |
| 60 | // its own swarm, sharing a contended connect budget so connections arrive | |
| 61 | // STAGGERED. Under the 15s-style per-connector deadline the ones behind the | |
| 62 | // contention exceed it and fail EMPTY; durableConnect lets every one land. | |
| 63 | test('repro — 5 concurrent pulls, staggered connects: legacy starves losers, durable lands all', async (t) => { | |
| 64 | const N = 5 | |
| 65 | const DEADLINE = 100 | |
| 66 | // Staggered arrival times (a shared, contended connect budget serializes them). | |
| 67 | const arrivals = [40, 90, 140, 190, 240] // 2 within the deadline, 3 past it | |
| 68 | ||
| 69 | // Legacy: each connector has its own swarm + its own deadline timer. | |
| 70 | const legacyResults = await Promise.allSettled(arrivals.map((ms) => { | |
| 71 | const swarm = new FakeSwarm() | |
| 72 | const p = legacyConnect(swarm, TOPIC, { timeout: DEADLINE }) | |
| 73 | later(ms, () => swarm.emit('connection', new FakeConn('l' + ms))) | |
| 74 | return p | |
| 75 | })) | |
| 76 | const legacyOk = legacyResults.filter((r) => r.status === 'fulfilled').length | |
| 77 | t.is(legacyOk, 2, 'legacy: only the 2 that beat the deadline complete — 3 starve EMPTY (the bug)') | |
| 78 | ||
| 79 | // Durable: same staggered arrivals, no aggressive deadline. | |
| 80 | const durableResults = await Promise.allSettled(arrivals.map((ms) => { | |
| 81 | const swarm = new FakeSwarm() | |
| 82 | const p = durableConnect(swarm, TOPIC, {}) | |
| 83 | later(ms, () => swarm.emit('connection', new FakeConn('d' + ms))) | |
| 84 | return p | |
| 85 | })) | |
| 86 | const durableOk = durableResults.filter((r) => r.status === 'fulfilled').length | |
| 87 | t.is(durableOk, N, 'durable: ALL 5 land — the losers waited out the contention instead of failing') | |
| 88 | }) | |
| 89 | ||
| 90 | // ── reconcile across connections: discard a bad peer, take the next ── | |
| 91 | test('reconcile — a connection that fails validate is discarded, the next is taken', async (t) => { | |
| 92 | const swarm = new FakeSwarm() | |
| 93 | const bad = new FakeConn('bad') | |
| 94 | const good = new FakeConn('good') | |
| 95 | const p = durableConnect(swarm, TOPIC, { | |
| 96 | validate: (conn) => conn.id === 'good' | |
| 97 | }) | |
| 98 | later(30, () => swarm.emit('connection', bad)) | |
| 99 | later(80, () => swarm.emit('connection', good)) | |
| 100 | const got = await p | |
| 101 | t.is(got, good, 'resolved with the peer that passed validate') | |
| 102 | t.ok(bad.destroyed, 'the failing peer was destroyed, not handed to the caller') | |
| 103 | }) | |
| 104 | ||
| 105 | // ── caller cancellation ── | |
| 106 | test('abort — rejects on AbortSignal, never on a built-in deadline', async (t) => { | |
| 107 | const swarm = new FakeSwarm() | |
| 108 | const ac = new AbortController() | |
| 109 | const p = durableConnect(swarm, TOPIC, { signal: ac.signal }) | |
| 110 | later(20, () => ac.abort()) | |
| 111 | await t.exception(p, /aborted/, 'aborts only when the caller asks') | |
| 112 | }) | |
| 113 | ||
| 114 | // ── generous no-progress escape (for the HTTPS fallback), NOT a connect race ── | |
| 115 | test('fallbackAfter — rejects after the generous bound when NO peer is reached', async (t) => { | |
| 116 | const swarm = new FakeSwarm() | |
| 117 | const p = durableConnect(swarm, TOPIC, { fallbackAfter: 60 }) // nothing ever connects | |
| 118 | await t.exception(p, /no peer reached/, 'offline server falls back instead of hanging forever') | |
| 119 | }) |