pear-git-v3 / bench / upload-rss.js

/**
 * E2E bench for /api/upload — measures RSS Δ + corestore growth for a
 * large attachment uploaded through a fresh pear-git-v3 daemon.
 *
 * Issue #18 done-when:
 *   2 GB attachment → < 200 MB RSS Δ, < 100 MB corestore growth on the daemon.
 *
 * Run: node bench/upload-rss.js [src-file]
 *      defaults to /tmp/stress-2gb.bin
 */

import { mkdtempSync, statSync, readdirSync, createReadStream } from 'fs'
import { join } from 'path'
import { tmpdir } from 'os'
import http from 'http'
import b4a from 'b4a'

import { startPearGit } from '../index.js'
import { signTestEnvelope } from '../lib/auth/envelope.js'

const SRC = process.argv[2] || '/tmp/stress-2gb.bin'
const SIZE = statSync(SRC).size
console.log('source:', SRC, '(' + (SIZE / 1024 / 1024).toFixed(0) + ' MB)\n')

function du (dir) {
  let total = 0
  let entries
  try { entries = readdirSync(dir, { withFileTypes: true }) } catch { return 0 }
  for (const e of entries) {
    const p = join(dir, e.name)
    if (e.isDirectory()) total += du(p)
    else if (e.isFile()) {
      try { total += statSync(p).size } catch {}
    }
  }
  return total
}

function postJson (url, body, headers = {}) {
  return new Promise((resolve, reject) => {
    const json = JSON.stringify(body || {})
    const u = new URL(url)
    const req = http.request({
      method: 'POST', hostname: u.hostname, port: u.port, path: u.pathname,
      headers: { 'Content-Type': 'application/json', 'Content-Length': b4a.byteLength(json), ...headers }
    }, (res) => {
      const parts = []
      res.on('data', c => parts.push(c))
      res.on('end', () => {
        const text = b4a.concat(parts).toString('utf8')
        let parsed; try { parsed = JSON.parse(text) } catch { parsed = text }
        resolve({ status: res.statusCode, body: parsed, setCookie: res.headers['set-cookie'] })
      })
      res.on('error', reject)
    })
    req.on('error', reject)
    req.write(json); req.end()
  })
}

async function loginAsHttpUser (port) {
  const { an } = await import('anproto')
  const kp = await an.gen()
  const env = await signTestEnvelope({ keypair: kp, method: 'whoami', params: {} })
  const res = await postJson('http://127.0.0.1:' + port + '/api/rpc', env)
  if (res.status !== 200 || !res.body || !res.body.ok) {
    throw new Error('login failed: ' + res.status)
  }
  const setCookie = res.setCookie
  return Array.isArray(setCookie) ? setCookie[0].split(';')[0] : setCookie.split(';')[0]
}

function streamUpload (port, srcPath, srcSize, cookie) {
  return new Promise((resolve, reject) => {
    const req = http.request({
      method: 'POST', hostname: '127.0.0.1', port,
      path: '/api/upload',
      headers: {
        'Content-Type': 'application/octet-stream',
        'Content-Length': srcSize,
        Cookie: cookie
      }
    }, (res) => {
      const parts = []
      res.on('data', c => parts.push(c))
      res.on('end', () => {
        let body; try { body = JSON.parse(b4a.concat(parts).toString('utf8')) } catch { body = null }
        resolve({ status: res.statusCode, body })
      })
      res.on('error', reject)
    })
    req.on('error', reject)
    createReadStream(srcPath).pipe(req)
  })
}

async function main () {
  const storage = mkdtempSync(join(tmpdir(), 'pgv3-upload-bench-'))
  const sock = join(mkdtempSync(join(tmpdir(), 'pgv3-upload-bench-sock-')), 'svc.sock')

  console.log('booting daemon at', storage)
  const node = await startPearGit({
    storage, uds: sock, http: { port: 0 },
    swarm: false, autoImportSSH: false
  })
  const port = node.svc.http.server.address().port

  const cookie = await loginAsHttpUser(port)

  const corestoreDir = join(storage, 'store')
  const baselineStore = du(corestoreDir)
  const baselineRss = process.memoryUsage().rss

  let peakRss = baselineRss
  let peakStore = baselineStore
  let stop = false
  const sampler = (async () => {
    while (!stop) {
      const r = process.memoryUsage().rss
      if (r > peakRss) peakRss = r
      const s = du(corestoreDir)
      if (s > peakStore) peakStore = s
      await new Promise(res => setTimeout(res, 200))
    }
  })()

  console.log('starting upload...')
  const t0 = Date.now()
  const result = await streamUpload(port, SRC, SIZE, cookie)
  const elapsed = Date.now() - t0
  stop = true
  await sampler

  const finalStore = du(corestoreDir)
  const spoolSize = du(join(storage, 'attachments-spool'))

  await node.close()

  const fmtMB = (b) => (b / 1024 / 1024).toFixed(0).padStart(5) + ' MB'
  console.log()
  console.log('result:        ' + result.status + ' ' + (result.body && result.body.hash ? result.body.hash.slice(0, 12) + '…' : ''))
  console.log('elapsed:       ' + (elapsed / 1000).toFixed(2) + ' s')
  console.log('peak RSS Δ:    ' + fmtMB(peakRss - baselineRss))
  console.log('peak store:    ' + fmtMB(peakStore - baselineStore))
  console.log('final store Δ: ' + fmtMB(finalStore - baselineStore))
  console.log('spool dir:     ' + fmtMB(spoolSize))
  console.log()
  console.log('thresholds (issue #18):')
  console.log('  RSS Δ < 200 MB:        ' + ((peakRss - baselineRss) < 200 * 1024 * 1024 ? 'PASS' : 'FAIL'))
  console.log('  store Δ < 100 MB:      ' + ((finalStore - baselineStore) < 100 * 1024 * 1024 ? 'PASS' : 'FAIL'))

  process.exit(0)
}

main().catch(err => { console.error(err); process.exit(1) })