feat: add benchmarks and optimize acquire_job

BENCHMARKS.md

@@ -0,0 +1,45 @@
+# Benchmarks
+
+## Running
+
+```bash
+bun run bench          # run all benchmarks
+bun run bench:add      # add-jobs throughput only
+bun run bench:acquire  # acquire-job latency only
+```
+
+Requires Postgres running (`bun run docker:up`).
+
+## add-jobs (2026-03-08)
+
+- **Pollocks version**: 1.0.0
+- **Runtime**: Bun 1.3.9
+- **CPU**: Apple M1 Pro (10 cores)
+- **Memory**: 16GB
+- **OS**: Darwin 24.6.0 (arm64)
+
+### Add Jobs Throughput
+
+| Batch Size | Jobs/sec | Total Jobs | Duration |
+|------------|----------|------------|----------|
+| 1 | 2,838 | 14,192 | 5.0s |
+| 10 | 17,956 | 89,820 | 5.0s |
+| 100 | 62,585 | 313,000 | 5.0s |
+| 1000 | 94,510 | 473,000 | 5.0s |
+
+## acquire-job (2026-03-08)
+
+- **Pollocks version**: 1.0.0
+- **Runtime**: Bun 1.3.9
+- **CPU**: Apple M1 Pro (10 cores)
+- **Memory**: 16GB
+- **OS**: Darwin 24.6.0 (arm64)
+
+### Acquire Job Latency
+
+| Table Size | Avg | Min | P50 | P99 | Max |
+|------------|-----|-----|-----|-----|-----|
+| 10,000 | 0.63ms | 0.45ms | 0.57ms | 1.44ms | 1.77ms |
+| 100,000 | 0.64ms | 0.44ms | 0.57ms | 1.07ms | 1.56ms |
+| 1,000,000 | 0.53ms | 0.43ms | 0.49ms | 0.96ms | 0.98ms |
+

package.json

@@ -38,6 +38,9 @@
     "migration": "bun run scripts/new-migration.ts",
     "docker:up": "docker-compose up -d",
     "docker:down": "docker-compose down",
+    "bench:add": "bun run src/benchmarks/add-jobs.ts",
+    "bench:acquire": "bun run src/benchmarks/acquire-job.ts",
+    "bench": "bun run bench:add && bun run bench:acquire",
     "recipe:basic": "bun run recipes/basic/run.ts",
     "recipe:nestjs": "bun run recipes/nestjs/run.ts",
     "prepare": "husky"

src/benchmarks/acquire-job.ts

