OverviewFilesCommitsIssuesDocsReleasesPeers

hyper-overlay commits / 23488ed

port(0.3.0): backpressure — detached single-flight pump + serve-lookup guard
Two fixes the full chunk-backpressure suite surfaced (my earlier inline P1 only passed a real-stream smoke; the unit test with a mock stream caught both): 1. DETACHED PUMP: _onChunkNeed awaited the whole serve loop inline, so the onmessage handler stayed pending for the entire multi-chunk transfer (under backpressure, the whole transfer). Restored our 0.2.13 contract: enqueue + kick a fire-and-forget _drainServeQueue, return promptly. Single-flight guard (_serving) unchanged. Ok mirrors hypercore replicator receiverBusy/receiverQueue. 2. SERVE-LOOKUP GUARD: the getChunkMapByHash/getChunkMap await sat outside any try; onmessage dispatch is fire-and-forget, so a rejecting index lookup became an unhandled rejection tearing down the whole Noise/replication stream (S1 hardening family). Wrapped it — drop the request, receiver re-requests, self-heals. test/chunk-backpressure: adapted the mock stream to the engine _waitForDrain interface (stream.on/removeListener + channel.drained, not once/mux.drained); pacing assertions unchanged. 3/3. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
23488edaac1044790b8dc383737046a6e48d8520
by plankatron <fleeky@gmail.com> · 2026-07-11T17:54:45.000Z

Changed files (2) +43 -12

lib/protocol.js +25 -3
……@@ -596,9 +596,19 @@ export class OverlayProtocol {
596596 if (!diskPath) return
597597
598598 const fileIndex = this._transferManager._fileIndex
599 const chunkMap = msg.path.startsWith('content:')
600 ? await fileIndex.getChunkMapByHash(msg.path.slice('content:'.length))
601 : await fileIndex.getChunkMap(msg.path)
599 let chunkMap
600 try {
601 chunkMap = msg.path.startsWith('content:')
602 ? await fileIndex.getChunkMapByHash(msg.path.slice('content:'.length))
603 : await fileIndex.getChunkMap(msg.path)
604 } catch {
605 // A chunk-map lookup that rejects (index I/O error) must not escape: the
606 // onmessage dispatch is fire-and-forget, so a throw here becomes an
607 // unhandled rejection that tears down the whole Noise/replication stream
608 // (the S1 malformed-frame hardening theme). Drop the request — the receiver
609 // re-requests and the transfer self-heals.
610 return
611 }
602612 if (!chunkMap) return
603613
604614 // Single-flight serve pump per peer (mirrors hypercore's `receiverBusy`).
……@@ -612,6 +622,18 @@ export class OverlayProtocol {
612622 peer._serveQueue.push({ path: msg.path, diskPath, chunkMap, indices: msg.indices })
613623 if (peer._serving) return
614624 peer._serving = true
625 // Kick the pump fire-and-forget: _onChunkNeed is the onmessage dispatch and
626 // must return promptly, NOT stay pending for the whole multi-chunk serve
627 // (which, under backpressure, spans the entire transfer). A second chunkNeed
628 // from this peer just enqueues above; the single-flight guard (_serving) means
629 // this already-running pump picks it up. This is our 0.2.13 contract — Ok's
630 // inline drain loop awaited the whole serve, which the unit test rejects.
631 this._drainServeQueue(peer)
632 }
633
634 // The per-peer single-flight serve pump. Drains _serveQueue one job at a time;
635 // detached from _onChunkNeed so the message handler returns immediately.
636 async _drainServeQueue (peer) {
615637 try {
616638 while (peer._serveQueue.length) {
617639 if (peer.channel && peer.channel.closed) { peer._serveQueue.length = 0; break }
test/chunk-backpressure.test.js +18 -9
……@@ -19,12 +19,15 @@ import b4a from 'b4a'
1919import { OverlayProtocol } from '../lib/protocol.js'
2020
2121// Minimal drain-controllable stream stub (only what _awaitDrain uses).
22// Minimal drain-controllable stream stub matching the engine's _waitForDrain,
23// which registers via stream.on('drain'|'close') + removeListener (not once).
2224function makeStream () {
23 const listeners = []
25 const listeners = { drain: [], close: [] }
2426 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() }
27 on (ev, cb) { (listeners[ev] || (listeners[ev] = [])).push(cb) },
28 removeListener (ev, cb) { const a = listeners[ev]; if (!a) return; const i = a.indexOf(cb); if (i >= 0) a.splice(i, 1) },
29 emitDrain () { const fire = listeners.drain.splice(0); for (const cb of fire) cb() },
30 emitClose () { const fire = listeners.close.splice(0); for (const cb of fire) cb() }
2831 }
2932}
3033const tick = () => new Promise((r) => setTimeout(r, 0))
……@@ -32,7 +35,13 @@ const tick = () => new Promise((r) => setTimeout(r, 0))
3235function makeProto (n) {
3336 const chunkMap = Array.from({ length: n }, (_, i) => ({ hash: 'h' + i, offset: i * 10, length: 10 }))
3437 const transferManager = {
35 _fileIndex: { getChunkMap: async () => chunkMap },
38 // The engine resolves a 'content:<hash>' path via getChunkMapByHash
39 // (chunkmap-oid: bee key), and a plain overlay path via getChunkMap. The
40 // test serves 'content:x', so the mock must expose both.
41 _fileIndex: {
42 getChunkMap: async () => chunkMap,
43 getChunkMapByHash: async () => chunkMap
44 },
3645 readChunk: (p, off, len) => b4a.alloc(len)
3746 }
3847 return new OverlayProtocol(null, transferManager, { filePaths: new Map([['content:x', '/fake']]) })
……@@ -45,8 +54,8 @@ test('backpressure: seeder paces chunkData by drain instead of flooding', async
4554 const sent = []
4655 let bufferFull = false
4756 const peer = {
48 channel: { closed: false },
49 mux: { get drained () { return !bufferFull }, stream },
57 channel: { closed: false, get drained () { return !bufferFull } },
58 mux: { stream },
5059 // every send fills the buffer and reports backpressure (returns false)
5160 msgs: { chunkData: { send: (m) => { sent.push(m.index); bufferFull = true; return false } } }
5261 }
……@@ -89,8 +98,8 @@ test('concurrent chunkNeed batches: one pump per peer, all delivered exactly onc
8998 const sent = []
9099 let bufferFull = false
91100 const peer = {
92 channel: { closed: false },
93 mux: { get drained () { return !bufferFull }, stream },
101 channel: { closed: false, get drained () { return !bufferFull } },
102 mux: { stream },
94103 msgs: { chunkData: { send: (m) => { sent.push(m.index); bufferFull = true; return false } } }
95104 }
96105