hyper-overlay / test / legacy-receive-release.test.js

// A legacy (single-peer) receive must release its partial fd when it cannot
// make progress: a chunk that fails verification, or the peer going away.
//
// stortron 2026-10-11: ~2,600 hyper-tps texture receives held one
// `.overlay-partial` fd each for days. The publisher served a stale chunk map,
// every chunk failed the hash check, and _onChunkData ignored the failed
// writeChunk; onclose released schedulers and tree requests but never legacy
// receives. pause() closes the fd and KEEPS the partial, so a later offer
// resumes and re-verifies it.

import test from 'brittle'
import Protomux from 'protomux'
import { Duplex } from 'streamx'
import b4a from 'b4a'
import { tmpStore } from './helpers.js'
import { FileIndex } from '../lib/file-index.js'
import { OverlayProtocol } from '../lib/protocol.js'

function makeDuplex () {
  let aWrite, bWrite
  const a = new Duplex({ write (data, cb) { bWrite(data); cb() }, read () {} })
  const b = new Duplex({ write (data, cb) { aWrite(data); cb() }, read () {} })
  aWrite = (data) => a.push(data)
  bWrite = (data) => b.push(data)
  return [a, b]
}
const tick = () => new Promise((resolve) => setTimeout(resolve, 50))

function recordingTransferManager (fileIndex, writeResult) {
  const calls = { started: [], paused: [], cancelled: [] }
  return {
    calls,
    _fileIndex: fileIndex,
    async prepareFile () { return null },
    readChunk () { return null },
    computeNeeded (chunks) { return chunks.map((_, i) => i) },
    async startReceive (p) { calls.started.push(p); return {} },
    writeChunk () { return writeResult },
    isComplete () { return false },
    async finalize () { return { ok: false } },
    async pause (p) { calls.paused.push(p) },
    cancel (p) { calls.cancelled.push(p) }
  }
}

async function makePeer (label, tm) {
  const fileIndex = new FileIndex(tmpStore(label))
  await fileIndex.ready()
  const syncEngine = {
    _fileIndex: fileIndex,
    feedKey: '0'.repeat(64),
    feed: { length: 0, async get () { return null } },
    async markSynced () {},
    async logPut () {}
  }
  const protocol = new OverlayProtocol(syncEngine, tm(fileIndex), {
    key: Buffer.alloc(32, 0xab),
    autoSync: false,
    filePaths: new Map([['/tex.png', '/spool/tex.png']])
  })
  return { fileIndex, protocol }
}

async function pair (tmA, tmB) {
  const A = await makePeer('lr-a', tmA)
  const B = await makePeer('lr-b', tmB)
  const [sA, sB] = makeDuplex()
  A.protocol.attach(Protomux.from(sA))
  const pB = B.protocol.attach(Protomux.from(sB))
  await tick()
  return { A, B, pB }
}

const MAP = [{ hash: 'a'.repeat(64), length: 191 }]

test('a chunk that fails verification pauses the receive (fd released, partial kept)', async (t) => {
  let tmB
  const { A, B, pB } = await pair(
    (fi) => recordingTransferManager(fi, { ok: true }),
    (fi) => (tmB = recordingTransferManager(fi, { ok: false, error: 'hash mismatch at index 0' }))
  )
  await B.protocol._onChunkHashes(pB, { path: '/tex.png', tier: 0, chunks: MAP, more: 0 })
  t.alike(tmB.calls.started, ['/spool/tex.png'], 'receive started')
  await B.protocol._onChunkData(pB, { path: '/tex.png', index: 0, data: b4a.alloc(191) })
  t.alike(tmB.calls.paused, ['/spool/tex.png'], 'paused on the failed chunk')
  t.alike(tmB.calls.cancelled, [], 'not cancelled: the partial is kept for resume')
  A.protocol.destroy(); B.protocol.destroy()
})

test('the peer closing pauses every legacy receive it was feeding', async (t) => {
  let tmB
  const { A, B, pB } = await pair(
    (fi) => recordingTransferManager(fi, { ok: true }),
    (fi) => (tmB = recordingTransferManager(fi, { ok: true }))
  )
  await B.protocol._onChunkHashes(pB, { path: '/tex.png', tier: 0, chunks: MAP, more: 0 })
  t.alike(tmB.calls.paused, [], 'still receiving')
  A.protocol.destroy() // closes the channel; B's onclose fires
  await tick()
  t.alike(tmB.calls.paused, ['/spool/tex.png'], 'paused when the peer went away')
  B.protocol.destroy()
})