// Backpressure regression test for the chunk SEND path (_onChunkNeed).
//
// The multi-source (ChunkScheduler) rewrite regressed bulk transfer: the
// seeder fired every chunkData.send() in a tight synchronous loop, ignoring
// send()'s drained return. On a real link that floods secret-stream's write
// buffer (unbounded backlog) → stall / ETIMEDOUT / ECONNRESET on large
// transfers (the 185MB+ bootstrap-tarball fetch that failed in prod). The
// pre-rewrite path honored backpressure (proven at 200MB/1.30GB, no stall).
//
// The fix serializes sends per peer through _pumpSendQueue, awaiting 'drain'
// whenever send() reports backpressure. These tests assert:
// 1. under backpressure, the seeder paces sends by drain (does NOT flood);
// 2. when NOT backpressured, all chunks still send immediately (no
// throughput regression);
// 3. it composes with concurrent chunkNeed batches (one pump per peer).
import test from 'brittle'
import b4a from 'b4a'
import { OverlayProtocol } from '../lib/protocol.js'
// Minimal drain-controllable stream stub (only what _awaitDrain uses).
// Minimal drain-controllable stream stub matching the engine's _waitForDrain,
// which registers via stream.on('drain'|'close') + removeListener (not once).
function makeStream () {
const listeners = { drain: [], close: [] }
return {
on (ev, cb) { (listeners[ev] || (listeners[ev] = [])).push(cb) },
removeListener (ev, cb) { const a = listeners[ev]; if (!a) return; const i = a.indexOf(cb); if (i >= 0) a.splice(i, 1) },
emitDrain () { const fire = listeners.drain.splice(0); for (const cb of fire) cb() },
emitClose () { const fire = listeners.close.splice(0); for (const cb of fire) cb() }
}
}
const tick = () => new Promise((r) => setTimeout(r, 0))
function makeProto (n) {
const chunkMap = Array.from({ length: n }, (_, i) => ({ hash: 'h' + i, offset: i * 10, length: 10 }))
const transferManager = {
// The engine resolves a 'content:<hash>' path via getChunkMapByHash
// (chunkmap-oid: bee key), and a plain overlay path via getChunkMap. The
// test serves 'content:x', so the mock must expose both.
_fileIndex: {
getChunkMap: async () => chunkMap,
getChunkMapByHash: async () => chunkMap
},
readChunk: (p, off, len) => b4a.alloc(len)
}
return new OverlayProtocol(null, transferManager, { filePaths: new Map([['content:x', '/fake']]) })
}
test('backpressure: seeder paces chunkData by drain instead of flooding', async (t) => {
const N = 6
const proto = makeProto(N)
const stream = makeStream()
const sent = []
let bufferFull = false
const peer = {
channel: { closed: false, get drained () { return !bufferFull } },
mux: { stream },
// every send fills the buffer and reports backpressure (returns false)
msgs: { chunkData: { send: (m) => { sent.push(m.index); bufferFull = true; return false } } }
}
await proto._onChunkNeed(peer, { path: 'content:x', indices: [0, 1, 2, 3, 4, 5] })
await tick()
t.is(sent.length, 1, 'only the first chunk is sent, then the pump waits for drain (no flood)')
for (let expect = 2; expect <= N; expect++) {
bufferFull = false
stream.emitDrain()
await tick()
t.is(sent.length, expect, `drain released exactly one more chunk (${expect} total)`)
}
t.alike(sent, [0, 1, 2, 3, 4, 5], 'all chunks delivered in order, paced by backpressure')
})
test('fast path: when the stream is drained, all chunks send without waiting', async (t) => {
const N = 5
const proto = makeProto(N)
const stream = makeStream()
const sent = []
const peer = {
channel: { closed: false },
mux: { drained: true, stream },
msgs: { chunkData: { send: (m) => { sent.push(m.index); return true } } } // never backpressured
}
await proto._onChunkNeed(peer, { path: 'content:x', indices: [0, 1, 2, 3, 4] })
await tick()
t.is(sent.length, N, 'all chunks sent immediately when not backpressured (no throughput regression)')
})
test('concurrent chunkNeed batches: one pump per peer, all delivered exactly once', async (t) => {
const N = 8
const proto = makeProto(N)
const stream = makeStream()
const sent = []
let bufferFull = false
const peer = {
channel: { closed: false, get drained () { return !bufferFull } },
mux: { stream },
msgs: { chunkData: { send: (m) => { sent.push(m.index); bufferFull = true; return false } } }
}
// Two overlapping batches (as the scheduler pipelines chunkNeed).
await proto._onChunkNeed(peer, { path: 'content:x', indices: [0, 1, 2, 3] })
await proto._onChunkNeed(peer, { path: 'content:x', indices: [4, 5, 6, 7] })
await tick()
// SERIALIZATION PROOF: a single paced pump has sent exactly ONE chunk and is
// awaiting drain. If the second chunkNeed had started a second concurrent
// pump (the re-flood bug), TWO chunks would already be out.
t.is(sent.length, 1, 'concurrent chunkNeed did NOT start a second pump (only 1 chunk out, paced)')
// Drain until everything is out.
let guard = 0
while (sent.length < N && guard++ < 50) { bufferFull = false; stream.emitDrain(); await tick() }
t.is(sent.length, N, 'all 8 chunks delivered across the two batches')
t.alike([...sent].sort((a, b) => a - b), [0, 1, 2, 3, 4, 5, 6, 7], 'each chunk delivered exactly once')
})
test('P3 cork-batching: cork/uncork stay balanced across a paced serve', async (t) => {
const N = 5
const proto = makeProto(N)
const stream = makeStream()
const sent = []
let bufferFull = false
let corkDepth = 0
let maxDepth = 0
const peer = {
channel: { closed: false, get drained () { return !bufferFull } },
mux: {
stream,
cork () { corkDepth++; if (corkDepth > maxDepth) maxDepth = corkDepth },
uncork () { corkDepth-- }
},
msgs: { chunkData: { send: (m) => { sent.push(m.index); bufferFull = true; return false } } }
}
proto._onChunkNeed(peer, { path: 'content:x', indices: [0, 1, 2, 3, 4] })
await tick()
let guard = 0
while (sent.length < N && guard++ < 50) { bufferFull = false; stream.emitDrain(); await tick() }
t.is(sent.length, N, 'all chunks served')
t.is(corkDepth, 0, 'P3: cork/uncork balanced — no leaked cork after the serve')
t.ok(maxDepth === 1, 'P3: exactly one cork depth (batch corked, flushed before each drain)')
})
test('corked serve is bounded: a backed-up stream pauses the job after ~one batch', async (t) => {
// protomux returns TRUE from send() while corked (it only appends to the
// batch), so `flushed === false` never fired inside a corked run: a job read
// EVERY requested chunk into memory before the first flush. stortron
// 2026-10-11: one chunkNeed for a ~572 MB .pck went 1.4 -> 4 GB RSS in 5 s
// (readChunk copy + frame copy + batch concat + encrypt + udx).
const CH = 256 * 1024
const N = 32 // 8 MiB requested
const chunkMap = Array.from({ length: N }, (_, i) => ({ hash: 'h' + i, offset: i * CH, length: CH }))
const proto = new OverlayProtocol(null, {
_fileIndex: { getChunkMap: async () => chunkMap, getChunkMapByHash: async () => chunkMap },
readChunk: (p, off, len) => b4a.alloc(len)
}, { filePaths: new Map([['content:x', '/fake']]) })
const stream = makeStream()
const sent = []
let corked = 0
let drained = true
const mux = {
stream,
get drained () { return drained },
cork () { corked++ },
uncork () { if (--corked === 0) drained = false } // flushing a batch fills the write buffer
}
const peer = {
channel: { closed: false, get drained () { return drained } },
mux,
msgs: { chunkData: { send: (m) => { sent.push(m.index); if (corked) return true; drained = false; return false } } }
}
proto._onChunkNeed(peer, { path: 'content:x', indices: chunkMap.map((_, i) => i) })
await tick()
t.ok(sent.length * CH <= 2 * 1024 * 1024, 'paused after ~one batch, not the whole job (sent ' + sent.length + '/' + N + ' before any drain)')
let guard = 0
while (sent.length < N && guard++ < 200) { drained = true; stream.emitDrain(); await tick() }
t.is(sent.length, N, 'all chunks still served')
t.is(corked, 0, 'cork/uncork balanced')
})