diff --git a/benchmarks/storage/providers.test.ts b/benchmarks/storage/providers.test.ts new file mode 100644 index 00000000..e0a79d91 --- /dev/null +++ b/benchmarks/storage/providers.test.ts @@ -0,0 +1,9 @@ +import assert from 'node:assert/strict'; +import { normalizeGcsPrivateKey } from './providers.js'; + +const pemBody = 'abc\\ndef'; +assert.equal(normalizeGcsPrivateKey(pemBody), 'abc\ndef'); +assert.equal(normalizeGcsPrivateKey(JSON.stringify(pemBody)), 'abc\ndef'); +assert.equal(normalizeGcsPrivateKey('abc\r\ndef'), 'abc\ndef'); + +console.log('Storage provider credential normalization checks passed'); diff --git a/benchmarks/storage/providers.ts b/benchmarks/storage/providers.ts index e4100431..f43017d7 100644 --- a/benchmarks/storage/providers.ts +++ b/benchmarks/storage/providers.ts @@ -10,6 +10,35 @@ import { archil } from '@storagesdk/adapters/archil'; import { neon } from '@storagesdk/adapters/neon'; import type { StorageProviderConfig } from './types.js'; +export function normalizeGcsPrivateKey(value: string): string { + let key = value.trim(); + if (key.startsWith('"') && key.endsWith('"')) { + try { + key = JSON.parse(key) as string; + } catch { + // Fall through and normalize the value as plain text. + } + } + return key + .replaceAll('\\\\r\\\\n', '\r\n') + .replaceAll('\\n', '\n') + .replaceAll('\\r\\n', '\r\n') + .replace(/\r\n/g, '\n'); +} + +export function isStorageProviderAvailable(provider: StorageProviderConfig): boolean { + if (provider.requiredEnvVars.some((name) => !process.env[name])) { + return false; + } + try { + provider.createStorage(); + return true; + } catch { + console.warn(`Skipping ${provider.name}: storage credentials could not be initialized`); + return false; + } +} + /** * Storage provider benchmark configurations. * @@ -153,8 +182,7 @@ export const storageProviders: StorageProviderConfig[] = [ projectId: process.env.GCS_PROJECT_ID!, credentials: { client_email: process.env.GCS_CLIENT_EMAIL!, - // Secrets store the key with literal "\n"; restore real newlines. - private_key: process.env.GCS_PRIVATE_KEY!.replace(/\\n/g, '\n'), + private_key: normalizeGcsPrivateKey(process.env.GCS_PRIVATE_KEY!), }, }), }), diff --git a/benchmarks/storage/report-storage-concurrency.ts b/benchmarks/storage/report-storage-concurrency.ts new file mode 100644 index 00000000..eb143563 --- /dev/null +++ b/benchmarks/storage/report-storage-concurrency.ts @@ -0,0 +1,114 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { + STORAGE_CONCURRENCY_RESULTS_PATH, + type StorageConcurrencyResults, +} from './storage-concurrency-results.js'; + +const root = path.resolve('.'); +const input = process.env.STORAGE_CONCURRENCY_RESULTS + ? path.resolve(process.env.STORAGE_CONCURRENCY_RESULTS) + : STORAGE_CONCURRENCY_RESULTS_PATH; +const data = JSON.parse(fs.readFileSync(input, 'utf8')) as StorageConcurrencyResults; + +function providerName(provider: string): string { + return { + 'aws-s3': 'AWS S3', + 'cloudflare-r2': 'Cloudflare R2', + 'vercel-blob': 'Vercel Blob', + 'azure-blob': 'Azure Blob', + }[provider] ?? provider; +} + +function number(value: number | null): string { + return value === null ? '—' : value.toFixed(1); +} + +function escapeXml(value: string): string { + return value.replace(/[<>&"']/g, (character) => ({ + '<': '<', + '>': '>', + '&': '&', + '"': '"', + "'": ''', + }[character]!)); +} + +const rows = data.results.flatMap((provider) => + provider.cells.map((cell) => ({ + provider: providerName(provider.provider), + ...cell, + })), +); + +const summaries = [...data.results] + .sort((a, b) => b.compositeScore - a.compositeScore) + .map((provider) => ({ + ...provider, + displayName: providerName(provider.provider), + peakThroughput: Math.max(...provider.cells.map((cell) => cell.throughputOpsPerSecond), 0), + })); + +const markdown = [ + '# Storage Concurrency Results', + '', + `Run: \`${data.runId}\` `, + `Generated: ${data.timestamp}`, + '', + '## Provider ranking', + '', + ...(summaries.length > 0 + ? [`Top provider: **${summaries[0].displayName}** (${summaries[0].compositeScore.toFixed(2)}/100)`, ''] + : ['No provider results were available.', '']), + '| Rank | Provider | Composite score | Success rate | Valid cells | Peak throughput (ops/s) |', + '|---:|---|---:|---:|---:|---:|', + ...summaries.map((provider, index) => + `| ${index + 1} | ${provider.displayName} | ${provider.compositeScore.toFixed(2)} | ${(provider.successRate * 100).toFixed(1)}% | ${(provider.validCellRate * 100).toFixed(1)}% | ${provider.peakThroughput.toFixed(1)} |`, + ), + '', + 'Scores combine throughput (45%), p50 latency (20%), p95 latency (20%), and p99 latency (15%) per cell, then apply success-rate penalties.', + '', + '## Cell details', + '', + '| Provider | Cell | Throughput (ops/s) | p50 (ms) | p95 (ms) | p99 (ms) | Success | Valid |', + '|---|---|---:|---:|---:|---:|---:|:---:|', + ...rows.map((row) => + `| ${row.provider} | ${row.phase} | ${row.throughputOpsPerSecond.toFixed(1)} | ${number(row.p50Ms)} | ${number(row.p95Ms)} | ${number(row.p99Ms)} | ${(row.successRate * 100).toFixed(1)}% | ${row.valid ? 'yes' : 'NO'} |`, + ), + '', + 'This is a closed-loop benchmark. Throughput is the output observed at the requested client concurrency.', +].join('\n'); + +fs.writeFileSync(path.join(root, 'storage-concurrency.md'), `${markdown}\n`); + +const width = 1200; +const rowHeight = 28; +const headerHeight = 70; +const chartHeight = Math.max(180, summaries.length * rowHeight + 90); +const barWidth = 700; +const chartRows = summaries.map((row, index) => { + const y = headerHeight + index * rowHeight; + const widthValue = Math.max(1, (row.compositeScore / 100) * barWidth); + return ` + ${escapeXml(row.displayName)} + + ${row.compositeScore.toFixed(2)}`; +}).join(''); + +const svg = ` + + +Storage Concurrency Composite Score +Run ${escapeXml(data.runId)} · higher is better +${chartRows} +`; + +fs.writeFileSync(path.join(root, 'storage-concurrency.svg'), svg); +console.log('Wrote storage-concurrency.md and storage-concurrency.svg'); diff --git a/benchmarks/storage/storage-concurrency-corpus.test.ts b/benchmarks/storage/storage-concurrency-corpus.test.ts new file mode 100644 index 00000000..c894ec3b --- /dev/null +++ b/benchmarks/storage/storage-concurrency-corpus.test.ts @@ -0,0 +1,28 @@ +import assert from 'node:assert/strict'; +import { requestKey } from './storage-concurrency-corpus.js'; + +function prefix(key: string): string { + return key.split('/')[2]!; +} + +const workerZeroKeys = Array.from( + { length: 32 }, + (_, opSeq) => requestKey(0, opSeq, 'SINGLE_PREFIX'), +); +const workerOneKeys = Array.from( + { length: 32 }, + (_, opSeq) => requestKey(1, opSeq, 'SINGLE_PREFIX'), +); + +assert.ok(workerZeroKeys.every((key) => prefix(key) === 'p00')); +assert.ok(workerOneKeys.every((key) => prefix(key) === 'p00')); +assert.notDeepEqual(workerZeroKeys, workerOneKeys); + +const singlePrefixKeys = new Set( + Array.from({ length: 128 }, (_, workerId) => + requestKey(workerId, 0, 'SINGLE_PREFIX'), + ), +); +assert.equal(singlePrefixKeys.size, 128); + +console.log('Storage concurrency corpus key distribution checks passed'); diff --git a/benchmarks/storage/storage-concurrency-corpus.ts b/benchmarks/storage/storage-concurrency-corpus.ts new file mode 100644 index 00000000..78a13b40 --- /dev/null +++ b/benchmarks/storage/storage-concurrency-corpus.ts @@ -0,0 +1,36 @@ +export const OBJECT_COUNT = 10_000; +export const PREFIX_COUNT = 64; +export const OBJECT_SIZE_BYTES = 1024; +export const KEY_STRIDE = 7919; +export const WORKER_PRIME = 104_729; +const SINGLE_PREFIX_OBJECT_COUNT = Math.ceil(OBJECT_COUNT / PREFIX_COUNT); + +export type KeyDistribution = 'SINGLE_PREFIX' | 'SPREAD_64'; + +export function corpusKey(index: number): string { + if (!Number.isInteger(index) || index < 0 || index >= OBJECT_COUNT) { + throw new Error(`Corpus index must be between 0 and ${OBJECT_COUNT - 1}`); + } + const prefix = index % PREFIX_COUNT; + return `bench/v1/p${prefix.toString().padStart(2, '0')}/obj${index.toString().padStart(6, '0')}`; +} + +export function requestKey( + workerId: number, + opSeq: number, + distribution: KeyDistribution, +): string { + const rawIndex = distribution === 'SINGLE_PREFIX' + ? (workerId * WORKER_PRIME + opSeq * KEY_STRIDE) % SINGLE_PREFIX_OBJECT_COUNT + : (workerId * WORKER_PRIME + opSeq * KEY_STRIDE) % OBJECT_COUNT; + const index = distribution === 'SINGLE_PREFIX' ? rawIndex * PREFIX_COUNT : rawIndex; + return corpusKey(index); +} + +/** Stable 1 KB body shared by every seeded object. */ +export function corpusBody(): Uint8Array { + return Uint8Array.from( + { length: OBJECT_SIZE_BYTES }, + (_, index) => (index * 31 + 17) % 256, + ); +} diff --git a/benchmarks/storage/storage-concurrency-results.ts b/benchmarks/storage/storage-concurrency-results.ts new file mode 100644 index 00000000..a5476a8f --- /dev/null +++ b/benchmarks/storage/storage-concurrency-results.ts @@ -0,0 +1,90 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import type { BenchmarkRunOutcome } from '@benchsdk/runner'; +import { scoreStorageConcurrencyProvider } from './storage-concurrency-scoring.js'; + +export const STORAGE_CONCURRENCY_RESULTS_DIR = path.resolve('results/storage-concurrency'); +export const STORAGE_CONCURRENCY_RESULTS_PATH = path.join( + STORAGE_CONCURRENCY_RESULTS_DIR, + 'latest.json', +); + +export interface StorageConcurrencyCellResult { + phase: string; + operations: number; + measuredOperations: number; + concurrency: number; + keyDistribution: 'SINGLE_PREFIX' | 'SPREAD_64'; + throughputOpsPerSecond: number; + p50Ms: number | null; + p95Ms: number | null; + p99Ms: number | null; + successRate: number; + throttleRate: number; + timeoutRate: number; + connectionErrorRate: number; + notFoundRate: number; + serverErrorRate: number; + clientErrorRate: number; + valid: boolean; + maxActiveRequests: number; +} + +export interface StorageConcurrencyProviderResult { + provider: string; + cells: StorageConcurrencyCellResult[]; + compositeScore: number; + successRate: number; + validCellRate: number; +} + +export interface StorageConcurrencyResults { + version: '1.0'; + timestamp: string; + runId: string; + environment: { + node: string; + platform: string; + arch: string; + }; + results: StorageConcurrencyProviderResult[]; +} + +function isCellResult(value: unknown): value is StorageConcurrencyCellResult { + return typeof value === 'object' && value !== null + && typeof (value as { phase?: unknown }).phase === 'string' + && typeof (value as { concurrency?: unknown }).concurrency === 'number' + && typeof (value as { throughputOpsPerSecond?: unknown }).throughputOpsPerSecond === 'number'; +} + +export function writeStorageConcurrencyResults( + outcome: BenchmarkRunOutcome, +): void { + const results: StorageConcurrencyProviderResult[] = outcome.participants.map((participant) => { + const result = { + provider: participant.participant, + cells: participant.records.flatMap((record) => { + const data = record.data; + return isCellResult(data) ? [data] : []; + }), + }; + return { ...result, ...scoreStorageConcurrencyProvider(result) }; + }); + + const output: StorageConcurrencyResults = { + version: '1.0', + timestamp: new Date().toISOString(), + runId: outcome.runId, + environment: { + node: process.version, + platform: os.platform(), + arch: os.arch(), + }, + results, + }; + + fs.mkdirSync(STORAGE_CONCURRENCY_RESULTS_DIR, { recursive: true }); + fs.writeFileSync(STORAGE_CONCURRENCY_RESULTS_PATH, `${JSON.stringify(output, null, 2)}\n`); + console.log(`Storage concurrency results written to ${STORAGE_CONCURRENCY_RESULTS_PATH}`); +} diff --git a/benchmarks/storage/storage-concurrency-scoring.ts b/benchmarks/storage/storage-concurrency-scoring.ts new file mode 100644 index 00000000..55b464b0 --- /dev/null +++ b/benchmarks/storage/storage-concurrency-scoring.ts @@ -0,0 +1,78 @@ +import type { + StorageConcurrencyCellResult, + StorageConcurrencyProviderResult, +} from './storage-concurrency-results.js'; + +export interface StorageConcurrencyScore { + compositeScore: number; + successRate: number; + validCellRate: number; +} + +/** + * Absolute scoring ceilings. Scores are stable when providers are added or + * removed, matching the repository's other benchmark score implementations. + */ +const THROUGHPUT_CEILING_OPS = 2_000; +const P50_CEILING_MS = 500; +const P95_CEILING_MS = 1_000; +const P99_CEILING_MS = 2_000; + +const WEIGHTS = { + throughput: 0.45, + p50: 0.20, + p95: 0.20, + p99: 0.15, +} as const; + +function scoreHigher(value: number, ceiling: number): number { + return Math.max(0, Math.min(100, (value / ceiling) * 100)); +} + +function scoreLower(value: number, ceiling: number): number { + return Math.max(0, Math.min(100, (1 - value / ceiling) * 100)); +} + +function scoreCell(cell: StorageConcurrencyCellResult): number { + const metrics: { weight: number; score: number | null }[] = [ + { + weight: WEIGHTS.throughput, + score: scoreHigher(cell.throughputOpsPerSecond, THROUGHPUT_CEILING_OPS), + }, + { + weight: WEIGHTS.p50, + score: cell.p50Ms === null ? null : scoreLower(cell.p50Ms, P50_CEILING_MS), + }, + { + weight: WEIGHTS.p95, + score: cell.p95Ms === null ? null : scoreLower(cell.p95Ms, P95_CEILING_MS), + }, + { + weight: WEIGHTS.p99, + score: cell.p99Ms === null ? null : scoreLower(cell.p99Ms, P99_CEILING_MS), + }, + ]; + const available = metrics.filter((metric) => metric.score !== null); + const weight = available.reduce((sum, metric) => sum + metric.weight, 0); + if (!cell.valid || weight === 0) return 0; + const score = available.reduce((sum, metric) => sum + metric.weight * metric.score!, 0) / weight; + return score; +} + +export function scoreStorageConcurrencyProvider( + result: Pick, +): StorageConcurrencyScore { + if (result.cells.length === 0) { + return { compositeScore: 0, successRate: 0, validCellRate: 0 }; + } + + const cellScore = result.cells.reduce((sum, cell) => sum + scoreCell(cell), 0) / result.cells.length; + const successRate = result.cells.reduce((sum, cell) => sum + cell.successRate, 0) / result.cells.length; + const validCellRate = result.cells.filter((cell) => cell.valid).length / result.cells.length; + + return { + compositeScore: Math.round(cellScore * successRate * 100) / 100, + successRate: Math.round(successRate * 10000) / 10000, + validCellRate: Math.round(validCellRate * 10000) / 10000, + }; +} diff --git a/benchmarks/storage/storage-concurrency-seed.ts b/benchmarks/storage/storage-concurrency-seed.ts new file mode 100644 index 00000000..5d013050 --- /dev/null +++ b/benchmarks/storage/storage-concurrency-seed.ts @@ -0,0 +1,144 @@ +/** + * Seed or verify the deterministic corpus used by storage-concurrency.bench.ts. + * + * Full seed (idempotent, checks every key): + * pnpm bench:storage-concurrency:seed -- --provider aws-s3 + * + * Cheap verification (checks 100 evenly distributed keys): + * pnpm bench:storage-concurrency:seed -- --verify --provider aws-s3 + */ +import '../src/env.js'; +import type { Storage } from '@storagesdk/core'; +import { isStorageProviderAvailable, storageProviders } from './providers.js'; +import type { StorageProviderConfig } from './types.js'; +import { corpusBody, corpusKey, OBJECT_COUNT, OBJECT_SIZE_BYTES } from './storage-concurrency-corpus.js'; +import { withTimeout } from '../src/util/timeout.js'; +import { formatError } from '../src/util/error.js'; + +const DEFAULT_CONCURRENCY = 16; +const REQUEST_TIMEOUT_MS = 30_000; +const VERIFY_SAMPLE_SIZE = 100; + +function argValue(flag: string): string | undefined { + const index = process.argv.indexOf(flag); + return index >= 0 ? process.argv[index + 1] : undefined; +} + +function hasFlag(flag: string): boolean { + return process.argv.includes(flag); +} + +function positiveInt(value: string | undefined, fallback: number, flag: string): number { + if (value === undefined) return fallback; + const parsed = Number(value); + if (!Number.isInteger(parsed) || parsed < 1) { + throw new Error(`${flag} must be an integer >= 1`); + } + return parsed; +} + +function selectedProviders(): StorageProviderConfig[] { + const requested = argValue('--provider') + ?.split(',') + .map((name) => name.trim()) + .filter(Boolean); + const candidates = requested + ? storageProviders.filter((provider) => requested.includes(provider.name)) + : storageProviders; + const unavailable = candidates.filter((provider) => + provider.requiredEnvVars.some((name) => !process.env[name]), + ); + const available = candidates.filter( + (provider) => !unavailable.includes(provider) && isStorageProviderAvailable(provider), + ); + for (const provider of unavailable) { + console.log(`Skipping ${provider.name}: missing required credentials`); + } + if (available.length === 0) { + throw new Error('No selected storage providers have complete credentials'); + } + return available; +} + +function verifyIndexes(): number[] { + return Array.from( + { length: VERIFY_SAMPLE_SIZE }, + (_, sample) => Math.floor((sample * OBJECT_COUNT) / VERIFY_SAMPLE_SIZE), + ); +} + +async function runProvider( + provider: StorageProviderConfig, + verifyOnly: boolean, + concurrency: number, +): Promise { + const storage = provider.createStorage(); + const body = corpusBody(); + const indexes = verifyOnly ? verifyIndexes() : Array.from({ length: OBJECT_COUNT }, (_, i) => i); + let present = 0; + let seeded = 0; + let invalid = 0; + let cursor = 0; + + async function worker(): Promise { + while (true) { + const position = cursor++; + if (position >= indexes.length) return; + const index = indexes[position]; + const key = corpusKey(index); + try { + const metadata = await withTimeout( + storage.head(key), + REQUEST_TIMEOUT_MS, + `HEAD timed out for ${key}`, + ); + if (metadata.size === OBJECT_SIZE_BYTES) { + present++; + continue; + } + invalid++; + } catch { + // A missing object is expected during a full seed. + } + + if (verifyOnly) continue; + await withTimeout( + storage.upload(key, body, { + contentType: 'application/octet-stream', + cacheControl: 'no-store', + }), + REQUEST_TIMEOUT_MS, + `PUT timed out for ${key}`, + ); + seeded++; + } + } + + await Promise.all(Array.from({ length: concurrency }, () => worker())); + const expected = indexes.length; + const mode = verifyOnly ? 'verified' : 'checked'; + console.log( + `[${provider.name}] ${mode} ${expected} objects: present=${present}, seeded=${seeded}, invalid=${invalid}`, + ); + if (verifyOnly && (present !== expected || invalid > 0)) { + throw new Error(`[${provider.name}] corpus verification failed`); + } +} + +const verifyOnly = hasFlag('--verify'); +const concurrency = positiveInt(argValue('--concurrency'), DEFAULT_CONCURRENCY, '--concurrency'); + +const failures: string[] = []; +for (const provider of selectedProviders()) { + try { + await runProvider(provider, verifyOnly, concurrency); + } catch (error) { + const message = `${provider.name}: ${formatError(error)}`; + failures.push(message); + console.error(`Skipping provider after seed failure: ${message}`); + } +} + +if (failures.length > 0) { + console.error(`Corpus seeding completed with ${failures.length} provider failure(s)`); +} diff --git a/benchmarks/storage/storage-concurrency.bench.ts b/benchmarks/storage/storage-concurrency.bench.ts new file mode 100644 index 00000000..9390d806 --- /dev/null +++ b/benchmarks/storage/storage-concurrency.bench.ts @@ -0,0 +1,228 @@ +/** + * Single-run storage concurrency benchmark. + * + * The runner creates one task per phase and one participant per provider. Each + * task owns its request concurrency, so the platform run contains a comparable + * cell for every provider without opening separate runs. + * + * pnpm bench:storage-concurrency + * pnpm bench:storage-concurrency -- --provider aws-s3 --concurrency 32 + * + * The optional runner --concurrency flag is intentionally not used for the + * workload. The internal pool below is the source of truth for request + * concurrency. + */ +import '../src/env.js'; +import { defineBenchmarkConfig, defineTask } from '@benchsdk/runner'; +import type { Storage } from '@storagesdk/core'; +import { withTimeout } from '../src/util/timeout.js'; +import { formatError } from '../src/util/error.js'; +import { isStorageProviderAvailable, storageProviders } from './providers.js'; +import type { StorageProviderConfig } from './types.js'; +import { requestKey } from './storage-concurrency-corpus.js'; +import type { KeyDistribution } from './storage-concurrency-corpus.js'; +import { writeStorageConcurrencyResults } from './storage-concurrency-results.js'; + +const OPERATIONS = 1_200; +const WARMUP_FRACTION = 0.05; +const REQUEST_TIMEOUT_MS = 30_000; + +interface Cell { + name: string; + concurrency: number; + keyDistribution: KeyDistribution; +} + +export const storageConcurrencyCells: Cell[] = [ + ...[1, 8, 32, 128].map((concurrency) => ({ + name: `c${concurrency}-single-prefix`, + concurrency, + keyDistribution: 'SINGLE_PREFIX' as const, + })), + ...[1, 8, 32, 128].map((concurrency) => ({ + name: `c${concurrency}-spread-64`, + concurrency, + keyDistribution: 'SPREAD_64' as const, + })), +]; + +const storageCache = new Map(); + +function getArgValue(flag: string): string | undefined { + const index = process.argv.indexOf(flag); + return index >= 0 ? process.argv[index + 1] : undefined; +} + +function getOperations(): number { + const value = getArgValue('--storage-operations') ?? process.env.STORAGE_CONCURRENCY_OPERATIONS; + if (value === undefined) return OPERATIONS; + const operations = Number(value); + if (!Number.isInteger(operations) || operations < 100) { + throw new Error('--storage-operations must be an integer >= 100'); + } + return operations; +} + +function percentile(values: number[], p: number): number | null { + if (values.length === 0) return null; + const sorted = [...values].sort((a, b) => a - b); + const index = Math.min(sorted.length - 1, Math.ceil(sorted.length * p) - 1); + return Number(sorted[index].toFixed(3)); +} + +function errorClass(error: unknown): string { + const text = formatError(error).toLowerCase(); + if (/timeout|timed out|abort/.test(text)) return 'TIMEOUT'; + if (/429|503|slowdown|throttl|rate.?limit|requestlimit/.test(text)) return 'THROTTLE'; + if (/reset|eof|socket|tls|connection/.test(text)) return 'CONN_ERROR'; + if (/notfound|not found/.test(text)) return 'NOT_FOUND'; + if (/unauthor|forbidden|accessdenied|invalidargument/.test(text)) return 'CLIENT'; + return 'SERVER'; +} + +interface RequestResult { + latencyMs: number; + bytes: number; + errorClass: string | null; + inWindow: boolean; + startMs: number; + endMs: number; +} + +interface CellResult { + operations: number; + measuredOperations: number; + concurrency: number; + keyDistribution: KeyDistribution; + throughputOpsPerSecond: number; + p50Ms: number | null; + p95Ms: number | null; + p99Ms: number | null; + successRate: number; + throttleRate: number; + timeoutRate: number; + connectionErrorRate: number; + notFoundRate: number; + serverErrorRate: number; + clientErrorRate: number; + valid: boolean; + status: 'COMPLETE'; + maxActiveRequests: number; +} + +async function runCell( + storage: Storage, + cell: Cell, + operations: number, +): Promise { + const results: RequestResult[] = []; + let nextOperation = 0; + let activeRequests = 0; + let maxActiveRequests = 0; + + async function worker(workerId: number): Promise { + let opSeq = 0; + while (true) { + const operation = nextOperation++; + if (operation >= operations) return; + + const startMs = performance.now(); + activeRequests++; + maxActiveRequests = Math.max(maxActiveRequests, activeRequests); + try { + const bytes = await withTimeout( + storage.download(requestKey(workerId, opSeq++, cell.keyDistribution), { as: 'bytes' }), + REQUEST_TIMEOUT_MS, + 'Storage GET timed out', + ); + const endMs = performance.now(); + results.push({ + latencyMs: endMs - startMs, + bytes: bytes.byteLength, + errorClass: null, + inWindow: operation >= Math.floor(operations * WARMUP_FRACTION), + startMs, + endMs, + }); + } catch (error) { + const endMs = performance.now(); + results.push({ + latencyMs: endMs - startMs, + bytes: 0, + errorClass: errorClass(error), + inWindow: operation >= Math.floor(operations * WARMUP_FRACTION), + startMs, + endMs, + }); + } finally { + activeRequests--; + } + } + } + + await Promise.all(Array.from({ length: cell.concurrency }, (_, workerId) => worker(workerId))); + + const measured = results.filter((result) => result.inWindow); + const latencies = measured.map((result) => result.latencyMs); + const successCount = measured.filter((result) => result.errorClass === null).length; + const throttleCount = measured.filter((result) => result.errorClass === 'THROTTLE').length; + const timeoutCount = measured.filter((result) => result.errorClass === 'TIMEOUT').length; + const connectionErrorCount = measured.filter((result) => result.errorClass === 'CONN_ERROR').length; + const notFoundCount = measured.filter((result) => result.errorClass === 'NOT_FOUND').length; + const serverErrorCount = measured.filter((result) => result.errorClass === 'SERVER').length; + const clientErrorCount = measured.filter((result) => result.errorClass === 'CLIENT').length; + const rate = (count: number): number => Number((count / measured.length).toFixed(5)); + const start = Math.min(...measured.map((result) => result.startMs)); + const end = Math.max(...measured.map((result) => result.endMs)); + const durationSeconds = Math.max((end - start) / 1000, Number.EPSILON); + + return { + operations, + measuredOperations: measured.length, + concurrency: cell.concurrency, + keyDistribution: cell.keyDistribution, + throughputOpsPerSecond: Number((measured.length / durationSeconds).toFixed(3)), + p50Ms: percentile(latencies, 0.5), + p95Ms: percentile(latencies, 0.95), + // 1,200 operations leaves 1,140 measured samples, meeting the p99 gate. + p99Ms: measured.length >= 1_000 ? percentile(latencies, 0.99) : null, + successRate: rate(successCount), + throttleRate: rate(throttleCount), + timeoutRate: rate(timeoutCount), + connectionErrorRate: rate(connectionErrorCount), + notFoundRate: rate(notFoundCount), + serverErrorRate: rate(serverErrorCount), + clientErrorRate: rate(clientErrorCount), + valid: successCount === measured.length, + status: 'COMPLETE', + maxActiveRequests, + }; +} + +export const config = defineBenchmarkConfig({ + benchmarkSlug: 'storage-concurrency', + benchmarkName: 'Storage Concurrency', + phases: storageConcurrencyCells.map((cell) => ({ name: cell.name, iterations: 1 })), + // The internal pool establishes storage-request concurrency. Keeping the + // runner at one task in flight prevents cells from overlapping. + concurrency: 1, + groupBy: 'participant', + participants: storageProviders.filter(isStorageProviderAvailable), + onComplete: writeStorageConcurrencyResults, +}); + +export const task = defineTask(async (ctx) => { + const cell = storageConcurrencyCells.find((candidate) => candidate.name === ctx.phase); + if (!cell) throw new Error(`Unknown storage concurrency cell: ${ctx.phase ?? '(missing phase)'}`); + + let storage = storageCache.get(ctx.participant.name); + if (!storage) { + storage = ctx.participant.createStorage(); + storageCache.set(ctx.participant.name, storage); + } + + const result = await ctx.step(`measure-${cell.name}`, () => + runCell(storage!, cell, getOperations()), + ); + return { data: { ...result } }; +}); diff --git a/benchmarks/storage/storage-concurrency.md b/benchmarks/storage/storage-concurrency.md new file mode 100644 index 00000000..e5f87163 --- /dev/null +++ b/benchmarks/storage/storage-concurrency.md @@ -0,0 +1,67 @@ +# Storage Concurrency Benchmark + +This benchmark creates one platform run containing every environment-available +storage provider and eight cells per provider: + +- concurrency: `1`, `8`, `32`, `128` +- key distributions: `SINGLE_PREFIX`, `SPREAD_64` + +Each cell is one runner task. The task owns an internal closed-loop pool with +the requested number of workers. A worker issues one GET, waits for it to +finish, and immediately issues the next GET until the cell operation budget is +complete. + +The runner's concurrency is fixed at `1` deliberately. This prevents cells +from overlapping; the internal pool is the source of truth for storage request +concurrency. + +## Run + +```bash +pnpm bench:storage-concurrency +``` + +Run selected providers: + +```bash +pnpm bench:storage-concurrency -- --provider aws-s3,cloudflare-r2 +``` + +Override the per-cell operation budget: + +```bash +pnpm bench:storage-concurrency -- --storage-operations 2000 +``` + +Seed the corpus for one provider: + +```bash +pnpm bench:storage-concurrency:seed -- --provider aws-s3 +``` + +Verify a 100-object sample: + +```bash +pnpm bench:storage-concurrency:seed -- --verify --provider aws-s3 +``` + +The corpus must be seeded at: + +```text +bench/v1/p00/obj000000 +... +bench/v1/p63/obj009999 +``` + +Each task reports throughput, latency percentiles, success/error rates by +class, a `valid` flag, and the observed maximum active request count. The first +5% of operations are warmup and excluded from the reported metrics. + +The workflow collects all provider cells into one `latest.json` artifact and +generates `storage-concurrency.md` plus `storage-concurrency.svg` for the +comparison. + +The composite score is 0–100 and is computed from each provider's cells: +throughput (45%), p50 latency (20%), p95 latency (20%), and p99 latency (15%). +Scores use fixed absolute ceilings and are multiplied by success rate, so +invalid or failing cells reduce the provider score. diff --git a/package.json b/package.json index 69e64213..004fdb56 100644 --- a/package.json +++ b/package.json @@ -43,6 +43,11 @@ "bench:browser-throughput:steel": "tsx packages/benchsdk-runner/dist/bin.js run benchmarks/browser/browser-throughput.bench.ts --provider steel", "bench:browser-throughput:tilion": "tsx packages/benchsdk-runner/dist/bin.js run benchmarks/browser/browser-throughput.bench.ts --provider tilion", "bench:storage": "tsx packages/benchsdk-runner/dist/bin.js run benchmarks/storage/storage.bench.ts", + "bench:storage-concurrency": "tsx packages/benchsdk-runner/dist/bin.js run benchmarks/storage/storage-concurrency.bench.ts", + "bench:storage-concurrency:seed": "tsx benchmarks/storage/storage-concurrency-seed.ts", + "test:storage-concurrency:corpus": "tsx benchmarks/storage/storage-concurrency-corpus.test.ts", + "test:storage:providers": "tsx benchmarks/storage/providers.test.ts", + "report:storage-concurrency": "tsx benchmarks/storage/report-storage-concurrency.ts", "bench:storage:s3": "tsx packages/benchsdk-runner/dist/bin.js run benchmarks/storage/storage.bench.ts --provider aws-s3", "bench:storage:r2": "tsx packages/benchsdk-runner/dist/bin.js run benchmarks/storage/storage.bench.ts --provider cloudflare-r2", "bench:storage:tigris": "tsx packages/benchsdk-runner/dist/bin.js run benchmarks/storage/storage.bench.ts --provider tigris",