chadtmiller.com/pds.js

Switch repository

Loading…

#!/usr/bin/env node
/**
 * Push-cost and read-path benchmark for @pdsjs/git against a local PDS.
 *
 * Pushes the same synthetic history under two chain policies and reports
 * uploaded bytes, worst-case push, chain length, and wall time. Then it
 * times smart HTTP reads with and without ETag revalidation, including the
 * cold-store case a fresh Workers isolate hits.
 *
 * Run from the workspace root after `pnpm install`:
 *
 *   node packages/git/bench/push-costs.mjs --commits 60 --bytes 8192
 */

import { execFile } from 'node:child_process';
import { randomBytes } from 'node:crypto';
import {
  chmodSync,
  mkdirSync,
  mkdtempSync,
  rmSync,
  writeFileSync,
} from 'node:fs';
import { tmpdir } from 'node:os';
import { dirname, join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { promisify } from 'node:util';
import { defineLexicon } from '@bigmoves/lexicon';
import { GIT_REPO_COLLECTION, gitRepoLexicon } from '@pdsjs/git';
import { LexiconResolver } from '@pdsjs/lexicon-resolver';
import { createServer } from '@pdsjs/node';

const execFileAsync = promisify(execFile);

function arg(name, fallback) {
  const i = process.argv.indexOf(`--${name}`);
  return i === -1 ? fallback : Number(process.argv[i + 1]);
}

const COMMITS = arg('commits', 60);
const FILE_BYTES = arg('bytes', 8 * 1024);
const PORT = arg('port', 2489);
const BASE = `http://127.0.0.1:${PORT}`;
const PASSWORD = 'bench-password';
const chars = 'abcdefghijklmnopqrstuvwxyz234567';
const DID = `did:plc:${Array.from(randomBytes(24))
  .map((b) => chars[b % 32])
  .join('')}`;

const CONFIGS = [
  {
    label: 'no merges, threshold 16',
    rkey: 'bench-old',
    env: {
      ATPROTO_GIT_COMPACT_FACTOR: '0',
      ATPROTO_GIT_REPACK_THRESHOLD: '16',
    },
  },
  {
    label: 'geometric factor 2',
    rkey: 'bench-geo',
    env: {
      ATPROTO_GIT_COMPACT_FACTOR: '2',
      ATPROTO_GIT_REPACK_THRESHOLD: '16',
    },
  },
];

const workDir = mkdtempSync(join(tmpdir(), 'pdsjs-git-bench-'));
const server = await createServer({
  port: PORT,
  dbPath: join(workDir, 'pds.db'),
  blobsDir: join(workDir, 'blobs'),
  jwtSecret: 'bench-secret',
  hostname: `127.0.0.1:${PORT}`,
  password: PASSWORD,
  lexiconResolver: new LexiconResolver({
    schemas: [defineLexicon(gitRepoLexicon)],
  }),
  experimental: { git: { http: true } },
});
await server.listen();
const init = await fetch(`${BASE}/init?did=${DID}`, {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    did: DID,
    privateKey: randomBytes(32).toString('hex'),
    handle: 'bench.test',
    password: PASSWORD,
  }),
});
if (!init.ok) throw new Error('account init failed');

const binDir = join(workDir, 'bin');
mkdirSync(binDir);
const cliPath = join(
  dirname(fileURLToPath(import.meta.url)),
  '..',
  'src',
  'cli.js',
);
writeFileSync(
  join(binDir, 'git-remote-atproto'),
  `#!/bin/sh\nexec node ${cliPath} "$@"\n`,
);
chmodSync(join(binDir, 'git-remote-atproto'), 0o755);

// The same content for every run.
const payloads = Array.from({ length: COMMITS }, () => randomBytes(FILE_BYTES));

const baseEnv = {
  ...process.env,
  PATH: `${binDir}:${process.env.PATH}`,
  ATPROTO_GIT_SERVICE: BASE,
  ATPROTO_GIT_IDENTIFIER: DID,
  ATPROTO_GIT_PASSWORD: PASSWORD,
  GIT_TERMINAL_PROMPT: '0',
  GIT_AUTHOR_NAME: 'Bench',
  GIT_AUTHOR_EMAIL: 'bench@example.com',
  GIT_COMMITTER_NAME: 'Bench',
  GIT_COMMITTER_EMAIL: 'bench@example.com',
};

function runGit(cwd, args, env) {
  return execFileAsync('git', args, { cwd, env, encoding: 'utf8' });
}

async function loadRecord(rkey) {
  const params = new URLSearchParams({
    repo: DID,
    collection: GIT_REPO_COLLECTION,
    rkey,
  });
  const res = await fetch(`${BASE}/xrpc/com.atproto.repo.getRecord?${params}`);
  const body = await res.json();
  return body.value;
}

