atproto-base / test / finalized-view.test.mjs

// ENGINE-LEVEL REORG-SAFETY — head()/indexedRecords() project ONLY the FINALIZED view (checked out
// at the view core's signedLength), so an un-finalized optimistic write NEVER leaks into the signed
// atproto chain. Proof: a plain writer, disconnected from the sole indexer, writes a record — it lands
// in the writer's optimistic view but is ABSENT from its finalized head, until an indexer finalizes it.
// (This is the head-level counterpart of proof #3/#5; it guarantees the projection can't sign a tip
// that later reorgs, and that an N=2 stall simply omits the un-finalized write rather than forking.)

import test from 'brittle'
import { mkdtempSync, rmSync } from 'node:fs'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import crypto from 'hypercore-crypto'
import { deriveAtprotoKey, createAtprotoIdentity } from 'atproto-identity'
import { readCar } from 'atproto-repo/lib/car.js'
import { verifyRepo } from 'atproto-repo'
import { AtprotoBase } from '../lib/base.js'

const MNEMONIC = 'test test test test test test test test test test test junk'
const ID = createAtprotoIdentity(MNEMONIC, { handle: 'x.7rnx.net', pdsUrl: 'https://pds.7rnx.net' })
const KEY = deriveAtprotoKey(MNEMONIC)
const post = (text) => ({ $type: 'app.bsky.feed.post', text, createdAt: '2026-07-23T00:00:00.000Z', langs: ['en'] })
const hex = (b) => b.toString('hex')
const sleep = (ms) => new Promise((r) => setTimeout(r, ms))
function mk (opts) { const dir = mkdtempSync(join(tmpdir(), 'fv-')); const base = new AtprotoBase({ path: dir, did: ID.did, privateKey: KEY.privateKey, ackInterval: 100, ...opts }); return { base, dir, close: async () => { await base.close(); rmSync(dir, { recursive: true, force: true }) } } }
const pipe = (a, b) => { const s1 = a.base.base.replicate(true); const s2 = b.base.base.replicate(false); s1.pipe(s2).pipe(s1); return () => { try { s1.destroy() } catch {} ; try { s2.destroy() } catch {} } }
async function settle (...ws) { for (let i = 0; i < 15; i++) { for (const w of ws) { try { await w.base.base.update() } catch {} } await sleep(60) } }
async function fullViewCount (bs) { let n = 0; for await (const { key } of bs.base.base.view.createReadStream()) { if (!String(key).startsWith('__')) n++ } return n }
async function recordCount (car) { return (await verifyRepo(await readCar(car))).records }

test('finalized-view: an un-finalized write is in the optimistic view but NEVER in the signed head', async (t) => {
  const creatorId = crypto.keyPair()
  const writerId = crypto.keyPair()
  const roster = [{ key: hex(creatorId.publicKey), role: 1 }, { key: hex(writerId.publicKey), role: 3 }]

  const C = mk({ identity: creatorId, members: roster }) // creator = sole indexer
  await C.base.ready()
  const P = mk({ bootstrap: C.base.base.key, identity: writerId }) // P = plain writer
  await P.base.ready()
  let unpipe = pipe(C, P)
  t.teardown(async () => { try { unpipe() } catch {} ; await C.close(); await P.close() })

  await C.base.projectMembers(roster); await settle(C, P)
  await P.base.putRecord('app.bsky.feed.post', '3one', post('finalized')) // admitted + written
  await settle(C, P); await P.base.settled()
  t.is(await recordCount((await P.base.head()).carBytes), 1, 'first write is finalized and in the signed head')

  // Disconnect P from the sole indexer; P writes again — this write CANNOT finalize.
  unpipe()
  await P.base.putRecord('app.bsky.feed.post', '3two', post('un-finalized'))
  await P.base.settled({ timeoutMs: 1500 }) // will not advance — nothing to finalize it

  t.is(await fullViewCount(P), 2, "the un-finalized write IS in P's optimistic view (durable-local)")
  t.is(await recordCount((await P.base.head()).carBytes), 1, 'but the un-finalized write is ABSENT from the finalized signed head (no tip leaks into the chain)')

  // Reconnect → the sole indexer finalizes it → it now appears in the head (no fork, just delayed).
  unpipe = pipe(C, P)
  await settle(C, P); await P.base.settled()
  t.is(await recordCount((await P.base.head()).carBytes), 2, 'after reconnect the indexer finalizes it → now in the signed head (stall resolved, never forked)')
})