atproto-room / test / membrane-stale-gate.test.mjs

// serveMembrane's projection gate must be REV/­content-aware, not length-aware.
//
// The old gate was `if (log.length <= seq.length) return`, which silently assumes the __commit__ log
// only ever GROWS. After a base rebuild or resync the log RESTARTS SHORTER, so a node carrying an old
// membrane (e.g. 83 projected frames) with a fresh log (e.g. 49 entries) satisfies 49 <= 83 and NEVER
// PROJECTS AGAIN. It keeps announcing a PRE-REBUILD head while its device head is current.
//
// Observed live on wj62: a co-owner stuck at rev 3mr436gzyts22 for four days, still announced, so a
// bridge could front it and serve a REV REGRESSION to the relay — which parks a forward-only cursor.
// The condition was invisible from every surface, which is why it survived.
//
// The gate must NOT "fix" this by appending the new log after the stale frames: the membrane firehose
// is append-only and replicated, so that would put a rev regression MID-STREAM, which is worse than
// freezing. Correct behaviour is to refuse and say so loudly.
//
// Run: bare test/membrane-stale-gate.test.mjs
import '../lib/bare-globals.js'
import test from 'brittle'
import b4a from 'b4a'
import { serveMembrane } from '../lib/serve-membrane.js'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import fs from 'bare-fs'

const DID = 'did:plc:membranestalegate000000'
const tmp = () => { const d = '/tmp/ar-msg-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d }

// serveMembrane only touches these few swarm members.
function fakeSwarm () {
  return {
    on () {}, off () {},
    keyPair: { publicKey: b4a.alloc(32, 7) },
    join () { return { flushed: async () => {} } }
  }
}

// A device whose firehose log we control outright. listRecords is empty so projectRepo is a no-op.
function fakeDevice (log = []) {
  return {
    _log: log,
    repo: { listRecords: async () => [] },
    async firehoseLog () { return this._log }
  }
}

async function capture (t, fn) {
  const lines = []
  const orig = console.error
  console.error = (s) => lines.push(String(s))
  t.teardown(() => { console.error = orig })
  try { await fn() } finally { console.error = orig }
  return lines
}

test('a log SHORTER than the membrane (base rebuild) must NOT append, and must say so', async (t) => {
  const device = fakeDevice([])
  const serving = await serveMembrane(device, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID })

  // Simulate an established membrane: 3 already-projected frames.
  for (let i = 0; i < 3; i++) await serving.seq.append(b4a.from('frame-' + i))
  t.is(serving.seq.core.length, 3, 'membrane has 3 projected frames')

  // The base was rebuilt: the __commit__ log restarts SHORTER than what the membrane already holds.
  device._log = [{ rev: '3knewbaserev0', seq: 1 }]

  const logs = await capture(t, () => serving.sync())

  t.is(serving.seq.core.length, 3, 'membrane was NOT appended to — no rev regression mid-stream')
  t.is(logs.length, 1, 'exactly one warning emitted')
  t.ok(/membrane STALE/.test(logs[0]), 'the stale condition is named explicitly')
  const j = JSON.parse(logs[0])
  t.is(j.did, DID, 'names the DID')
  t.is(j.logLength, 1, 'reports the log length')
  t.is(j.membraneLength, 3, 'reports the membrane length — the two numbers that diagnose it')

  const again = await capture(t, () => serving.sync())
  t.is(again.length, 0, 'deduped — it does not spam every sync tick')
})

test('an EQUAL-length log is fully projected — the anti-balloon guard still holds', async (t) => {
  const device = fakeDevice([])
  const serving = await serveMembrane(device, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID })
  for (let i = 0; i < 2; i++) await serving.seq.append(b4a.from('f' + i))

  device._log = [{ rev: 'a', seq: 1 }, { rev: 'b', seq: 2 }]
  await serving.sync()

  // The original balloon was projectRepo re-putting the WHOLE record set on every ~2s tick with zero
  // new commits. Nothing new => nothing written, still.
  t.is(serving.seq.core.length, 2, 'nothing appended when the log is fully projected')
})

test('an empty log is a no-op (a device that has never signed)', async (t) => {
  const device = fakeDevice([])
  const serving = await serveMembrane(device, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID })
  await serving.sync()
  t.is(serving.seq.core.length, 0, 'no frames, no writes, no warning')
})