atproto-bridge / pds / test / fleet-sequencer.test.js

// FleetSequencer — the proof for §11's "one genuinely hard sub-problem": the edge assigns
// a stable, append-only, monotonic firehose seq even when the underlying fleet aggregation
// RELINEARIZES (reorders history, re-delivers events). This is the fan-in + edge-seq
// pattern the audit prefers over autobase-as-firehose (docs/ARCHITECTURE-AUDIT.md).

import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { join } from 'node:path'
import { tmpdir } from 'node:os'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { FleetSequencer } from '../lib/fleet-sequencer.js'

function open (dir) {
  const store = openCorestore(dir)
  const core = store.get({ name: 'fleet-firehose', valueEncoding: 'json' })
  return { store, core }
}

// simulate events from two fleet instances, each with a stable content-addressed key
const evt = (writer, n) => ({ key: `${writer}#commit:${n}`, data: { writer, n } })

test('seq is stable + append-only under relinearization, dedup, and re-delivery', async (t) => {
  const dir = mkdtempSync(join(tmpdir(), 'fleetseq-'))
  const { store, core } = open(dir)
  const fs = await new FleetSequencer(core).ready()
  t.teardown(async () => { await store.close(); rmSync(dir, { recursive: true, force: true }) })

  // 1. ingest in causal order A1,B1,A2,B2,A3 -> seqs 1..5, all fresh
  const first = [evt('A', 1), evt('B', 1), evt('A', 2), evt('B', 2), evt('A', 3)]
  const r1 = await fs.ingestAll(first)
  t.alike(r1.map(r => r.seq), [1, 2, 3, 4, 5], 'first pass assigns 1..5')
  t.ok(r1.every(r => !r.deduped), 'all fresh')
  const assigned = new Map(first.map((e, i) => [e.key, i + 1]))

  // 2. RELINEARIZE: the autobase view reorders the SAME events and re-delivers with dups
  const reordered = [evt('B', 2), evt('A', 1), evt('A', 3), evt('B', 1), evt('A', 2), evt('B', 2)]
  const r2 = await fs.ingestAll(reordered)
  t.ok(r2.every(r => r.deduped), 'every re-delivered event deduped (no new append)')
  t.is(fs.length, 5, 'length unchanged — nothing re-appended')
  for (const e of reordered) t.is(fs.seqOf(e.key), assigned.get(e.key), `seq stable for ${e.key}`)

  // 3. a genuinely NEW event after the reorder gets the next seq (append-only continues)
  const r3 = await fs.ingest(evt('B', 3))
  t.is(r3.seq, 6, 'new event -> seq 6')
  t.absent(r3.deduped, 'new event is fresh')

  // 4. subscribe(cursor=0) replays the ORIGINAL emit order, not the reordered ingest
  const seen = []
  const ac = new AbortController()
  for await (const f of fs.subscribe(0, { signal: ac.signal })) {
    seen.push([f.seq, f.key])
    if (seen.length === 6) { ac.abort(); break }
  }
  t.alike(seen, [
    [1, 'A#commit:1'], [2, 'B#commit:1'], [3, 'A#commit:2'],
    [4, 'B#commit:2'], [5, 'A#commit:3'], [6, 'B#commit:3']
  ], 'firehose replays in stable first-seen order — the cursor contract holds')

  // 5. cursor replay from a midpoint is exactly the tail
  const tail = []
  const ac2 = new AbortController()
  for await (const f of fs.subscribe(3, { signal: ac2.signal })) {
    tail.push(f.seq)
    if (tail.length === 3) { ac2.abort(); break }
  }
  t.alike(tail, [4, 5, 6], 'subscribe(cursor=3) yields exactly seq 4,5,6')
})

test('seq survives a restart — the emit log IS the durable cursor map', async (t) => {
  const dir = mkdtempSync(join(tmpdir(), 'fleetdur-'))
  t.teardown(() => rmSync(dir, { recursive: true, force: true }))

  // session 1: assign three
  {
    const { store, core } = open(dir)
    const fs = await new FleetSequencer(core).ready()
    await fs.ingestAll([evt('A', 1), evt('B', 1), evt('A', 2)])
    t.is(fs.length, 3, 'assigned 3')
    await store.close()
  }
  // session 2: reopen — seen-map rebuilt from the log, seqs stable, re-delivery deduped,
  // and a new event continues at 4
  {
    const { store, core } = open(dir)
    const fs = await new FleetSequencer(core).ready()
    t.is(fs.length, 3, 'reopened with 3 emitted')
    t.is(fs.seqOf('A#commit:1'), 1, 'seq for A1 survived restart')
    t.is(fs.seqOf('A#commit:2'), 3, 'seq for A2 survived restart')
    const redelivered = await fs.ingest(evt('B', 1))
    t.ok(redelivered.deduped && redelivered.seq === 2, 're-delivered post-restart still dedups to its original seq')
    const fresh = await fs.ingest(evt('C', 1))
    t.ok(!fresh.deduped && fresh.seq === 4, 'new event continues append-only at seq 4')
    await store.close()
  }
})