| … | … | @@ -86,7 +86,15 @@ export class ChunkScheduler { |
| 86 | 86 | this._chunks = null // [{ hash, offset, length }] |
| 87 | 87 | this._needed = new Set() // indices not yet written |
| 88 | 88 | this._inflight = new Map() // index → peer (one peer per chunk) |
| 89 | this._inflightAt = new Map() // index → assign timestamp (P5 hotswap staleness) | |
| 89 | 90 | this._peerInflight = new Map() // peer → count |
| 91 | // P5 — hotswap: reassign a chunk stalled on a slow peer to an idle faster peer. | |
| 92 | // A chunk inflight this long (ms) while another peer has spare capacity is | |
| 93 | // eligible to be stolen. opts.sendCancel(peer, indices) best-effort tells the | |
| 94 | // slow peer to stop serving them. <0 disables hotswap. | |
| 95 | this._hotswapStallMs = opts.hotswapStallMs ?? 2000 | |
| 96 | this._sendCancel = opts.sendCancel || null | |
| 97 | this._hotswaps = 0 | |
| 90 | 98 | this._received = new Map() // peer → chunks accepted (multi-source proof) |
| 91 | 99 | |
| 92 | 100 | this._resolve = null |
| … | … | @@ -242,6 +250,7 @@ export class ChunkScheduler { |
| 242 | 250 | // Free the inflight slot regardless — a bad chunk should be re-fetched. |
| 243 | 251 | if (this._inflight.get(index) === peer) { |
| 244 | 252 | this._inflight.delete(index) |
| 253 | this._inflightAt.delete(index) | |
| 245 | 254 | this._peerInflight.set(peer, Math.max(0, (this._peerInflight.get(peer) || 1) - 1)) |
| 246 | 255 | } |
| 247 | 256 | if (!res.ok) { |
| … | … | @@ -295,7 +304,7 @@ export class ChunkScheduler { |
| 295 | 304 | this._peers.delete(peer) |
| 296 | 305 | this._peerInflight.delete(peer) |
| 297 | 306 | for (const [index, p] of this._inflight) { |
| 298 | if (p === peer) this._inflight.delete(index) // back to needed (still in _needed) | |
| 307 | if (p === peer) { this._inflight.delete(index); this._inflightAt.delete(index) } // back to needed (still in _needed) | |
| 299 | 308 | } |
| 300 | 309 | if (this._done) return |
| 301 | 310 | if (this._peers.size === 0 && this._needed.size > 0) { |
| … | … | @@ -340,6 +349,7 @@ export class ChunkScheduler { |
| 340 | 349 | if (this._done || !this._chunks) return |
| 341 | 350 | // Per peer, batch the indices we hand it this round. |
| 342 | 351 | const batches = new Map() |
| 352 | const now = Date.now() | |
| 343 | 353 | for (const peer of this._peers) { |
| 344 | 354 | let slots = this._maxInflight(peer) - (this._peerInflight.get(peer) || 0) |
| 345 | 355 | if (slots <= 0) continue |
| … | … | @@ -347,16 +357,52 @@ export class ChunkScheduler { |
| 347 | 357 | if (slots <= 0) break |
| 348 | 358 | if (this._inflight.has(index)) continue |
| 349 | 359 | this._inflight.set(index, peer) |
| 360 | this._inflightAt.set(index, now) | |
| 350 | 361 | this._peerInflight.set(peer, (this._peerInflight.get(peer) || 0) + 1) |
| 351 | 362 | if (!batches.has(peer)) batches.set(peer, []) |
| 352 | 363 | batches.get(peer).push(index) |
| 353 | 364 | slots-- |
| 354 | 365 | } |
| 366 | // P5: this peer has spare capacity and no un-assigned chunks are left for it — | |
| 367 | // steal the most-stalled chunk(s) inflight to slower peers (the hotswap tail). | |
| 368 | if (slots > 0) this._hotswap(peer, slots, batches, now) | |
| 355 | 369 | } |
| 356 | 370 | for (const [peer, indices] of batches) { |
| 357 | 371 | if (indices.length) this._sendNeed(peer, indices) |
| 358 | 372 | } |
| 359 | 373 | } |
| 374 | ||
| 375 | // P5 — hotswap tail: reassign chunks that have been inflight to OTHER peers beyond | |
| 376 | // the stall window to `peer` (which is idle-capable), so one slow peer can't hold up | |
| 377 | // the final chunks while faster peers sit idle. Mirrors hypercore's HotswapQueue. | |
| 378 | // Duplicate arrivals (the slow peer may still deliver its copy) are handled | |
| 379 | // idempotently in onChunkData. sendCancel is best-effort bandwidth saving. | |
| 380 | _hotswap (peer, slots, batches, now) { | |
| 381 | if (this._hotswapStallMs < 0) return | |
| 382 | const cands = [] | |
| 383 | for (const [index, holder] of this._inflight) { | |
| 384 | if (holder === peer) continue | |
| 385 | if ((now - (this._inflightAt.get(index) || now)) < this._hotswapStallMs) continue | |
| 386 | cands.push(index) | |
| 387 | } | |
| 388 | if (!cands.length) return | |
| 389 | cands.sort((a, b) => (this._inflightAt.get(a) || 0) - (this._inflightAt.get(b) || 0)) // most-stalled first | |
| 390 | const cancels = new Map() | |
| 391 | for (const index of cands) { | |
| 392 | if (slots <= 0) break | |
| 393 | const holder = this._inflight.get(index) | |
| 394 | this._inflight.set(index, peer) | |
| 395 | this._inflightAt.set(index, now) | |
| 396 | this._peerInflight.set(peer, (this._peerInflight.get(peer) || 0) + 1) | |
| 397 | this._peerInflight.set(holder, Math.max(0, (this._peerInflight.get(holder) || 1) - 1)) | |
| 398 | if (!batches.has(peer)) batches.set(peer, []) | |
| 399 | batches.get(peer).push(index) | |
| 400 | this._hotswaps++ | |
| 401 | if (this._sendCancel) { if (!cancels.has(holder)) cancels.set(holder, []); cancels.get(holder).push(index) } | |
| 402 | slots-- | |
| 403 | } | |
| 404 | for (const [holder, indices] of cancels) { try { this._sendCancel(holder, indices) } catch {} } | |
| 405 | } | |
| 360 | 406 | } |
| 361 | 407 | |
| 362 | 408 | export { hashChunk } |
| … | … | @@ -237,9 +237,23 @@ export const chunkData = { |
| 237 | 237 | |
| 238 | 238 | // Message 6: chunk-cancel — cancel a pending transfer |
| 239 | 239 | export const chunkCancel = { |
| 240 | preencode (state, m) { c.string.preencode(state, m.path) }, | |
| 241 | encode (state, m) { c.string.encode(state, m.path) }, | |
| 242 | decode (state) { return { path: c.string.decode(state) } } | |
| 240 | preencode (state, m) { | |
| 241 | c.string.preencode(state, m.path) | |
| 242 | // P5 hotswap: optional per-chunk cancel. Appended so pre-0.3.0 peers still | |
| 243 | // decode {path} and ignore the trailing bytes; absent = whole-transfer cancel. | |
| 244 | if (m.indices && m.indices.length) c.array(c.uint).preencode(state, m.indices) | |
| 245 | }, | |
| 246 | encode (state, m) { | |
| 247 | c.string.encode(state, m.path) | |
| 248 | if (m.indices && m.indices.length) c.array(c.uint).encode(state, m.indices) | |
| 249 | }, | |
| 250 | decode (state) { | |
| 251 | const path = c.string.decode(state) | |
| 252 | // Keep the pre-0.3.0 {path} shape byte-for-byte when no indices trail, so a | |
| 253 | // whole-transfer cancel decodes identically on old and new peers. | |
| 254 | if (state.start < state.end) return { path, indices: c.array(c.uint).decode(state) } | |
| 255 | return { path } | |
| 256 | } | |
| 243 | 257 | } |
| 244 | 258 | |
| 245 | 259 | // Message 7: transfer-complete — confirm file fully received |
| … | … | @@ -148,6 +148,8 @@ export class OverlayProtocol { |
| 148 | 148 | destPath: opts.destPath, |
| 149 | 149 | transfer: this._transferManager, |
| 150 | 150 | sendNeed: (peer, indices) => peer.msgs.chunkNeed.send({ path: p, indices }), |
| 151 | // P5 hotswap: best-effort tell a slow peer to stop serving reassigned chunks. | |
| 152 | sendCancel: (peer, indices) => { try { peer.msgs.chunkCancel.send({ path: p, indices }) } catch {} }, | |
| 151 | 153 | cap: opts.cap, |
| 152 | 154 | timeout: opts.timeout, |
| 153 | 155 | onProgress: opts.onProgress, |
| … | … | @@ -686,6 +688,8 @@ export class OverlayProtocol { |
| 686 | 688 | for (let i = 0; i < indices.length; i++) { |
| 687 | 689 | const index = indices[i] |
| 688 | 690 | if (index >= chunkMap.length) continue |
| 691 | // P5: skip a chunk the downloader hotswapped away (one-shot). | |
| 692 | if (peer._cancelledServe && peer._cancelledServe.delete(path + '\x00' + index)) continue | |
| 689 | 693 | const c = chunkMap[index] |
| 690 | 694 | const data = this._transferManager.readChunk(diskPath, c.offset, c.length) |
| 691 | 695 | if (!data) continue |
| … | … | @@ -787,6 +791,14 @@ export class OverlayProtocol { |
| 787 | 791 | } |
| 788 | 792 | |
| 789 | 793 | _onChunkCancel (peer, msg) { |
| 794 | if (msg.indices && msg.indices.length) { | |
| 795 | // P5 hotswap: the downloader reassigned these chunk indices to a faster peer. | |
| 796 | // Stop serving just them (bandwidth save); the rest of the transfer continues. | |
| 797 | // Consumed one-shot in _serveJob. | |
| 798 | if (!peer._cancelledServe) peer._cancelledServe = new Set() | |
| 799 | for (const i of msg.indices) peer._cancelledServe.add(msg.path + '\x00' + i) | |
| 800 | return | |
| 801 | } | |
| 790 | 802 | const diskPath = this._filePaths.get(msg.path) |
| 791 | 803 | if (diskPath) this._transferManager.cancel(diskPath) |
| 792 | 804 | } |
| … | … | @@ -4,7 +4,19 @@ |
| 4 | 4 | // udx → clamp(round(min * (rtt<=50 ? 1 : rtt/50 * min(1, 2/peers))), min, max) |
| 5 | 5 | // with our tuned range INFLIGHT_RANGE = [4, 64]. |
| 6 | 6 | import test from 'brittle' |
| 7 | import c from 'compact-encoding' | |
| 8 | import b4a from 'b4a' | |
| 7 | 9 | import { ChunkScheduler } from '../lib/chunk-scheduler.js' |
| 10 | import * as messages from '../lib/messages.js' | |
| 11 | ||
| 12 | function roundtrip (codec, value) { | |
| 13 | const state = { start: 0, end: 0, buffer: null } | |
| 14 | codec.preencode(state, value) | |
| 15 | state.buffer = b4a.alloc(state.end) | |
| 16 | codec.encode(state, value) | |
| 17 | state.start = 0 | |
| 18 | return codec.decode(state) | |
| 19 | } | |
| 8 | 20 | |
| 9 | 21 | function mkSched (opts = {}) { |
| 10 | 22 | const s = new ChunkScheduler({ path: 'content:x', destPath: '/tmp/none', transfer: {}, sendNeed: () => {}, ...opts }) |
| … | … | @@ -43,3 +55,57 @@ test('P2: opts.cap overrides the dynamic window (back-compat)', (t) => { |
| 43 | 55 | t.is(s._maxInflight(udx(99999)), 3, 'fixed cap wins over RTT scaling') |
| 44 | 56 | t.is(s._maxInflight(plain()), 3, 'fixed cap wins over the no-udx default') |
| 45 | 57 | }) |
| 58 | ||
| 59 | test('P5 hotswap: an idle peer steals a chunk stalled on a slow peer', (t) => { | |
| 60 | const needs = [] | |
| 61 | const cancels = [] | |
| 62 | const s = mkSched({ | |
| 63 | cap: 4, // fixed window so both peers have spare capacity deterministically | |
| 64 | hotswapStallMs: 50, | |
| 65 | sendNeed: (peer, idx) => needs.push([peer, idx]), | |
| 66 | sendCancel: (peer, idx) => cancels.push([peer, idx]) | |
| 67 | }) | |
| 68 | s._chunks = [{ hash: 'h0', offset: 0, length: 10 }] | |
| 69 | const slow = udx(10) | |
| 70 | const fast = udx(10) | |
| 71 | s._peers.add(slow) // iterated first (holds nothing to steal), then fast steals | |
| 72 | s._peers.add(fast) | |
| 73 | // chunk 0 is inflight to the slow peer and has been for a while (stalled). | |
| 74 | s._inflight.set(0, slow) | |
| 75 | s._inflightAt.set(0, Date.now() - 1000) | |
| 76 | s._peerInflight.set(slow, 1) | |
| 77 | s._peerInflight.set(fast, 0) | |
| 78 | // _needed is empty — nothing fresh to fetch, only the stalled chunk remains. | |
| 79 | ||
| 80 | s._assign() | |
| 81 | ||
| 82 | t.is(s._inflight.get(0), fast, 'chunk 0 reassigned from slow → fast') | |
| 83 | t.ok(needs.some(([p, idx]) => p === fast && idx.includes(0)), 'fast peer re-requested chunk 0') | |
| 84 | t.ok(cancels.some(([p, idx]) => p === slow && idx.includes(0)), 'slow peer told to cancel chunk 0') | |
| 85 | t.is(s._hotswaps, 1, 'one hotswap recorded') | |
| 86 | t.is(s._peerInflight.get(slow), 0, 'slow peer inflight decremented') | |
| 87 | t.is(s._peerInflight.get(fast), 1, 'fast peer inflight incremented') | |
| 88 | }) | |
| 89 | ||
| 90 | test('P5 hotswap: fresh chunks are assigned before any stealing', (t) => { | |
| 91 | const needs = [] | |
| 92 | const s = mkSched({ cap: 4, hotswapStallMs: 50, sendNeed: (peer, idx) => needs.push([peer, idx]) }) | |
| 93 | s._chunks = [0, 1].map((i) => ({ hash: 'h' + i, offset: i * 10, length: 10 })) | |
| 94 | const a = udx(10) | |
| 95 | s._peers.add(a) | |
| 96 | s._needed.add(1) // chunk 1 is fresh/needed | |
| 97 | s._inflight.set(0, udx(10)) // chunk 0 inflight to some other (already-gone) peer, stalled | |
| 98 | s._inflightAt.set(0, Date.now() - 1000) | |
| 99 | s._assign() | |
| 100 | // peer a should take the FRESH chunk 1 first (not immediately hotswap chunk 0) | |
| 101 | t.ok(s._inflight.get(1) === a, 'fresh chunk 1 assigned to peer a') | |
| 102 | }) | |
| 103 | ||
| 104 | test('P5 wire: chunkCancel round-trips with indices, back-compat without', (t) => { | |
| 105 | const withIdx = roundtrip(messages.chunkCancel, { path: 'content:abc', indices: [3, 7, 9] }) | |
| 106 | t.is(withIdx.path, 'content:abc') | |
| 107 | t.alike(withIdx.indices, [3, 7, 9], 'per-chunk indices preserved') | |
| 108 | const noIdx = roundtrip(messages.chunkCancel, { path: 'content:abc' }) | |
| 109 | t.is(noIdx.path, 'content:abc') | |
| 110 | t.absent(noIdx.indices, 'absent indices → bare {path} (whole-transfer cancel; pre-0.3.0 shape)') | |
| 111 | }) |