| … | … | @@ -60,6 +60,10 @@ const CONTROL_STOPPED = 1 |
| 60 | 60 | // Abandon a serve loop whose stream neither drains nor closes within |
| 61 | 61 | // this window (a wedged-but-TCP-alive peer); the receiver re-requests, so it self-heals. |
| 62 | 62 | const DRAIN_TIMEOUT_MS = 60000 |
| 63 | // Max bytes per corked chunkData batch in _serveJob: flush, check real | |
| 64 | // backpressure, park on drain if the stream is backed up. Bounds a serve job's | |
| 65 | // memory to ~one batch instead of the whole request (see _serveJob). | |
| 66 | const SERVE_BATCH_BYTES = 1024 * 1024 | |
| 63 | 67 | |
| 64 | 68 | export class OverlayProtocol { |
| 65 | 69 | constructor (syncEngine, transferManager, opts = {}) { |
| … | … | @@ -744,6 +748,12 @@ export class OverlayProtocol { |
| 744 | 748 | const cork = () => { if (mux && mux.cork && !corked) { mux.cork(); corked = true } } |
| 745 | 749 | const uncork = () => { if (mux && mux.uncork && corked) { mux.uncork(); corked = false } } |
| 746 | 750 | cork() |
| 751 | // Bytes queued in the current corked batch. protomux's send() returns TRUE | |
| 752 | // while corked (it only appends to the batch), so `flushed === false` never | |
| 753 | // fires inside a corked run — before this cap a job read EVERY requested | |
| 754 | // chunk into memory before the first flush (stortron 2026-10-11: one | |
| 755 | // chunkNeed for a ~572 MB .pck took the sidecar 1.4 -> 4 GB RSS in 5 s). | |
| 756 | let batchBytes = 0 | |
| 747 | 757 | try { |
| 748 | 758 | for (let i = 0; i < indices.length; i++) { |
| 749 | 759 | const index = indices[i] |
| … | … | @@ -757,8 +767,18 @@ export class OverlayProtocol { |
| 757 | 767 | if (this._chunkServeCb) { |
| 758 | 768 | try { this._chunkServeCb({ path, index, bytes: data.length, peer, from: peer.authorizedServe.get(path) || null }) } catch {} |
| 759 | 769 | } |
| 760 | if (flushed === false && i < indices.length - 1) { | |
| 770 | batchBytes += data.length | |
| 771 | if (i === indices.length - 1) continue | |
| 772 | let backedUp = flushed === false | |
| 773 | if (!backedUp && corked && batchBytes >= SERVE_BATCH_BYTES) { | |
| 774 | uncork() // flush the full batch; the mux's drained now reflects the real write | |
| 775 | batchBytes = 0 | |
| 776 | backedUp = !!mux && mux.drained === false | |
| 777 | if (!backedUp) cork() | |
| 778 | } | |
| 779 | if (backedUp) { | |
| 761 | 780 | uncork() // flush the corked batch before we park on drain |
| 781 | batchBytes = 0 | |
| 762 | 782 | const alive = await this._waitForDrain(peer) |
| 763 | 783 | if (!alive) return |
| 764 | 784 | if (this._serveAuthorizer && !peer.authorizedServe.has(path)) return |
| … | … | @@ -1,6 +1,6 @@ |
| 1 | 1 | { |
| 2 | 2 | "name": "hyper-overlay", |
| 3 | "version": "0.3.15", | |
| 3 | "version": "0.3.16", | |
| 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", |
| … | … | @@ -148,3 +148,42 @@ test('P3 cork-batching: cork/uncork stay balanced across a paced serve', async ( |
| 148 | 148 | t.is(corkDepth, 0, 'P3: cork/uncork balanced — no leaked cork after the serve') |
| 149 | 149 | t.ok(maxDepth === 1, 'P3: exactly one cork depth (batch corked, flushed before each drain)') |
| 150 | 150 | }) |
| 151 | ||
| 152 | test('corked serve is bounded: a backed-up stream pauses the job after ~one batch', async (t) => { | |
| 153 | // protomux returns TRUE from send() while corked (it only appends to the | |
| 154 | // batch), so `flushed === false` never fired inside a corked run: a job read | |
| 155 | // EVERY requested chunk into memory before the first flush. stortron | |
| 156 | // 2026-10-11: one chunkNeed for a ~572 MB .pck went 1.4 -> 4 GB RSS in 5 s | |
| 157 | // (readChunk copy + frame copy + batch concat + encrypt + udx). | |
| 158 | const CH = 256 * 1024 | |
| 159 | const N = 32 // 8 MiB requested | |
| 160 | const chunkMap = Array.from({ length: N }, (_, i) => ({ hash: 'h' + i, offset: i * CH, length: CH })) | |
| 161 | const proto = new OverlayProtocol(null, { | |
| 162 | _fileIndex: { getChunkMap: async () => chunkMap, getChunkMapByHash: async () => chunkMap }, | |
| 163 | readChunk: (p, off, len) => b4a.alloc(len) | |
| 164 | }, { filePaths: new Map([['content:x', '/fake']]) }) | |
| 165 | const stream = makeStream() | |
| 166 | const sent = [] | |
| 167 | let corked = 0 | |
| 168 | let drained = true | |
| 169 | const mux = { | |
| 170 | stream, | |
| 171 | get drained () { return drained }, | |
| 172 | cork () { corked++ }, | |
| 173 | uncork () { if (--corked === 0) drained = false } // flushing a batch fills the write buffer | |
| 174 | } | |
| 175 | const peer = { | |
| 176 | channel: { closed: false, get drained () { return drained } }, | |
| 177 | mux, | |
| 178 | msgs: { chunkData: { send: (m) => { sent.push(m.index); if (corked) return true; drained = false; return false } } } | |
| 179 | } | |
| 180 | ||
| 181 | proto._onChunkNeed(peer, { path: 'content:x', indices: chunkMap.map((_, i) => i) }) | |
| 182 | await tick() | |
| 183 | t.ok(sent.length * CH <= 2 * 1024 * 1024, 'paused after ~one batch, not the whole job (sent ' + sent.length + '/' + N + ' before any drain)') | |
| 184 | ||
| 185 | let guard = 0 | |
| 186 | while (sent.length < N && guard++ < 200) { drained = true; stream.emitDrain(); await tick() } | |
| 187 | t.is(sent.length, N, 'all chunks still served') | |
| 188 | t.is(corked, 0, 'cork/uncork balanced') | |
| 189 | }) |