| … | … | @@ -377,17 +377,34 @@ export class OverlayProtocol { |
| 377 | 377 | */ |
| 378 | 378 | requestTree (peer, hash, opts = {}) { |
| 379 | 379 | return new Promise((resolve, reject) => { |
| 380 | // If a prior request is already in-flight for the same hash, reject | |
| 381 | // this caller rather than silently overwrite. | |
| 382 | if (peer.pendingTrees.has(hash)) { | |
| 383 | return reject(new Error('tree request already pending for hash: ' + hash)) | |
| 380 | // A request already in flight for the same hash: JOIN it (one request on | |
| 381 | // the wire, every caller settles with its outcome). A tree hash is | |
| 382 | // content-addressed, so they are owed the same entries. This used to | |
| 383 | // reject the second caller, and pgh's materializer (16 subtree fetches | |
| 384 | // at once) failed a whole hyper-tps install on two identical | |
| 385 | // directories (2026-10-11). A joining caller's opts.timeout is ignored. | |
| 386 | const inflight = peer.pendingTrees.get(hash) | |
| 387 | if (inflight) { | |
| 388 | inflight.waiters.push({ resolve, reject }) | |
| 389 | return | |
| 384 | 390 | } |
| 385 | 391 | // Per-request nonce (per-peer monotonic, non-zero) lets the requester drop |
| 386 | 392 | // pages echoed for a superseded attempt. pages/pageBytes buffer this |
| 387 | 393 | // request's own paged response, so a fresh attempt never inherits an old one. |
| 388 | 394 | const nonce = peer._treeReqSeq = (peer._treeReqSeq || 0) + 1 |
| 389 | 395 | const timeout = opts.timeout || this._treeRequestTimeout |
| 390 | const pending = { resolve, reject, timer: null, timeout, nonce, pages: null, pageBytes: 0 } | |
| 396 | // resolve/reject fan out to every waiter; all settle paths call these. | |
| 397 | const waiters = [{ resolve, reject }] | |
| 398 | const pending = { | |
| 399 | waiters, | |
| 400 | resolve: (v) => { for (const w of waiters) w.resolve(v) }, | |
| 401 | reject: (e) => { for (const w of waiters) w.reject(e) }, | |
| 402 | timer: null, | |
| 403 | timeout, | |
| 404 | nonce, | |
| 405 | pages: null, | |
| 406 | pageBytes: 0 | |
| 407 | } | |
| 391 | 408 | this._armTreeTimeout(peer, hash, pending) |
| 392 | 409 | peer.pendingTrees.set(hash, pending) |
| 393 | 410 | peer.msgs.treeRequest.send({ hash, nonce }) |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "hyper-overlay", |
| 3 | "version": "0.3.14", | |
| 3 | "version": "0.3.15", | |
| 4 | 4 | "description": "Content-addressed P2P file distribution over Protomux — FastCDC chunking, multi-source fetch, resumable transfers, zero blob duplication. Files live on disk; the overlay is a thin index and transfer protocol.", |
| 5 | 5 | "type": "module", |
| 6 | 6 | "main": "index.js", |
| … | … | @@ -150,24 +150,32 @@ test('requestTree — times out when peer never responds', async (t) => { |
| 150 | 150 | await B.fileIndex.close() |
| 151 | 151 | }) |
| 152 | 152 | |
| 153 | test('requestTree — rejects concurrent duplicate requests for same hash', async (t) => { | |
| 153 | test('requestTree — concurrent requests for the same hash JOIN one in-flight request', async (t) => { | |
| 154 | // Was: the second caller was rejected ("already pending"). A tree hash is | |
| 155 | // content-addressed, so both callers are owed the same entries. pgh's | |
| 156 | // materializer fetches up to 16 subtrees at once, and a repo with two | |
| 157 | // identical directories (hyper-tps, 2026-10-11) failed the WHOLE install on | |
| 158 | // that rejection. | |
| 154 | 159 | const A = await makePeer('tree-peer-dup-a') |
| 155 | 160 | const B = await makePeer('tree-peer-dup-b') |
| 156 | 161 | |
| 157 | await A.fileIndex.putTree('d'.repeat(64), { entries: [], size: 0 }) | |
| 162 | const entries = [{ kind: KIND_FILE, exec: false, name: 'x.png', childHash: 'e'.repeat(64), size: 3 }] | |
| 163 | const hash = computeTreeHash(entries) | |
| 164 | await A.fileIndex.putTree(hash, { entries, size: 3 }) | |
| 158 | 165 | const { pB } = connect(A, B) |
| 159 | 166 | await tick() |
| 160 | 167 | |
| 161 | const hash = 'd'.repeat(64) | |
| 162 | const p1 = B.protocol.requestTree(pB, hash, { timeout: 2000 }) | |
| 163 | let dupErr = null | |
| 164 | try { | |
| 165 | await B.protocol.requestTree(pB, hash, { timeout: 2000 }) | |
| 166 | } catch (e) { | |
| 167 | dupErr = e | |
| 168 | } | |
| 169 | t.ok(dupErr, 'duplicate request rejected') | |
| 170 | await p1 // first one should still succeed | |
| 168 | let served = 0 | |
| 169 | const orig = A.protocol._onTreeRequest.bind(A.protocol) | |
| 170 | A.protocol._onTreeRequest = (peer, msg) => { served++; return orig(peer, msg) } | |
| 171 | ||
| 172 | const [r1, r2] = await Promise.all([ | |
| 173 | B.protocol.requestTree(pB, hash, { timeout: 2000 }), | |
| 174 | B.protocol.requestTree(pB, hash, { timeout: 2000 }) | |
| 175 | ]) | |
| 176 | t.alike(r1, r2, 'both callers get the same entries') | |
| 177 | t.is(r1.length, 1) | |
| 178 | t.is(served, 1, 'one request on the wire') | |
| 171 | 179 | |
| 172 | 180 | A.protocol.destroy() |
| 173 | 181 | B.protocol.destroy() |