| … | … | @@ -0,0 +1,102 @@ |
| 1 | // A legacy (single-peer) receive must release its partial fd when it cannot |
| 2 | // make progress: a chunk that fails verification, or the peer going away. |
| 3 | // |
| 4 | // stortron 2026-10-11: ~2,600 hyper-tps texture receives held one |
| 5 | // `.overlay-partial` fd each for days. The publisher served a stale chunk map, |
| 6 | // every chunk failed the hash check, and _onChunkData ignored the failed |
| 7 | // writeChunk; onclose released schedulers and tree requests but never legacy |
| 8 | // receives. pause() closes the fd and KEEPS the partial, so a later offer |
| 9 | // resumes and re-verifies it. |
| 10 | |
| 11 | import test from 'brittle' |
| 12 | import Protomux from 'protomux' |
| 13 | import { Duplex } from 'streamx' |
| 14 | import b4a from 'b4a' |
| 15 | import { tmpStore } from './helpers.js' |
| 16 | import { FileIndex } from '../lib/file-index.js' |
| 17 | import { OverlayProtocol } from '../lib/protocol.js' |
| 18 | |
| 19 | function makeDuplex () { |
| 20 | let aWrite, bWrite |
| 21 | const a = new Duplex({ write (data, cb) { bWrite(data); cb() }, read () {} }) |
| 22 | const b = new Duplex({ write (data, cb) { aWrite(data); cb() }, read () {} }) |
| 23 | aWrite = (data) => a.push(data) |
| 24 | bWrite = (data) => b.push(data) |
| 25 | return [a, b] |
| 26 | } |
| 27 | const tick = () => new Promise((resolve) => setTimeout(resolve, 50)) |
| 28 | |
| 29 | function recordingTransferManager (fileIndex, writeResult) { |
| 30 | const calls = { started: [], paused: [], cancelled: [] } |
| 31 | return { |
| 32 | calls, |
| 33 | _fileIndex: fileIndex, |
| 34 | async prepareFile () { return null }, |
| 35 | readChunk () { return null }, |
| 36 | computeNeeded (chunks) { return chunks.map((_, i) => i) }, |
| 37 | async startReceive (p) { calls.started.push(p); return {} }, |
| 38 | writeChunk () { return writeResult }, |
| 39 | isComplete () { return false }, |
| 40 | async finalize () { return { ok: false } }, |
| 41 | async pause (p) { calls.paused.push(p) }, |
| 42 | cancel (p) { calls.cancelled.push(p) } |
| 43 | } |
| 44 | } |
| 45 | |
| 46 | async function makePeer (label, tm) { |
| 47 | const fileIndex = new FileIndex(tmpStore(label)) |
| 48 | await fileIndex.ready() |
| 49 | const syncEngine = { |
| 50 | _fileIndex: fileIndex, |
| 51 | feedKey: '0'.repeat(64), |
| 52 | feed: { length: 0, async get () { return null } }, |
| 53 | async markSynced () {}, |
| 54 | async logPut () {} |
| 55 | } |
| 56 | const protocol = new OverlayProtocol(syncEngine, tm(fileIndex), { |
| 57 | key: Buffer.alloc(32, 0xab), |
| 58 | autoSync: false, |
| 59 | filePaths: new Map([['/tex.png', '/spool/tex.png']]) |
| 60 | }) |
| 61 | return { fileIndex, protocol } |
| 62 | } |
| 63 | |
| 64 | async function pair (tmA, tmB) { |
| 65 | const A = await makePeer('lr-a', tmA) |
| 66 | const B = await makePeer('lr-b', tmB) |
| 67 | const [sA, sB] = makeDuplex() |
| 68 | A.protocol.attach(Protomux.from(sA)) |
| 69 | const pB = B.protocol.attach(Protomux.from(sB)) |
| 70 | await tick() |
| 71 | return { A, B, pB } |
| 72 | } |
| 73 | |
| 74 | const MAP = [{ hash: 'a'.repeat(64), length: 191 }] |
| 75 | |
| 76 | test('a chunk that fails verification pauses the receive (fd released, partial kept)', async (t) => { |
| 77 | let tmB |
| 78 | const { A, B, pB } = await pair( |
| 79 | (fi) => recordingTransferManager(fi, { ok: true }), |
| 80 | (fi) => (tmB = recordingTransferManager(fi, { ok: false, error: 'hash mismatch at index 0' })) |
| 81 | ) |
| 82 | await B.protocol._onChunkHashes(pB, { path: '/tex.png', tier: 0, chunks: MAP, more: 0 }) |
| 83 | t.alike(tmB.calls.started, ['/spool/tex.png'], 'receive started') |
| 84 | await B.protocol._onChunkData(pB, { path: '/tex.png', index: 0, data: b4a.alloc(191) }) |
| 85 | t.alike(tmB.calls.paused, ['/spool/tex.png'], 'paused on the failed chunk') |
| 86 | t.alike(tmB.calls.cancelled, [], 'not cancelled: the partial is kept for resume') |
| 87 | A.protocol.destroy(); B.protocol.destroy() |
| 88 | }) |
| 89 | |
| 90 | test('the peer closing pauses every legacy receive it was feeding', async (t) => { |
| 91 | let tmB |
| 92 | const { A, B, pB } = await pair( |
| 93 | (fi) => recordingTransferManager(fi, { ok: true }), |
| 94 | (fi) => (tmB = recordingTransferManager(fi, { ok: true })) |
| 95 | ) |
| 96 | await B.protocol._onChunkHashes(pB, { path: '/tex.png', tier: 0, chunks: MAP, more: 0 }) |
| 97 | t.alike(tmB.calls.paused, [], 'still receiving') |
| 98 | A.protocol.destroy() // closes the channel; B's onclose fires |
| 99 | await tick() |
| 100 | t.alike(tmB.calls.paused, ['/spool/tex.png'], 'paused when the peer went away') |
| 101 | B.protocol.destroy() |
| 102 | }) |