cthuloid / webui / test / unit-v3.test.js

// Unit against a proto-3 unit: grants, per-slot coalescing, release ordering.
import test from 'brittle'
import { Unit } from '../lib/unit.js'
import { FakeUnitV3, waitFor, sleep } from './helpers.js'

async function connected (fake, opts = {}) {
  const u = new Unit({ id: fake.name, host: '127.0.0.1', port: fake.port, backoffStartMs: 100, ...opts })
  u.open()
  await waitFor(() => u.status === 'connected' && u.proto === 3)
  return u
}

test('hello: proto, epoch and key are learned', async (t) => {
  const fake = await new FakeUnitV3('u3', { epoch: 9 }).listen()
  const u = await connected(fake)
  t.is(u.proto, 3)
  t.is(u.epoch, 9)
  t.is(u.pk, 'cd'.repeat(32))
  u.close()
  await fake.close()
})

test('grant: first come first served, across sockets', async (t) => {
  const fake = await new FakeUnitV3().listen()
  const a = await connected(fake)
  const b = await connected(fake)
  const ga = await a.grant(1, 'alice')
  t.alike(ga.granted, [{ slot: 1, epoch: 5, n: 1 }])
  const gb = await b.grant(1, 'bob')
  t.alike(gb.busy, { slot: 1, holder: 'alice' })
  const all = await b.grant('all', 'bob')
  t.alike(all.busy, { slot: 1, holder: 'alice' }, 'all is atomic')
  a.close()
  b.close()
  await fake.close()
})

test('slot frames coalesce per slot and never cross slots', async (t) => {
  const fake = await new FakeUnitV3().listen()
  const u = await connected(fake, { minIntervalMs: 40 })
  const g0 = (await u.grant(0, 'alice')).granted[0]
  const g1 = (await u.grant(1, 'alice')).granted[0]
  for (let i = 1; i <= 40; i++) {
    u.sendSlot(0, g0, [i, 0, 0], 0)
    u.sendSlot(1, g1, [0, i, 0], 0)
  }
  t.ok(await waitFor(() => fake.values[0] === 40 && fake.values[4] === 40), 'both slots reach the latest value')
  const sent = fake.received((l) => l.startsWith('slot '))
  t.ok(sent.length <= 8, 'intermediate frames dropped: sent ' + sent.length)
  u.close()
  await fake.close()
})

test('release flushes the pending frame first, so the final value is kept', async (t) => {
  const fake = await new FakeUnitV3().listen()
  const u = await connected(fake, { minIntervalMs: 1000 }) // a long window: frame still pending
  const frees = []
  u.on('free', (f) => frees.push(f))
  const g = (await u.grant(2, 'alice')).granted[0]
  u.sendSlot(2, g, [1, 1, 1], 0)
  u.sendSlot(2, g, [7, 8, 9], 0)    // pending behind the rate limit
  u.release(2)
  t.ok(await waitFor(() => frees.length === 1), 'free arrived')
  t.alike([frees[0].b, frees[0].g, frees[0].r], [7, 8, 9], 'free carries the last value, not an older one')
  t.is(fake.received((l) => l.startsWith('ERR')).length, 0)
  u.close()
  await fake.close()
})

test('broadcasts are emitted: held, free, values', async (t) => {
  const fake = await new FakeUnitV3().listen()
  const u = await connected(fake)
  const seen = { held: [], free: [], values: [] }
  for (const k of Object.keys(seen)) u.on(k, (x) => seen[k].push(x))
  const ext = await fake.externalGrab(3, 'blender')
  t.ok(await waitFor(() => seen.held.length === 1), 'held from another controller')
  t.is(seen.held[0].holder, 'blender')
  ext.write('slot 3 5 1 10 20 30 0\n')
  t.ok(await waitFor(() => seen.values.length > 0 && u.actual.values[9] === 10), 'values update actual')
  ext.write('release 3\n')
  t.ok(await waitFor(() => seen.free.length === 1), 'free from another controller')
  ext.destroy()
  u.close()
  await fake.close()
})

test('a dropped connection fails pending grants and resets', async (t) => {
  const fake = await new FakeUnitV3().listen()
  const u = await connected(fake)
  let resets = 0
  u.on('reset', () => resets++)
  const pending = u.grant(0, 'alice')
  for (const c of fake.conns.values()) c.destroy()
  const r = await pending
  t.ok(r.error || r.granted, 'grant settled: ' + JSON.stringify(r))
  t.ok(await waitFor(() => resets >= 1), 'reset emitted')
  u.close()
  await fake.close()
})