hyper-proto-peer / test / conflict-watch.test.js

// conflict-watch — does a REAL hypercore conflict actually get recorded?
//
// This test drives an end-to-end conflict between two live cores rather than calling `recordConflict`
// directly, and that is the entire point of it. The watcher hooks an INTERNAL hypercore API (`conflict` is
// emitted on a Core's monitors, and most cores in the daemon are opened by autobase, so there is no session
// of ours to attach a public listener to). A unit test of the registry would keep passing forever while the
// hook silently stopped firing — and a conflict surface that always reports zero is worse than no surface,
// because it converts "we are not looking" into "we looked and it is fine". That is precisely the failure
// #17 records: loser-club was declared converged while both machines were logging conflicts.
//
// So: two stores, ONE keypair, two different histories, replicated against each other. If hypercore ever
// stops emitting the way this expects, this goes red.
//
// Run: bare test/conflict-watch.test.js
import '../lib/bare-globals.js'
import test from 'brittle'
import fs from 'bare-fs'
import { join } from 'bare-path'
import b4a from 'b4a'
import Corestore from 'corestore'
import crypto from 'hypercore-crypto'
import { watchStore, conflictDiag, recordConflict, _reset } from '../lib/conflict-watch.js'

const tmp = () => {
  const d = '/tmp/hpp-conflict-' + Math.floor(Math.random() * 1e9)
  fs.mkdirSync(d, { recursive: true })
  return d
}

test('a real conflict between two cores is recorded', async (t) => {
  _reset()
  const root = tmp()

  // One keypair, two divergent histories — the shape that makes "two conflicting signatures exist for
  // length N" true. B is LONGER so A has a reason to request an upgrade from it: a conflict is only
  // discovered when someone actually asks for blocks they do not have.
  const kp = crypto.keyPair()
  const a = watchStore(new Corestore(join(root, 'a')), 'A')
  const b = watchStore(new Corestore(join(root, 'b')), 'B')

  const ca = a.get({ keyPair: kp })
  const cb = b.get({ keyPair: kp })
  await ca.ready()
  await cb.ready()
  for (const blk of ['same-0', 'same-1', 'MINE']) await ca.append(b4a.from(blk))
  for (const blk of ['same-0', 'same-1', 'THEIRS', 'extra-3', 'extra-4']) await cb.append(b4a.from(blk))

  const s1 = a.replicate(true)
  const s2 = b.replicate(false)
  s1.on('error', () => {})
  s2.on('error', () => {})
  s1.pipe(s2).pipe(s1)

  // Give the upgrade attempt a bounded window, polling rather than sleeping a fixed time so a fast
  // detection does not pay for a slow one.
  const deadline = Date.now() + 15000
  while (Date.now() < deadline && conflictDiag().seenSinceBoot === 0) {
    try { await ca.update() } catch {}
    try { await ca.get(3, { timeout: 200 }) } catch {}
    await new Promise((resolve) => setTimeout(resolve, 200))
  }

  const report = conflictDiag()
  t.ok(report.seenSinceBoot > 0, 'the conflict was RECORDED — the hook into hypercore still fires')

  const rec = report.conflicts[0]
  t.is(rec.discoveryKey, b4a.toString(ca.discoveryKey, 'hex'), 'and names the core that forked')
  t.is(rec.key, b4a.toString(ca.key, 'hex'), 'with its public key')
  t.is(rec.writable, true, 'and reports that THIS side can sign it — our key authored two histories')
  t.ok(rec.length > 0, 'at a length')
  t.ok(rec.firstSeenAt, 'timestamped')

  try { s1.destroy() } catch {}
  try { s2.destroy() } catch {}
  try { await a.close() } catch {}
  try { await b.close() } catch {}
})

test('a healthy store records nothing', async (t) => {
  _reset()
  const dir = join(tmp(), 'clean')
  const store = watchStore(new Corestore(dir), 'clean')
  const core = store.get({ name: 'ok' })
  await core.ready()
  await core.append(b4a.from('fine'))

  // The common case MUST stay silent, or the surface is noise and gets ignored — which is how the
  // upstream console.log ended up unread in the first place.
  t.is(conflictDiag().seenSinceBoot, 0, 'no conflict on a store that has none')
  await store.close()
})

test('a repeating conflict aggregates instead of flooding', async (t) => {
  _reset()
  const one = { discoveryKey: 'dd', key: 'kk', length: 7, fork: 0, writable: false, quorum: 1, store: 's' }
  recordConflict(one)
  recordConflict(one)
  recordConflict(one)

  // A conflicted core is re-offered on every reconnect, so an unbounded log would bury the finding under
  // its own repetitions. One row, a rising count: the count is itself the signal that it keeps happening.
  const r = conflictDiag()
  t.is(r.seenSinceBoot, 1, 'one distinct conflict')
  t.is(r.total, 3, 'seen three times')
  t.is(r.conflicts[0].count, 3)
  t.not(r.conflicts[0].firstSeenAt, undefined, 'first and last sighting both kept')
  t.ok(r.conflicts[0].lastSeenAt)

  // A DIFFERENT length on the same core is a different fact, not a repeat.
  recordConflict({ ...one, length: 9 })
  t.is(conflictDiag().seenSinceBoot, 2, 'a conflict at another length is its own row')
})