@@ -0,0 +1,137 @@
+import { Pool } from "pg";
+import { Tools } from "../tools.ts";
+import { getSystemInfo, formatNumber, appendResults } from "./utils.ts";
+
+const DATABASE_URL =
+  process.env.DATABASE_URL ??
+  "postgres://postgres:postgres@localhost:5432/pollocks";
+
+const pool = new Pool({
+  connectionString: DATABASE_URL,
+  max: 10,
+  allowExitOnIdle: true,
+});
+
+const tools = new Tools(pool);
+await tools.migrate();
+
+// Clean up any leftover jobs from previous runs
+await pool.query("DELETE FROM jobs");
+
+const TABLE_SIZES = [10_000, 100_000, 1_000_000];
+const ITERATIONS = 100;
+const BATCH_INSERT = 1000;
+
+interface Result {
+  tableSize: number;
+  avgMs: number;
+  minMs: number;
+  maxMs: number;
+  p50Ms: number;
+  p99Ms: number;
+}
+
+async function seedJobs(count: number): Promise<void> {
+  const runAfter = new Date();
+  let inserted = 0;
+
+  while (inserted < count) {
+    const batchSize = Math.min(BATCH_INSERT, count - inserted);
+    const inputs = Array.from({ length: batchSize }, (_, i) => ({
+      pattern: "benchmark.seed",
+      payload: { index: inserted + i },
+      runAfter,
+    }));
+    await tools.addJobs(inputs);
+    inserted += batchSize;
+
+    if (inserted % 50_000 === 0) {
+      console.log(`  Seeded ${formatNumber(inserted)}/${formatNumber(count)} jobs...`);
+    }
+  }
+}
+
+function percentile(sorted: number[], p: number): number {
+  const idx = Math.ceil((p / 100) * sorted.length) - 1;
+  return sorted[Math.max(0, idx)]!;
+}
+
+const results: Result[] = [];
+
+for (const tableSize of TABLE_SIZES) {
+  console.log(`\nPreparing ${formatNumber(tableSize)} jobs...`);
+
+  await pool.query("DELETE FROM jobs");
+  await seedJobs(tableSize);
+
+  // Let Postgres update statistics and flush dirty buffers
+  await pool.query("VACUUM ANALYZE jobs");
+
+  // Verify count
+  const countResult = await pool.query("SELECT count(*) FROM jobs");
+  console.log(`  Table has ${formatNumber(Number(countResult.rows[0].count))} jobs`);
+
+  // Add one acquirable job per iteration, then acquire and complete it
+  const latencies: number[] = [];
+
+  for (let i = 0; i < ITERATIONS; i++) {
+    // Add a job that's immediately acquirable
+    await tools.addJob({
+      pattern: "benchmark.acquire",
+      payload: { iteration: i },
+      runAfter: new Date(),
+    });
+
+    const start = performance.now();
+    const job = await tools.acquireJob("benchmark-worker", ["benchmark.acquire"]);
+    const elapsed = performance.now() - start;
+
+    latencies.push(elapsed);
+
+    if (job) {
+      await tools.completeJob(job.id);
+    }
+  }
+
+  latencies.sort((a, b) => a - b);
+
+  const avg = latencies.reduce((s, v) => s + v, 0) / latencies.length;
+
+  results.push({
+    tableSize,
+    avgMs: Number(avg.toFixed(2)),
+    minMs: Number(latencies[0]!.toFixed(2)),
+    maxMs: Number(latencies[latencies.length - 1]!.toFixed(2)),
+    p50Ms: Number(percentile(latencies, 50).toFixed(2)),
+    p99Ms: Number(percentile(latencies, 99).toFixed(2)),
+  });
+
+  console.log(
+    `  ${formatNumber(tableSize)} jobs: avg=${avg.toFixed(2)}ms p50=${percentile(latencies, 50).toFixed(2)}ms p99=${percentile(latencies, 99).toFixed(2)}ms`,
+  );
+}
+
+// Clean up
+await pool.query("DELETE FROM jobs");
+
+const sysInfo = await getSystemInfo();
+
+await appendResults("acquire-job", sysInfo, (lines) => {
+  lines.push("");
+  lines.push("### Acquire Job Latency");
+  lines.push("");
+  lines.push(
+    "| Table Size | Avg | Min | P50 | P99 | Max |",
+  );
+  lines.push(
+    "|------------|-----|-----|-----|-----|-----|",
+  );
+  for (const r of results) {
+    lines.push(
+      `| ${formatNumber(r.tableSize)} | ${r.avgMs}ms | ${r.minMs}ms | ${r.p50Ms}ms | ${r.p99Ms}ms | ${r.maxMs}ms |`,
+    );
+  }
+});
+
+await pool.end();
+console.log("\nResults written to BENCHMARKS.md");

src/benchmarks/add-jobs.ts

@@ -0,0 +1,91 @@
+import { Pool } from "pg";
+import { Tools } from "../tools.ts";
+import { getSystemInfo, formatNumber, appendResults } from "./utils.ts";
+
+const DATABASE_URL =
+  process.env.DATABASE_URL ??
+  "postgres://postgres:postgres@localhost:5432/pollocks";
+
+const pool = new Pool({
+  connectionString: DATABASE_URL,
+  max: 10,
+  allowExitOnIdle: true,
+});
+
+const tools = new Tools(pool);
+await tools.migrate();
+
+// Clean up any leftover jobs from previous runs
+await pool.query("DELETE FROM jobs");
+
+const BATCH_SIZES = [1, 10, 100, 1000];
+const DURATION_MS = 5_000;
+
+interface Result {
+  batchSize: number;
+  totalJobs: number;
+  durationMs: number;
+  jobsPerSecond: number;
+}
+
+const results: Result[] = [];
+
+for (const batchSize of BATCH_SIZES) {
+  await pool.query("DELETE FROM jobs");
+
+  const input = Array.from({ length: batchSize }, (_, i) => ({
+    pattern: "benchmark.add",
+    payload: { index: i },
+  }));
+
+  let totalJobs = 0;
+  const start = performance.now();
+
+  while (performance.now() - start < DURATION_MS) {
+    if (batchSize === 1) {
+      await tools.addJob(input[0]!);
+    } else {
+      await tools.addJobs(input);
+    }
+    totalJobs += batchSize;
+  }
+
+  const elapsed = performance.now() - start;
+  const jobsPerSecond = (totalJobs / elapsed) * 1000;
+
+  results.push({
+    batchSize,
+    totalJobs,
+    durationMs: Math.round(elapsed),
+    jobsPerSecond: Math.round(jobsPerSecond),
+  });
+
+  console.log(
+    `Batch size ${batchSize}: ${formatNumber(Math.round(jobsPerSecond))} jobs/sec (${formatNumber(totalJobs)} jobs in ${Math.round(elapsed)}ms)`,
+  );
+}
+
+// Clean up
+await pool.query("DELETE FROM jobs");
+
+const sysInfo = await getSystemInfo();
+
+await appendResults("add-jobs", sysInfo, (lines) => {
+  lines.push("");
+  lines.push("### Add Jobs Throughput");
+  lines.push("");
+  lines.push(
+    "| Batch Size | Jobs/sec | Total Jobs | Duration |",
+  );
+  lines.push(
+    "|------------|----------|------------|----------|",
+  );
+  for (const r of results) {
+    lines.push(
+      `| ${r.batchSize} | ${formatNumber(r.jobsPerSecond)} | ${formatNumber(r.totalJobs)} | ${(r.durationMs / 1000).toFixed(1)}s |`,
+    );
+  }
+});
+
+await pool.end();
+console.log("\nResults written to BENCHMARKS.md");