const mb = (n) => `${(n / 1024 / 1024).toFixed(2)} MB`;

async function benchPushes({ label, rkey, env: extraEnv }) {
  const env = { ...baseEnv, ...extraEnv };
  const repoDir = join(workDir, rkey);
  mkdirSync(repoDir);
  await runGit(repoDir, ['init', '-b', 'main'], env);
  await runGit(
    repoDir,
    ['remote', 'add', 'origin', `atproto://${DID}/${rkey}`],
    env,
  );
  let total = 0;
  let maxPush = 0;
  let maxChain = 0;
  const checkpoints = [];
  const step = Math.max(1, Math.floor(COMMITS / 4));
  const t0 = performance.now();
  for (let i = 0; i < COMMITS; i++) {
    writeFileSync(join(repoDir, `data-${i}.bin`), payloads[i]);
    await runGit(repoDir, ['add', '.'], env);
    await runGit(repoDir, ['commit', '-m', `commit ${i}`], env);
    const { stderr } = await runGit(repoDir, ['push', 'origin', 'main'], env);
    let pushBytes = 0;
    for (const m of stderr.matchAll(/Uploading bundle \((\d+) bytes/g)) {
      pushBytes += Number(m[1]);
    }
    total += pushBytes;
    maxPush = Math.max(maxPush, pushBytes);
    if ((i + 1) % step === 0) checkpoints.push(`${i + 1}:${mb(total)}`);
    const record = await loadRecord(rkey);
    maxChain = Math.max(maxChain, record.bundles.length);
  }
  const secs = (performance.now() - t0) / 1000;
  const record = await loadRecord(rkey);
  const heads = record.bundles.reduce((sum, b) => sum + b.heads.length, 0);
  console.log(`\n${label}`);
  console.log(`  total uploaded:        ${mb(total)}`);
  console.log(`  worst single push:     ${mb(maxPush)}`);
  console.log(`  cumulative at push:    ${checkpoints.join('  ')}`);
  console.log(
    `  chain max/final:       ${maxChain}/${record.bundles.length} bundles, ${heads} recorded heads`,
  );
  console.log(
    `  wall time:             ${secs.toFixed(1)}s (${((secs / COMMITS) * 1000).toFixed(0)} ms/push)`,
  );
}

async function timeReads(label, url, headers) {
  await fetch(url, { headers }).then((r) => r.arrayBuffer());
  const times = [];
  let status = 0;
  let bytes = 0;
  for (let i = 0; i < 40; i++) {
    const t0 = performance.now();
    const res = await fetch(url, { headers });
    const body = await res.arrayBuffer();
    times.push(performance.now() - t0);
    status = res.status;
    bytes = body.byteLength;
  }
  times.sort((a, b) => a - b);
  console.log(
    `  ${label}: status ${status}, body ${bytes} B, median ${times[20].toFixed(1)} ms`,
  );
}

// Bust the single-entry browser store cache between reads, the way a fresh
// Workers isolate arrives with no store at all.
async function timeColdReads(label, url, otherUrl, headers) {
  const times = [];
  let status = 0;
  for (let i = 0; i < 20; i++) {
    await fetch(otherUrl).then((r) => r.arrayBuffer());
    const t0 = performance.now();
    const res = await fetch(url, { headers });
    await res.arrayBuffer();
    times.push(performance.now() - t0);
    status = res.status;
  }
  times.sort((a, b) => a - b);
  console.log(
    `  ${label}: status ${status}, median ${times[10].toFixed(1)} ms`,
  );
}

console.log(
  `pushing ${COMMITS} commits of ${FILE_BYTES} random bytes each, twice`,
);
for (const config of CONFIGS) await benchPushes(config);

console.log('\nread path (repo from the geometric run)');
const RAW = `${BASE}/git/${DID}/bench-geo/raw/main/data-1.bin`;
const OTHER = `${BASE}/git/${DID}/bench-old/raw/main/data-1.bin`;
const REFS = `${BASE}/git/${DID}/bench-geo/info/refs?service=git-upload-pack`;
const etag = (await fetch(RAW)).headers.get('ETag');
const conditional = { 'If-None-Match': etag ?? '' };
await timeReads('raw 200, warm store  ', RAW);
await timeReads('raw 304, warm store  ', RAW, conditional);
await timeColdReads('raw 200, cold store  ', RAW, OTHER);
await timeColdReads('raw 304, cold store  ', RAW, OTHER, conditional);
await timeReads('info/refs 200        ', REFS);
await timeReads('info/refs 304        ', REFS, conditional);

await server.close();
rmSync(workDir, { recursive: true, force: true });