diff --git a/BENCHMARKS.md b/BENCHMARKS.md new file mode 100644 index 0000000..ecbc200 --- /dev/null +++ b/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 | + diff --git a/package.json b/package.json index 2c345e3..e36ec01 100644 --- a/package.json +++ b/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" diff --git a/src/benchmarks/acquire-job.ts b/src/benchmarks/acquire-job.ts new file mode 100644 index 0000000..e3ec1f4 --- /dev/null +++ b/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 { + 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"); diff --git a/src/benchmarks/add-jobs.ts b/src/benchmarks/add-jobs.ts new file mode 100644 index 0000000..a2ef1ea --- /dev/null +++ b/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"); diff --git a/src/benchmarks/utils.ts b/src/benchmarks/utils.ts new file mode 100644 index 0000000..129dd9d --- /dev/null +++ b/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 { + 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 { + 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); +} diff --git a/src/migrations/013_add_acquire_indexes.sql b/src/migrations/013_add_acquire_indexes.sql new file mode 100644 index 0000000..1345336 --- /dev/null +++ b/src/migrations/013_add_acquire_indexes.sql @@ -0,0 +1,12 @@ +-- Index to support acquire_job/acquire_jobs queries. +-- The acquire query filters on: run_after <= now(), locked_until IS NULL or expired, +-- attempts < max_attempts, and optionally pattern. It orders by run_after. +-- +-- A composite index on (run_after) lets Postgres walk the index in order and +-- stop at LIMIT 1 after finding the first qualifying row, avoiding a full +-- table scan + sort. +CREATE INDEX idx_jobs_run_after ON jobs (run_after); + +-- Pattern-filtered acquires benefit from a dedicated index so Postgres can +-- narrow to matching patterns first, then scan by run_after within that subset. +CREATE INDEX idx_jobs_pattern_run_after ON jobs (pattern, run_after); diff --git a/src/migrations/014_split_acquire_for_index_usage.sql b/src/migrations/014_split_acquire_for_index_usage.sql new file mode 100644 index 0000000..ceedd00 --- /dev/null +++ b/src/migrations/014_split_acquire_for_index_usage.sql @@ -0,0 +1,137 @@ +-- Rewrite acquire_job and acquire_jobs as PL/pgSQL with separate code paths +-- so Postgres can use idx_jobs_pattern_run_after when patterns are provided +-- and idx_jobs_run_after when they are not. +-- +-- The previous SQL-language functions used (p_patterns IS NULL OR pattern = ANY(p_patterns)) +-- which prevented the planner from pushing the pattern condition into an index scan. +-- +-- We use EXECUTE (dynamic SQL) instead of RETURN QUERY to prevent PL/pgSQL +-- from caching a generic plan that ignores the composite index. + +DROP FUNCTION IF EXISTS acquire_job(text, text[]); + +CREATE FUNCTION acquire_job(p_locked_by text DEFAULT NULL, p_patterns text[] DEFAULT NULL) +RETURNS SETOF jobs +LANGUAGE plpgsql +AS $$ +BEGIN + IF p_patterns IS NOT NULL THEN + RETURN QUERY EXECUTE + 'WITH selected AS ( + SELECT id, lock_for + FROM jobs + WHERE pattern = ANY($1) + AND run_after <= now() + AND (locked_until IS NULL OR locked_until < now()) + AND attempts < max_attempts + ORDER BY run_after + FOR UPDATE SKIP LOCKED + LIMIT 1 + ), + updated AS ( + UPDATE jobs j + SET + updated_at = now(), + locked_by = coalesce($2, ''administrator''), + locked_until = now() + s.lock_for * interval ''1 second'', + locked_at = now(), + attempts = j.attempts + 1 + FROM selected s + WHERE j.id = s.id + RETURNING j.* + ) + SELECT * FROM updated' + USING p_patterns, p_locked_by; + ELSE + RETURN QUERY EXECUTE + 'WITH selected AS ( + SELECT id, lock_for + FROM jobs + WHERE run_after <= now() + AND (locked_until IS NULL OR locked_until < now()) + AND attempts < max_attempts + ORDER BY run_after + FOR UPDATE SKIP LOCKED + LIMIT 1 + ), + updated AS ( + UPDATE jobs j + SET + updated_at = now(), + locked_by = coalesce($1, ''administrator''), + locked_until = now() + s.lock_for * interval ''1 second'', + locked_at = now(), + attempts = j.attempts + 1 + FROM selected s + WHERE j.id = s.id + RETURNING j.* + ) + SELECT * FROM updated' + USING p_locked_by; + END IF; +END; +$$; + +DROP FUNCTION IF EXISTS acquire_jobs(integer, text, text[]); + +CREATE FUNCTION acquire_jobs(p_max integer, p_locked_by text DEFAULT NULL, p_patterns text[] DEFAULT NULL) +RETURNS SETOF jobs +LANGUAGE plpgsql +AS $$ +BEGIN + IF p_patterns IS NOT NULL THEN + RETURN QUERY EXECUTE + 'WITH selected AS ( + SELECT id, lock_for + FROM jobs + WHERE pattern = ANY($1) + AND run_after <= now() + AND (locked_until IS NULL OR locked_until < now()) + AND attempts < max_attempts + ORDER BY run_after + FOR UPDATE SKIP LOCKED + LIMIT $2 + ), + updated AS ( + UPDATE jobs j + SET + updated_at = now(), + locked_by = coalesce($3, ''administrator''), + locked_until = now() + s.lock_for * interval ''1 second'', + locked_at = now(), + attempts = j.attempts + 1 + FROM selected s + WHERE j.id = s.id + RETURNING j.* + ) + SELECT * FROM updated ORDER BY run_after' + USING p_patterns, p_max, p_locked_by; + ELSE + RETURN QUERY EXECUTE + 'WITH selected AS ( + SELECT id, lock_for + FROM jobs + WHERE run_after <= now() + AND (locked_until IS NULL OR locked_until < now()) + AND attempts < max_attempts + ORDER BY run_after + FOR UPDATE SKIP LOCKED + LIMIT $1 + ), + updated AS ( + UPDATE jobs j + SET + updated_at = now(), + locked_by = coalesce($2, ''administrator''), + locked_until = now() + s.lock_for * interval ''1 second'', + locked_at = now(), + attempts = j.attempts + 1 + FROM selected s + WHERE j.id = s.id + RETURNING j.* + ) + SELECT * FROM updated ORDER BY run_after' + USING p_max, p_locked_by; + END IF; +END; +$$; diff --git a/src/tools.spec.ts b/src/tools.spec.ts index 3c8382c..378eccf 100644 --- a/src/tools.spec.ts +++ b/src/tools.spec.ts @@ -26,15 +26,15 @@ describe("migrate", () => { const result = await pool.query( `SELECT name FROM _migrations ORDER BY name`, ); - expect(result.rows.length).toBe(12); + expect(result.rows.length).toBe(14); expect(result.rows[0]?.name).toMatch(/^001_/); - expect(result.rows[11]?.name).toMatch(/^012_/); + expect(result.rows[13]?.name).toMatch(/^014_/); }); test("is idempotent", async () => { await migrate(); const result = await pool.query(`SELECT count(*) FROM _migrations`); - expect(Number(result.rows[0]?.count)).toBe(12); + expect(Number(result.rows[0]?.count)).toBe(14); }); });