| … | … | @@ -12,11 +12,11 @@ |
| 12 | 12 | * 4. Sender chunks on demand → streams chunks → receiver writes to disk |
| 13 | 13 | */ |
| 14 | 14 | |
| 15 | | import path from '#path' |
| 15 | import path from 'bare-path' |
| 16 | 16 | import * as messages from './messages.js' |
| 17 | 17 | import { hashChunk, selectTier } from './chunker.js' |
| 18 | 18 | import { ChunkScheduler } from './chunk-scheduler.js' |
| 19 | | import fs from '#fs' |
| 19 | import fs from 'bare-fs' |
| 20 | 20 | |
| 21 | 21 | const PROTOCOL = 'hyper-overlay/v2' |
| 22 | 22 | const VERSION = 2 |
| … | … | @@ -601,26 +601,49 @@ export class OverlayProtocol { |
| 601 | 601 | : await fileIndex.getChunkMap(msg.path) |
| 602 | 602 | if (!chunkMap) return |
| 603 | 603 | |
| 604 | | for (let i = 0; i < msg.indices.length; i++) { |
| 605 | | const index = msg.indices[i] |
| 604 | // Single-flight serve pump per peer (mirrors hypercore's `receiverBusy`). |
| 605 | // Enqueue this request; if no pump is running for this peer, drain the queue |
| 606 | // one job at a time. protomux dispatches every frame of a socket read |
| 607 | // synchronously, so two chunkNeed frames from the same peer would otherwise |
| 608 | // start rival serve loops — interleaving sends and admitting N chunks per |
| 609 | // drain instead of honouring backpressure once (the discipline our 0.2.13 |
| 610 | // serialized pump had, which Ok's inline loop dropped). |
| 611 | if (!peer._serveQueue) peer._serveQueue = [] |
| 612 | peer._serveQueue.push({ path: msg.path, diskPath, chunkMap, indices: msg.indices }) |
| 613 | if (peer._serving) return |
| 614 | peer._serving = true |
| 615 | try { |
| 616 | while (peer._serveQueue.length) { |
| 617 | if (peer.channel && peer.channel.closed) { peer._serveQueue.length = 0; break } |
| 618 | try { await this._serveJob(peer, peer._serveQueue.shift()) } catch {} |
| 619 | } |
| 620 | } finally { |
| 621 | peer._serving = false |
| 622 | } |
| 623 | } |
| 624 | |
| 625 | // Serve one chunkNeed job: stream the requested chunk indices, honouring |
| 626 | // stream backpressure — stop producing when send() reports the write buffer is |
| 627 | // full and resume on drain, so sibling protocols multiplexed on the same Noise |
| 628 | // stream (Corestore replication, app handshakes; protomux MAX_BACKLOG is |
| 629 | // Infinity) aren't starved. Runs under the per-peer single-flight guard in |
| 630 | // _onChunkNeed, so exactly one job streams at a time per peer. |
| 631 | async _serveJob (peer, job) { |
| 632 | const { path, diskPath, chunkMap, indices } = job |
| 633 | for (let i = 0; i < indices.length; i++) { |
| 634 | const index = indices[i] |
| 606 | 635 | if (index >= chunkMap.length) continue |
| 607 | 636 | const c = chunkMap[index] |
| 608 | 637 | const data = this._transferManager.readChunk(diskPath, c.offset, c.length) |
| 609 | 638 | if (!data) continue |
| 610 | | const flushed = peer.msgs.chunkData.send({ path: msg.path, index, data }) |
| 639 | const flushed = peer.msgs.chunkData.send({ path, index, data }) |
| 611 | 640 | if (this._chunkServeCb) { |
| 612 | | try { this._chunkServeCb({ path: msg.path, index, bytes: data.length, peer, from: peer.authorizedServe.get(msg.path) || null }) } catch {} |
| 641 | try { this._chunkServeCb({ path, index, bytes: data.length, peer, from: peer.authorizedServe.get(path) || null }) } catch {} |
| 613 | 642 | } |
| 614 | | // Head-of-line blocking — chunkData and every other protocol the host |
| 615 | | // multiplexes over the SAME Noise stream (Corestore replication, app |
| 616 | | // handshakes) share one socket, and protomux's MAX_BACKLOG is Infinity. |
| 617 | | // send() returns the stream's writable state; on backpressure, stop |
| 618 | | // producing content and let the queue drain, else sibling protocols on |
| 619 | | // the stream are starved for the rest of the download. |
| 620 | | if (flushed === false && i < msg.indices.length - 1) { |
| 643 | if (flushed === false && i < indices.length - 1) { |
| 621 | 644 | const alive = await this._waitForDrain(peer) |
| 622 | 645 | if (!alive) return |
| 623 | | if (this._serveAuthorizer && !peer.authorizedServe.has(msg.path)) return |
| 646 | if (this._serveAuthorizer && !peer.authorizedServe.has(path)) return |
| 624 | 647 | } |
| 625 | 648 | } |
| 626 | 649 | } |