src/benchmarks/utils.ts

@@ -0,0 +1,108 @@
+import fs from "fs/promises";
+import path from "path";
+import os from "os";
+
+const BENCHMARKS_FILE = path.join(import.meta.dirname, "../../BENCHMARKS.md");
+
+const HEADER = `# Benchmarks
+
+## Running
+
+\`\`\`bash
+bun run bench          # run all benchmarks
+bun run bench:add      # add-jobs throughput only
+bun run bench:acquire  # acquire-job latency only
+\`\`\`
+
+Requires Postgres running (\`bun run docker:up\`).
+
+`;
+
+export interface SystemInfo {
+  platform: string;
+  arch: string;
+  cpuModel: string;
+  cpuCores: number;
+  memoryGb: number;
+  nodeVersion: string;
+  bunVersion: string;
+  pollocksVersion: string;
+}
+
+export async function getSystemInfo(): Promise<SystemInfo> {
+  const cpus = os.cpus();
+  const pkg = JSON.parse(
+    await fs.readFile(
+      path.join(import.meta.dirname, "../../package.json"),
+      "utf-8",
+    ),
+  );
+
+  let bunVersion = "unknown";
+  try {
+    const proc = Bun.spawnSync(["bun", "--version"]);
+    bunVersion = proc.stdout.toString().trim();
+  } catch {}
+
+  return {
+    platform: `${os.type()} ${os.release()}`,
+    arch: os.arch(),
+    cpuModel: cpus[0]?.model ?? "unknown",
+    cpuCores: cpus.length,
+    memoryGb: Math.round(os.totalmem() / 1024 / 1024 / 1024),
+    nodeVersion: process.version,
+    bunVersion,
+    pollocksVersion: pkg.version,
+  };
+}
+
+export function formatNumber(n: number): string {
+  return n.toLocaleString("en-US");
+}
+
+export async function appendResults(
+  benchmarkId: string,
+  sysInfo: SystemInfo,
+  buildTable: (lines: string[]) => void,
+): Promise<void> {
+  let existing = "";
+  try {
+    existing = await fs.readFile(BENCHMARKS_FILE, "utf-8");
+  } catch {}
+
+  // Ensure the file starts with the header
+  if (!existing.includes("## Running")) {
+    existing = HEADER;
+  }
+
+  // Build the new section
+  const sectionLines: string[] = [];
+  const timestamp = new Date().toISOString().split("T")[0];
+
+  sectionLines.push(`## ${benchmarkId} (${timestamp})`);
+  sectionLines.push("");
+  sectionLines.push(`- **Pollocks version**: ${sysInfo.pollocksVersion}`);
+  sectionLines.push(`- **Runtime**: Bun ${sysInfo.bunVersion}`);
+  sectionLines.push(`- **CPU**: ${sysInfo.cpuModel} (${sysInfo.cpuCores} cores)`);
+  sectionLines.push(`- **Memory**: ${sysInfo.memoryGb}GB`);
+  sectionLines.push(`- **OS**: ${sysInfo.platform} (${sysInfo.arch})`);
+
+  buildTable(sectionLines);
+
+  sectionLines.push("");
+
+  const sectionContent = sectionLines.join("\n");
+
+  // Replace existing section (match by benchmarkId, ignoring date) or append
+  const sectionRegex = new RegExp(
+    `## ${benchmarkId.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")} \\([^)]+\\)[\\s\\S]*?(?=\\n## |$)`,
+  );
+
+  if (sectionRegex.test(existing)) {
+    existing = existing.replace(sectionRegex, sectionContent);
+  } else {
+    existing = existing.trimEnd() + "\n\n" + sectionContent + "\n";
+  }
+
+  await fs.writeFile(BENCHMARKS_FILE, existing);
+}