favela-web / zz-repro-leak.mjs

// MINIMAL REPRO of the favela-web write leak — one peer (wj62), throwaway storage.
// Isolates favela's ingest: is it a FRAME FLOOD (#sync/#commit in a loop) or REPLICATION re-download
// (write_bytes climbs with ~no frames)? Instruments both. Run: bare zz-repro-leak.mjs ; Ctrl-C to stop.
import './lib/bare-globals.js'
import Hyperswarm from 'hyperswarm'
import { DidPeerRouter } from 'atproto-did-peer'
import { openCorestore } from 'atproto-store/lib/corestore.js'
import { discoveryTopic } from 'atproto-identity/lib/discovery.js'
import { decodeFrame } from 'atproto-store/lib/firehose.js'
import { resolveVerifyKey } from './lib/plc.js'
import bareFetch from 'bare-fetch'
import fs from 'bare-fs'

const DID = (typeof Bare !== 'undefined' && Bare.argv && Bare.argv[2]) ? Bare.argv[2] : 'did:plc:wj62ebbusln6gxo5azo6ogzo'
const STORE = '/tmp/zz-repro-did-' + Date.now()

const swarm = new Hyperswarm()
const store = openCorestore(STORE)
const router = new DidPeerRouter(store)
swarm.on('connection', (c) => router.onconnection(c))

const verifyKey = await resolveVerifyKey(DID, { fetch: bareFetch })
const peer = router.add(DID, { verifyKey })
swarm.join(discoveryTopic(DID), { client: true, server: true })
await swarm.flush()
console.log('joined; waiting for verified head...')
const h = await peer.ready({ timeout: 60000 })
console.log('head:', h ? h.rev : null, '| firehoseLen:', peer.seq ? peer.seq.core.length : 0, '| repoLen:', peer.repo ? peer.repo.core.length : 0)
// Mimic favela's indexPeer: does this peer SERVE its records to a fresh consumer? (0 = serving/key issue)
try {
  for (let i = 0; i < 20 && (!peer.repo || peer.repo.core.length < 2); i++) { await new Promise(r => setTimeout(r, 300)) }
  const recs = await peer.listRecords('app.bsky.feed.post')
  console.log('LISTRECORDS app.bsky.feed.post:', recs.length, recs.slice(0, 4).map(r => (r.uri || r.key || '').split('/').pop()))
} catch (e) { console.log('LISTRECORDS error:', e.message) }

let commits = 0, syncs = 0, other = 0, total = 0
const writeBytes = () => { try { const s = fs.readFileSync('/proc/self/io', 'utf8'); const m = s.match(/write_bytes:\s*(\d+)/); return m ? Number(m[1]) : 0 } catch { return 0 } }
let lastW = writeBytes()
let ticks = 0
const timer = setInterval(() => {
  const w = writeBytes(); const dW = ((w - lastW) / 1e6).toFixed(1); lastW = w
  console.log(JSON.stringify({
    t: ++ticks * 3 + 's', frames_3s: { commits, syncs, other, total }, write_MB_per_3s: dW,
    firehoseLen: peer.seq ? peer.seq.core.length : 0, repoLen: peer.repo ? peer.repo.core.length : 0,
    swarmConns: swarm.connections.size
  }))
  commits = syncs = other = total = 0
  if (ticks >= 12) { console.log('DONE'); clearInterval(timer); swarm.destroy(); process.exit(0) }
}, 3000)

const from = peer.seq ? peer.seq.core.length : 0
console.log('subscribing from', from)
;(async () => {
  for await (const { bytes } of peer.subscribeRepos(from)) {
    total++
    const d = decodeFrame(bytes)
    if (!d) { other++; continue }
    if (d.t === '#commit') commits++
    else if (d.t === '#sync') syncs++
    else other++
  }
})().catch((e) => console.error('loop err', e.message))