| … | … | @@ -0,0 +1,113 @@ |
| 1 | // Backpressure regression test for the chunk SEND path (_onChunkNeed). |
| 2 | // |
| 3 | // The multi-source (ChunkScheduler) rewrite regressed bulk transfer: the |
| 4 | // seeder fired every chunkData.send() in a tight synchronous loop, ignoring |
| 5 | // send()'s drained return. On a real link that floods secret-stream's write |
| 6 | // buffer (unbounded backlog) → stall / ETIMEDOUT / ECONNRESET on large |
| 7 | // transfers (the 185MB+ bootstrap-tarball fetch that failed in prod). The |
| 8 | // pre-rewrite path honored backpressure (proven at 200MB/1.30GB, no stall). |
| 9 | // |
| 10 | // The fix serializes sends per peer through _pumpSendQueue, awaiting 'drain' |
| 11 | // whenever send() reports backpressure. These tests assert: |
| 12 | // 1. under backpressure, the seeder paces sends by drain (does NOT flood); |
| 13 | // 2. when NOT backpressured, all chunks still send immediately (no |
| 14 | // throughput regression); |
| 15 | // 3. it composes with concurrent chunkNeed batches (one pump per peer). |
| 16 | |
| 17 | import test from 'brittle' |
| 18 | import b4a from 'b4a' |
| 19 | import { OverlayProtocolV2 } from '../lib/protocol-v2.js' |
| 20 | |
| 21 | // Minimal drain-controllable stream stub (only what _awaitDrain uses). |
| 22 | function makeStream () { |
| 23 | const listeners = [] |
| 24 | return { |
| 25 | once (ev, cb) { if (ev === 'drain') listeners.push(cb) }, |
| 26 | removeListener (ev, cb) { const i = listeners.indexOf(cb); if (i >= 0) listeners.splice(i, 1) }, |
| 27 | emitDrain () { const fire = listeners.splice(0); for (const cb of fire) cb() } |
| 28 | } |
| 29 | } |
| 30 | const tick = () => new Promise((r) => setTimeout(r, 0)) |
| 31 | |
| 32 | function makeProto (n) { |
| 33 | const chunkMap = Array.from({ length: n }, (_, i) => ({ hash: 'h' + i, offset: i * 10, length: 10 })) |
| 34 | const transferManager = { |
| 35 | _fileIndex: { getChunkMap: async () => chunkMap }, |
| 36 | readChunk: (p, off, len) => b4a.alloc(len) |
| 37 | } |
| 38 | return new OverlayProtocolV2(null, transferManager, { filePaths: new Map([['content:x', '/fake']]) }) |
| 39 | } |
| 40 | |
| 41 | test('backpressure: seeder paces chunkData by drain instead of flooding', async (t) => { |
| 42 | const N = 6 |
| 43 | const proto = makeProto(N) |
| 44 | const stream = makeStream() |
| 45 | const sent = [] |
| 46 | let bufferFull = false |
| 47 | const peer = { |
| 48 | channel: { closed: false }, |
| 49 | mux: { get drained () { return !bufferFull }, stream }, |
| 50 | // every send fills the buffer and reports backpressure (returns false) |
| 51 | msgs: { chunkData: { send: (m) => { sent.push(m.index); bufferFull = true; return false } } } |
| 52 | } |
| 53 | |
| 54 | await proto._onChunkNeed(peer, { path: 'content:x', indices: [0, 1, 2, 3, 4, 5] }) |
| 55 | await tick() |
| 56 | |
| 57 | t.is(sent.length, 1, 'only the first chunk is sent, then the pump waits for drain (no flood)') |
| 58 | |
| 59 | for (let expect = 2; expect <= N; expect++) { |
| 60 | bufferFull = false |
| 61 | stream.emitDrain() |
| 62 | await tick() |
| 63 | t.is(sent.length, expect, `drain released exactly one more chunk (${expect} total)`) |
| 64 | } |
| 65 | t.alike(sent, [0, 1, 2, 3, 4, 5], 'all chunks delivered in order, paced by backpressure') |
| 66 | }) |
| 67 | |
| 68 | test('fast path: when the stream is drained, all chunks send without waiting', async (t) => { |
| 69 | const N = 5 |
| 70 | const proto = makeProto(N) |
| 71 | const stream = makeStream() |
| 72 | const sent = [] |
| 73 | const peer = { |
| 74 | channel: { closed: false }, |
| 75 | mux: { drained: true, stream }, |
| 76 | msgs: { chunkData: { send: (m) => { sent.push(m.index); return true } } } // never backpressured |
| 77 | } |
| 78 | |
| 79 | await proto._onChunkNeed(peer, { path: 'content:x', indices: [0, 1, 2, 3, 4] }) |
| 80 | await tick() |
| 81 | |
| 82 | t.is(sent.length, N, 'all chunks sent immediately when not backpressured (no throughput regression)') |
| 83 | }) |
| 84 | |
| 85 | test('concurrent chunkNeed batches: one pump per peer, all delivered exactly once', async (t) => { |
| 86 | const N = 8 |
| 87 | const proto = makeProto(N) |
| 88 | const stream = makeStream() |
| 89 | const sent = [] |
| 90 | let bufferFull = false |
| 91 | const peer = { |
| 92 | channel: { closed: false }, |
| 93 | mux: { get drained () { return !bufferFull }, stream }, |
| 94 | msgs: { chunkData: { send: (m) => { sent.push(m.index); bufferFull = true; return false } } } |
| 95 | } |
| 96 | |
| 97 | // Two overlapping batches (as the scheduler pipelines chunkNeed). |
| 98 | await proto._onChunkNeed(peer, { path: 'content:x', indices: [0, 1, 2, 3] }) |
| 99 | await proto._onChunkNeed(peer, { path: 'content:x', indices: [4, 5, 6, 7] }) |
| 100 | await tick() |
| 101 | |
| 102 | // SERIALIZATION PROOF: a single paced pump has sent exactly ONE chunk and is |
| 103 | // awaiting drain. If the second chunkNeed had started a second concurrent |
| 104 | // pump (the re-flood bug), TWO chunks would already be out. |
| 105 | t.is(sent.length, 1, 'concurrent chunkNeed did NOT start a second pump (only 1 chunk out, paced)') |
| 106 | |
| 107 | // Drain until everything is out. |
| 108 | let guard = 0 |
| 109 | while (sent.length < N && guard++ < 50) { bufferFull = false; stream.emitDrain(); await tick() } |
| 110 | |
| 111 | t.is(sent.length, N, 'all 8 chunks delivered across the two batches') |
| 112 | t.alike([...sent].sort((a, b) => a - b), [0, 1, 2, 3, 4, 5, 6, 7], 'each chunk delivered exactly once') |
| 113 | }) |