| … | … | @@ -0,0 +1,97 @@ |
| 1 | // serveMembrane's projection gate must be REV/content-aware, not length-aware. |
| 2 | // |
| 3 | // The old gate was `if (log.length <= seq.length) return`, which silently assumes the __commit__ log |
| 4 | // only ever GROWS. After a base rebuild or resync the log RESTARTS SHORTER, so a node carrying an old |
| 5 | // membrane (e.g. 83 projected frames) with a fresh log (e.g. 49 entries) satisfies 49 <= 83 and NEVER |
| 6 | // PROJECTS AGAIN. It keeps announcing a PRE-REBUILD head while its device head is current. |
| 7 | // |
| 8 | // Observed live on wj62: a co-owner stuck at rev 3mr436gzyts22 for four days, still announced, so a |
| 9 | // bridge could front it and serve a REV REGRESSION to the relay — which parks a forward-only cursor. |
| 10 | // The condition was invisible from every surface, which is why it survived. |
| 11 | // |
| 12 | // The gate must NOT "fix" this by appending the new log after the stale frames: the membrane firehose |
| 13 | // is append-only and replicated, so that would put a rev regression MID-STREAM, which is worse than |
| 14 | // freezing. Correct behaviour is to refuse and say so loudly. |
| 15 | // |
| 16 | // Run: bare test/membrane-stale-gate.test.mjs |
| 17 | import '../lib/bare-globals.js' |
| 18 | import test from 'brittle' |
| 19 | import b4a from 'b4a' |
| 20 | import { serveMembrane } from '../lib/serve-membrane.js' |
| 21 | import { openCorestore } from 'atproto-store/lib/corestore.js' |
| 22 | import fs from 'bare-fs' |
| 23 | |
| 24 | const DID = 'did:plc:membranestalegate000000' |
| 25 | const tmp = () => { const d = '/tmp/ar-msg-' + Math.floor(Math.random() * 1e9); fs.mkdirSync(d, { recursive: true }); return d } |
| 26 | |
| 27 | // serveMembrane only touches these few swarm members. |
| 28 | function fakeSwarm () { |
| 29 | return { |
| 30 | on () {}, off () {}, |
| 31 | keyPair: { publicKey: b4a.alloc(32, 7) }, |
| 32 | join () { return { flushed: async () => {} } } |
| 33 | } |
| 34 | } |
| 35 | |
| 36 | // A device whose firehose log we control outright. listRecords is empty so projectRepo is a no-op. |
| 37 | function fakeDevice (log = []) { |
| 38 | return { |
| 39 | _log: log, |
| 40 | repo: { listRecords: async () => [] }, |
| 41 | async firehoseLog () { return this._log } |
| 42 | } |
| 43 | } |
| 44 | |
| 45 | async function capture (t, fn) { |
| 46 | const lines = [] |
| 47 | const orig = console.error |
| 48 | console.error = (s) => lines.push(String(s)) |
| 49 | t.teardown(() => { console.error = orig }) |
| 50 | try { await fn() } finally { console.error = orig } |
| 51 | return lines |
| 52 | } |
| 53 | |
| 54 | test('a log SHORTER than the membrane (base rebuild) must NOT append, and must say so', async (t) => { |
| 55 | const device = fakeDevice([]) |
| 56 | const serving = await serveMembrane(device, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID }) |
| 57 | |
| 58 | // Simulate an established membrane: 3 already-projected frames. |
| 59 | for (let i = 0; i < 3; i++) await serving.seq.append(b4a.from('frame-' + i)) |
| 60 | t.is(serving.seq.core.length, 3, 'membrane has 3 projected frames') |
| 61 | |
| 62 | // The base was rebuilt: the __commit__ log restarts SHORTER than what the membrane already holds. |
| 63 | device._log = [{ rev: '3knewbaserev0', seq: 1 }] |
| 64 | |
| 65 | const logs = await capture(t, () => serving.sync()) |
| 66 | |
| 67 | t.is(serving.seq.core.length, 3, 'membrane was NOT appended to — no rev regression mid-stream') |
| 68 | t.is(logs.length, 1, 'exactly one warning emitted') |
| 69 | t.ok(/membrane STALE/.test(logs[0]), 'the stale condition is named explicitly') |
| 70 | const j = JSON.parse(logs[0]) |
| 71 | t.is(j.did, DID, 'names the DID') |
| 72 | t.is(j.logLength, 1, 'reports the log length') |
| 73 | t.is(j.membraneLength, 3, 'reports the membrane length — the two numbers that diagnose it') |
| 74 | |
| 75 | const again = await capture(t, () => serving.sync()) |
| 76 | t.is(again.length, 0, 'deduped — it does not spam every sync tick') |
| 77 | }) |
| 78 | |
| 79 | test('an EQUAL-length log is fully projected — the anti-balloon guard still holds', async (t) => { |
| 80 | const device = fakeDevice([]) |
| 81 | const serving = await serveMembrane(device, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID }) |
| 82 | for (let i = 0; i < 2; i++) await serving.seq.append(b4a.from('f' + i)) |
| 83 | |
| 84 | device._log = [{ rev: 'a', seq: 1 }, { rev: 'b', seq: 2 }] |
| 85 | await serving.sync() |
| 86 | |
| 87 | // The original balloon was projectRepo re-putting the WHOLE record set on every ~2s tick with zero |
| 88 | // new commits. Nothing new => nothing written, still. |
| 89 | t.is(serving.seq.core.length, 2, 'nothing appended when the log is fully projected') |
| 90 | }) |
| 91 | |
| 92 | test('an empty log is a no-op (a device that has never signed)', async (t) => { |
| 93 | const device = fakeDevice([]) |
| 94 | const serving = await serveMembrane(device, { corestore: openCorestore(tmp()), swarm: fakeSwarm(), did: DID }) |
| 95 | await serving.sync() |
| 96 | t.is(serving.seq.core.length, 0, 'no frames, no writes, no warning') |
| 97 | }) |