hyper-overlay / test / chunk-scheduler.test.js

// P2 — per-peer inflight window (hypercore replicator getMaxInflight model).
// Verifies the window math directly against the reference formula:
//   no-udx  → min(max, min*3)
//   udx     → clamp(round(min * (rtt<=50 ? 1 : rtt/50 * min(1, 2/peers))), min, max)
// with our tuned range INFLIGHT_RANGE = [4, 64].
import test from 'brittle'
import c from 'compact-encoding'
import b4a from 'b4a'
import { ChunkScheduler } from '../lib/chunk-scheduler.js'
import * as messages from '../lib/messages.js'

function roundtrip (codec, value) {
  const state = { start: 0, end: 0, buffer: null }
  codec.preencode(state, value)
  state.buffer = b4a.alloc(state.end)
  codec.encode(state, value)
  state.start = 0
  return codec.decode(state)
}

function mkSched (opts = {}) {
  const s = new ChunkScheduler({ path: 'content:x', destPath: '/tmp/none', transfer: {}, sendNeed: () => {}, ...opts })
  clearTimeout(s._timer) // don't leave the idle watchdog armed
  s._timer = null
  return s
}
const udx = (rtt) => ({ mux: { stream: { rawStream: { udx: {}, rtt } } } })
const plain = () => ({ mux: { stream: { rawStream: {} } } }) // no udx → no RTT signal

test('P2: no-udx stream gets a fixed min*3 window', (t) => {
  const s = mkSched(); s._peers.add('a')
  t.is(s._maxInflight(plain()), 12, 'min(64, 4*3) = 12')
})

test('P2: udx link at/under SCALE_LATENCY stays at min', (t) => {
  const s = mkSched(); s._peers.add('a')
  t.is(s._maxInflight(udx(20)), 4, 'rtt<=50 → scale 1 → min = 4')
  t.is(s._maxInflight(udx(50)), 4, 'rtt==50 boundary → min = 4')
})

test('P2: udx window scales up with RTT and clamps to max', (t) => {
  const s = mkSched(); s._peers.add('a') // 1 peer → min(1, 2/1) = 1
  t.is(s._maxInflight(udx(200)), 16, 'rtt 200 → 4 * (200/50) = 16')
  t.is(s._maxInflight(udx(100000)), 64, 'huge rtt clamps to max 64')
})

test('P2: more peers shrink the per-peer share (fairness divisor)', (t) => {
  const s = mkSched()
  for (const p of ['a', 'b', 'c', 'd']) s._peers.add(p) // 4 peers → 2/4 = 0.5
  t.is(s._maxInflight(udx(200)), 8, 'rtt 200, 4 peers → 4 * (200/50 * 0.5) = 8')
})

test('P2: opts.cap overrides the dynamic window (back-compat)', (t) => {
  const s = mkSched({ cap: 3 }); s._peers.add('a')
  t.is(s._maxInflight(udx(99999)), 3, 'fixed cap wins over RTT scaling')
  t.is(s._maxInflight(plain()), 3, 'fixed cap wins over the no-udx default')
})

test('P5 hotswap: an idle peer steals a chunk stalled on a slow peer', (t) => {
  const needs = []
  const cancels = []
  const s = mkSched({
    cap: 4, // fixed window so both peers have spare capacity deterministically
    hotswapStallMs: 50,
    sendNeed: (peer, idx) => needs.push([peer, idx]),
    sendCancel: (peer, idx) => cancels.push([peer, idx])
  })
  s._chunks = [{ hash: 'h0', offset: 0, length: 10 }]
  const slow = udx(10)
  const fast = udx(10)
  s._peers.add(slow) // iterated first (holds nothing to steal), then fast steals
  s._peers.add(fast)
  // chunk 0 is inflight to the slow peer and has been for a while (stalled).
  s._inflight.set(0, slow)
  s._inflightAt.set(0, Date.now() - 1000)
  s._peerInflight.set(slow, 1)
  s._peerInflight.set(fast, 0)
  // _needed is empty — nothing fresh to fetch, only the stalled chunk remains.

  s._assign()

  t.is(s._inflight.get(0), fast, 'chunk 0 reassigned from slow → fast')
  t.ok(needs.some(([p, idx]) => p === fast && idx.includes(0)), 'fast peer re-requested chunk 0')
  t.ok(cancels.some(([p, idx]) => p === slow && idx.includes(0)), 'slow peer told to cancel chunk 0')
  t.is(s._hotswaps, 1, 'one hotswap recorded')
  t.is(s._peerInflight.get(slow), 0, 'slow peer inflight decremented')
  t.is(s._peerInflight.get(fast), 1, 'fast peer inflight incremented')
})

test('P5 hotswap: fresh chunks are assigned before any stealing', (t) => {
  const needs = []
  const s = mkSched({ cap: 4, hotswapStallMs: 50, sendNeed: (peer, idx) => needs.push([peer, idx]) })
  s._chunks = [0, 1].map((i) => ({ hash: 'h' + i, offset: i * 10, length: 10 }))
  const a = udx(10)
  s._peers.add(a)
  s._needed.add(1) // chunk 1 is fresh/needed
  s._inflight.set(0, udx(10)) // chunk 0 inflight to some other (already-gone) peer, stalled
  s._inflightAt.set(0, Date.now() - 1000)
  s._assign()
  // peer a should take the FRESH chunk 1 first (not immediately hotswap chunk 0)
  t.ok(s._inflight.get(1) === a, 'fresh chunk 1 assigned to peer a')
})

test('P5 wire: chunkCancel round-trips with indices, back-compat without', (t) => {
  const withIdx = roundtrip(messages.chunkCancel, { path: 'content:abc', indices: [3, 7, 9] })
  t.is(withIdx.path, 'content:abc')
  t.alike(withIdx.indices, [3, 7, 9], 'per-chunk indices preserved')
  const noIdx = roundtrip(messages.chunkCancel, { path: 'content:abc' })
  t.is(noIdx.path, 'content:abc')
  t.absent(noIdx.indices, 'absent indices → bare {path} (whole-transfer cancel; pre-0.3.0 shape)')
})