chadtmiller.com/pds.js

Switch repository

Loading…

#!/usr/bin/env node
/**
 * Server-side cost of a smart HTTP clone, measured against a local PDS.
 *
 * The upload-pack response is assembled in the server process, so the peak
 * this reports is what a Cloudflare isolate must hold. An isolate gets 128
 * MB, which is the number the ratio column is read against.
 *
 * Repository sizes are random bytes, so packs do not compress and the
 * content size is the pack size.
 *
 * --gc collects on every sample, which separates memory the response holds
 * from memory it has dropped but V8 has not swept.
 *
 * Run from the workspace root after `pnpm install`:
 *
 *   node packages/git/bench/clone-memory.mjs --sizes 8,24,48
 */

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 : process.argv[i + 1];
}

const SIZES = String(arg('sizes', '8,24,48'))
  .split(',')
  .map((s) => Number(s.trim()))
  .filter((n) => Number.isFinite(n) && n > 0);
const COMMITS = Number(arg('commits', 6));
const PORT = Number(arg('port', 2490));
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 workDir = mkdtempSync(join(tmpdir(), 'pdsjs-git-clone-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);

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',
    maxBuffer: 1 << 28,
  });
}

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

/**
 * Chain policies differ in what a clone must merge. A repacked chain is one
 * bundle; an unmerged chain is one bundle per push.
 */
const POLICIES = [
  { label: 'single bundle', factor: '2', threshold: '2' },
  { label: 'chain of 6   ', factor: '0', threshold: '100' },
];

async function measure(sizeMb, policy) {
  const rkey = `bench-${sizeMb}-${policy.threshold}`;
  const env = {
    ...baseEnv,
    ATPROTO_GIT_COMPACT_FACTOR: policy.factor,
    ATPROTO_GIT_REPACK_THRESHOLD: policy.threshold,
  };
  const repoDir = join(workDir, rkey);
  mkdirSync(repoDir);
  await runGit(repoDir, ['init', '-b', 'main'], env);
  await runGit(
    repoDir,
    ['remote', 'add', 'origin', `atproto://${DID}/${rkey}`],
    env,
  );
  const perCommit = Math.floor((sizeMb * 1024 * 1024) / COMMITS);
  for (let i = 0; i < COMMITS; i++) {
    writeFileSync(join(repoDir, `data-${i}.bin`), randomBytes(perCommit));
    await runGit(repoDir, ['add', '.'], env);
    await runGit(repoDir, ['commit', '-m', `commit ${i}`], env);
    await runGit(repoDir, ['push', 'origin', 'main'], env);
  }

  const params = new URLSearchParams({
    repo: DID,
    collection: GIT_REPO_COLLECTION,
    rkey,
  });
  const res = await fetch(`${BASE}/xrpc/com.atproto.repo.getRecord?${params}`);
  const record = (await res.json()).value;
  const packBytes = record.bundles.reduce(
    (sum, b) => sum + b.parts.reduce((t, p) => t + p.size, 0),
    0,
  );

  if (global.gc) global.gc();
  await new Promise((resolve) => setTimeout(resolve, 100));
  const before = process.memoryUsage();
  let peakBuffers = before.arrayBuffers;
  let peakRss = before.rss;
  const collect = process.argv.includes('--gc');
  const sampler = setInterval(() => {
    if (collect && global.gc) global.gc();
    const m = process.memoryUsage();
    peakBuffers = Math.max(peakBuffers, m.arrayBuffers);
    peakRss = Math.max(peakRss, m.rss);
  }, 20);
  const t0 = performance.now();
  let failure = null;
  try {
    await runGit(
      workDir,
      ['clone', `${BASE}/git/${DID}/${rkey}`, join(workDir, `${rkey}-clone`)],
      baseEnv,
    );
  } catch (err) {
    failure = err.stderr?.trim().split('\n').pop() ?? String(err);
  }
  const secs = (performance.now() - t0) / 1000;
  clearInterval(sampler);

  return {
    sizeMb,
    policy: policy.label,
    bundles: record.bundles.length,
    packBytes,
    buffers: peakBuffers - before.arrayBuffers,
    rss: peakRss - before.rss,
    secs,
    failure,
  };
}

console.log(
  `clone cost by repository size, ${COMMITS} commits each, sampling the server process every 20 ms\n`,
);
console.log(
  '  policy          size    bundles   peak buffers   peak rss    x pack   clone',
);
for (const sizeMb of SIZES) {
  for (const policy of POLICIES) {
    const r = await measure(sizeMb, policy);
    const ratio = (r.buffers / r.packBytes).toFixed(1);
    console.log(
      `  ${r.policy}  ${String(sizeMb).padStart(3)} MB   ${String(r.bundles).padStart(4)}    ${fmt(r.buffers).padStart(9)}    ${fmt(r.rss).padStart(9)}   ${ratio.padStart(5)}x   ${r.failure ? `FAILED ${r.failure}` : `${r.secs.toFixed(1)}s`}`,
    );
  }
}

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